You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
47 lines
1.3 KiB
JavaScript
47 lines
1.3 KiB
JavaScript
3 years ago
|
"use strict";
|
||
|
|
||
|
const pipe = require("@promistream/pipe");
|
||
|
const propagatePeek = require("@promistream/propagate-peek");
|
||
|
const propagateAbort = require("@promistream/propagate-abort");
|
||
|
const EndOfStream = require("@promistream/end-of-stream");
|
||
|
const unreachable = require("@joepie91/unreachable")("@promistream/dynamic");
|
||
|
|
||
|
module.exports = function singleValueProcessor(processingStream) {
|
||
|
// NOTE: This is *not* compliant with the Promistream spec! It can only be used with code that is specifically written to handle it.
|
||
|
|
||
|
let streamEnded = true;
|
||
|
|
||
|
let gateStream = {
|
||
|
_promistreamVersion: 0,
|
||
|
description: `single-value processor gate stream`,
|
||
|
peek: propagatePeek, // FIXME: Is this correct? Should this not be dependent on whether a peek and/or read has already been done for this iteration?
|
||
|
abort: propagateAbort,
|
||
|
read: function produceValue_singleValueGateStream(source) {
|
||
|
if (streamEnded === false) {
|
||
|
streamEnded = true;
|
||
|
return source.read();
|
||
|
} else {
|
||
|
throw new EndOfStream;
|
||
|
}
|
||
|
},
|
||
|
};
|
||
|
|
||
|
let wrappedPipeline = pipe([
|
||
|
gateStream,
|
||
|
processingStream
|
||
|
]);
|
||
|
|
||
|
return {
|
||
|
read: function (source) {
|
||
|
return wrappedPipeline.read(source);
|
||
|
},
|
||
|
reset: function () {
|
||
|
if (streamEnded === true) {
|
||
|
streamEnded = false;
|
||
|
} else {
|
||
|
throw unreachable("Tried to reset a non-ended stream");
|
||
|
}
|
||
|
}
|
||
|
};
|
||
|
};
|