Some small readability improvements in the MessageHandler code
In particular the `_processStreamMessage`-method is a bit cumbersome to read, given the way that the current streamController/streamSink is accessed, which we can improve with a couple of local variables.
This commit is contained in:
Jonas Jenwald
parent
c3ca78fdf8
commit
890a6c1108
@ -122,6 +122,7 @@ class MessageHandler {
|
||||
if (data.callbackId) {
|
||||
const cbSourceName = this.sourceName;
|
||||
const cbTargetName = data.sourceName;
|
||||
|
||||
new Promise(function (resolve) {
|
||||
resolve(action(data.data));
|
||||
}).then(
|
||||
@ -230,10 +231,10 @@ class MessageHandler {
|
||||
* @returns {ReadableStream} ReadableStream to read data in chunks.
|
||||
*/
|
||||
sendWithStream(actionName, data, queueingStrategy, transfers) {
|
||||
const streamId = this.streamId++;
|
||||
const sourceName = this.sourceName;
|
||||
const targetName = this.targetName;
|
||||
const comObj = this.comObj;
|
||||
const streamId = this.streamId++,
|
||||
sourceName = this.sourceName,
|
||||
targetName = this.targetName,
|
||||
comObj = this.comObj;
|
||||
|
||||
return new ReadableStream(
|
||||
{
|
||||
@ -300,12 +301,12 @@ class MessageHandler {
|
||||
* @private
|
||||
*/
|
||||
_createStreamSink(data) {
|
||||
const self = this;
|
||||
const action = this.actionHandler[data.action];
|
||||
const streamId = data.streamId;
|
||||
const sourceName = this.sourceName;
|
||||
const targetName = data.sourceName;
|
||||
const comObj = this.comObj;
|
||||
const streamId = data.streamId,
|
||||
sourceName = this.sourceName,
|
||||
targetName = data.sourceName,
|
||||
comObj = this.comObj;
|
||||
const self = this,
|
||||
action = this.actionHandler[data.action];
|
||||
|
||||
const streamSink = {
|
||||
enqueue(chunk, size = 1, transfers) {
|
||||
@ -373,6 +374,7 @@ class MessageHandler {
|
||||
streamSink.sinkCapability.resolve();
|
||||
streamSink.ready = streamSink.sinkCapability.promise;
|
||||
this.streamSinks[streamId] = streamSink;
|
||||
|
||||
new Promise(function (resolve) {
|
||||
resolve(action(data.data, streamSink));
|
||||
}).then(
|
||||
@ -401,33 +403,31 @@ class MessageHandler {
|
||||
* @private
|
||||
*/
|
||||
_processStreamMessage(data) {
|
||||
const streamId = data.streamId;
|
||||
const sourceName = this.sourceName;
|
||||
const targetName = data.sourceName;
|
||||
const comObj = this.comObj;
|
||||
const streamId = data.streamId,
|
||||
sourceName = this.sourceName,
|
||||
targetName = data.sourceName,
|
||||
comObj = this.comObj;
|
||||
const streamController = this.streamControllers[streamId],
|
||||
streamSink = this.streamSinks[streamId];
|
||||
|
||||
switch (data.stream) {
|
||||
case StreamKind.START_COMPLETE:
|
||||
if (data.success) {
|
||||
this.streamControllers[streamId].startCall.resolve();
|
||||
streamController.startCall.resolve();
|
||||
} else {
|
||||
this.streamControllers[streamId].startCall.reject(
|
||||
wrapReason(data.reason)
|
||||
);
|
||||
streamController.startCall.reject(wrapReason(data.reason));
|
||||
}
|
||||
break;
|
||||
case StreamKind.PULL_COMPLETE:
|
||||
if (data.success) {
|
||||
this.streamControllers[streamId].pullCall.resolve();
|
||||
streamController.pullCall.resolve();
|
||||
} else {
|
||||
this.streamControllers[streamId].pullCall.reject(
|
||||
wrapReason(data.reason)
|
||||
);
|
||||
streamController.pullCall.reject(wrapReason(data.reason));
|
||||
}
|
||||
break;
|
||||
case StreamKind.PULL:
|
||||
// Ignore any pull after close is called.
|
||||
if (!this.streamSinks[streamId]) {
|
||||
if (!streamSink) {
|
||||
comObj.postMessage({
|
||||
sourceName,
|
||||
targetName,
|
||||
@ -437,20 +437,16 @@ class MessageHandler {
|
||||
});
|
||||
break;
|
||||
}
|
||||
// Pull increases the desiredSize property of sink,
|
||||
// so when it changes from negative to positive,
|
||||
// set ready property as resolved promise.
|
||||
if (
|
||||
this.streamSinks[streamId].desiredSize <= 0 &&
|
||||
data.desiredSize > 0
|
||||
) {
|
||||
this.streamSinks[streamId].sinkCapability.resolve();
|
||||
// Pull increases the desiredSize property of sink, so when it changes
|
||||
// from negative to positive, set ready property as resolved promise.
|
||||
if (streamSink.desiredSize <= 0 && data.desiredSize > 0) {
|
||||
streamSink.sinkCapability.resolve();
|
||||
}
|
||||
// Reset desiredSize property of sink on every pull.
|
||||
this.streamSinks[streamId].desiredSize = data.desiredSize;
|
||||
const { onPull } = this.streamSinks[streamId];
|
||||
streamSink.desiredSize = data.desiredSize;
|
||||
|
||||
new Promise(function (resolve) {
|
||||
resolve(onPull && onPull());
|
||||
resolve(streamSink.onPull && streamSink.onPull());
|
||||
}).then(
|
||||
function () {
|
||||
comObj.postMessage({
|
||||
@ -473,54 +469,43 @@ class MessageHandler {
|
||||
);
|
||||
break;
|
||||
case StreamKind.ENQUEUE:
|
||||
assert(
|
||||
this.streamControllers[streamId],
|
||||
"enqueue should have stream controller"
|
||||
);
|
||||
if (this.streamControllers[streamId].isClosed) {
|
||||
assert(streamController, "enqueue should have stream controller");
|
||||
if (streamController.isClosed) {
|
||||
break;
|
||||
}
|
||||
this.streamControllers[streamId].controller.enqueue(data.chunk);
|
||||
streamController.controller.enqueue(data.chunk);
|
||||
break;
|
||||
case StreamKind.CLOSE:
|
||||
assert(
|
||||
this.streamControllers[streamId],
|
||||
"close should have stream controller"
|
||||
);
|
||||
if (this.streamControllers[streamId].isClosed) {
|
||||
assert(streamController, "close should have stream controller");
|
||||
if (streamController.isClosed) {
|
||||
break;
|
||||
}
|
||||
this.streamControllers[streamId].isClosed = true;
|
||||
this.streamControllers[streamId].controller.close();
|
||||
this._deleteStreamController(streamId);
|
||||
streamController.isClosed = true;
|
||||
streamController.controller.close();
|
||||
this._deleteStreamController(streamController, streamId);
|
||||
break;
|
||||
case StreamKind.ERROR:
|
||||
assert(
|
||||
this.streamControllers[streamId],
|
||||
"error should have stream controller"
|
||||
);
|
||||
this.streamControllers[streamId].controller.error(
|
||||
wrapReason(data.reason)
|
||||
);
|
||||
this._deleteStreamController(streamId);
|
||||
assert(streamController, "error should have stream controller");
|
||||
streamController.controller.error(wrapReason(data.reason));
|
||||
this._deleteStreamController(streamController, streamId);
|
||||
break;
|
||||
case StreamKind.CANCEL_COMPLETE:
|
||||
if (data.success) {
|
||||
this.streamControllers[streamId].cancelCall.resolve();
|
||||
streamController.cancelCall.resolve();
|
||||
} else {
|
||||
this.streamControllers[streamId].cancelCall.reject(
|
||||
wrapReason(data.reason)
|
||||
);
|
||||
streamController.cancelCall.reject(wrapReason(data.reason));
|
||||
}
|
||||
this._deleteStreamController(streamId);
|
||||
this._deleteStreamController(streamController, streamId);
|
||||
break;
|
||||
case StreamKind.CANCEL:
|
||||
if (!this.streamSinks[streamId]) {
|
||||
if (!streamSink) {
|
||||
break;
|
||||
}
|
||||
const { onCancel } = this.streamSinks[streamId];
|
||||
|
||||
new Promise(function (resolve) {
|
||||
resolve(onCancel && onCancel(wrapReason(data.reason)));
|
||||
resolve(
|
||||
streamSink.onCancel && streamSink.onCancel(wrapReason(data.reason))
|
||||
);
|
||||
}).then(
|
||||
function () {
|
||||
comObj.postMessage({
|
||||
@ -541,10 +526,8 @@ class MessageHandler {
|
||||
});
|
||||
}
|
||||
);
|
||||
this.streamSinks[streamId].sinkCapability.reject(
|
||||
wrapReason(data.reason)
|
||||
);
|
||||
this.streamSinks[streamId].isCancelled = true;
|
||||
streamSink.sinkCapability.reject(wrapReason(data.reason));
|
||||
streamSink.isCancelled = true;
|
||||
delete this.streamSinks[streamId];
|
||||
break;
|
||||
default:
|
||||
@ -555,18 +538,14 @@ class MessageHandler {
|
||||
/**
|
||||
* @private
|
||||
*/
|
||||
async _deleteStreamController(streamId) {
|
||||
async _deleteStreamController(streamController, streamId) {
|
||||
// Delete the `streamController` only when the start, pull, and cancel
|
||||
// capabilities have settled, to prevent `TypeError`s.
|
||||
await Promise.allSettled(
|
||||
[
|
||||
this.streamControllers[streamId].startCall,
|
||||
this.streamControllers[streamId].pullCall,
|
||||
this.streamControllers[streamId].cancelCall,
|
||||
].map(function (capability) {
|
||||
return capability && capability.promise;
|
||||
})
|
||||
);
|
||||
await Promise.allSettled([
|
||||
streamController.startCall && streamController.startCall.promise,
|
||||
streamController.pullCall && streamController.pullCall.promise,
|
||||
streamController.cancelCall && streamController.cancelCall.promise,
|
||||
]);
|
||||
delete this.streamControllers[streamId];
|
||||
}
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user