Skip to content

Refactor dispatch handling #425

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 8 commits into from
Oct 4, 2017
Merged

Refactor dispatch handling #425

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
f95d59a
Refactor promise and object stream dispatches
wooldridge Sep 26, 2017
3331ae2
Don't bother pausing response if it has ended
wooldridge Sep 29, 2017
cc64415
Remove drain event in responder's chunked stream dispatch
wooldridge Sep 29, 2017
0336c94
Update drainListener for multipart object stream dispatch
wooldridge Oct 2, 2017
7e5b0a9
Remove unused QueueReader code
wooldridge Oct 2, 2017
a8c493b
Check if concatenating before drain event concats
wooldridge Oct 2, 2017
27ae556
Comment cleanup
wooldridge Oct 2, 2017
83e2c59
Ignore documentation of internal methods
wooldridge Oct 4, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 40 additions & 37 deletions lib/operation.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ function Operation(name, client, options, requestType, responseType) {
this.outputStreamMode = null;
this.outputStream = null;
this.streamModes = this.STREAM_MODES_CHUNKED_OBJECT;
this.nextMetadataBuffer = null;
}
Operation.prototype.STREAM_MODES_CHUNKED_OBJECT = {chunked: true, object: true};

Expand Down Expand Up @@ -90,17 +91,25 @@ Operation.prototype.collectBodyObject = function collectBodyObject(data) {

return bodyObject;
};
Operation.prototype.queueDocument = function queueDocument(
data, rawHeaderQueue, metadataBuffer, objectQueue
/**
* Make an object out of multipart part data
* @param {Buffer} data - Multipart part data
* @param {FifoQueue} rawHeaderQueue - Queue of headers for part data
* @return {Object} - The object
*
* @ignore
*/
Operation.prototype.makeObject = function makeObject(
data, rawHeaderQueue
) {
var operation = this;

var outputTransform = operation.outputTransform;
var partObject = null;

// Get corresponding multipart header for part
var partRawHeaders = rawHeaderQueue.pollFirst();

var partHeaders = parsePartHeaders(partRawHeaders);

var partUri = partHeaders.uri;

var isInline = (partUri == null);
Expand All @@ -110,66 +119,60 @@ Operation.prototype.queueDocument = function queueDocument(
partHeaders.category !== 'content'
);

// Convert buffer data to object
var partData = mlutil.unmarshal(partHeaders.format, data);

var nextMetadataBuffer = null;
var partObject = null;
// Inline case
if (isInline) {
if (metadataBuffer !== null) {
operation.queueMetadata(metadataBuffer, objectQueue);
}
operation.logger.debug('parsed inline');
// Resource execs and server-side evals
if (operation.inlineAsDocument) {
partHeaders.content = partData;
partObject = partHeaders;
} else {
partObject = partData;
}
} else if (isMetadata) {
if (metadataBuffer !== null) {
operation.queueMetadata(metadataBuffer, objectQueue);
// Search summaries
else {
partObject = partData;
}
}
// Metadata case
else if (isMetadata) {
operation.logger.debug('parsed metadata for %s', partUri);
nextMetadataBuffer = [partHeaders, partData];
} else {
if (this.nextMetadataBuffer !== null) {
var metadataHeaders = this.nextMetadataBuffer[0];
mlutil.copyProperties(this.nextMetadataBuffer[1], metadataHeaders);
partObject = metadataHeaders;
}
this.nextMetadataBuffer = [partHeaders, partData];
}
// Content case
else {
operation.logger.debug('parsed content for %s', partUri);
if (metadataBuffer !== null) {
if (metadataBuffer[0].uri === partUri) {
operation.logger.debug('copying metadata for %s', partUri);
mlutil.copyProperties(metadataBuffer[1], partHeaders);
} else {
operation.queueMetadata(metadataBuffer, objectQueue);
}
// If metadata exists, copy to object
if (this.nextMetadataBuffer !== null) {
operation.logger.debug('copying metadata for %s', partUri);
mlutil.copyProperties(this.nextMetadataBuffer[1], partHeaders);
this.nextMetadataBuffer = null;
}
partHeaders.content = partData;
partObject = partHeaders;
}

if (partObject !== null) {
// Subdata processing (poor man's XPath)
var subdata = operation.subdata;
if (Array.isArray(subdata)) {
partObject = projectData(partObject, subdata, 0);
}

// Transform
if (outputTransform != null) {
partObject = outputTransform.call(operation, partRawHeaders, partObject);
}

if (partObject !== void 0) {
objectQueue.addLast(partObject);
} else {
if (partObject === void 0) {
operation.logger.debug('skipped undefined output from transform');
}
}

return nextMetadataBuffer;
};
Operation.prototype.queueMetadata = function queueMetadata(
metadataBuffer, objectQueue
) {
var metadataHeaders = metadataBuffer[0];
mlutil.copyProperties(metadataBuffer[1], metadataHeaders);
objectQueue.addLast(metadataHeaders);
return partObject;
};
Operation.prototype.dispatchError = function dispatchError(error) {
var operation = this;
Expand Down
Loading