$ npm install corsa --save
writable.write3writable.write2writable.write1writable.endfor await of readableconsole.log'done'
output: > 3 > 2 > 1 > done
Corsa approaches this problem by making the channels sender await and suspend at buffer capacity. This helps to ensure the senders send rate is locked to the throughput allowed by a receiver.
Requires async/await and AsyncIteration support. Tested natively on Node v10.
A channel is a uni-directional pipe for which data can flow. The following code creates an
unbounded channel which allows for near infinite buffering of messages between
readable. The call to channel returns a
channel object, which we destructure into the readable and writable pairs.
The following creates a bounded channel which allows for sending
5 values before suspending (see bounded vs unbounded)
The following code creates an unbounded channel and sends the values
1, 2, 3 followed by a call to
end() to signal
EOF to a receiver.
Reader<T> is the receiving side of a channel and supports
for-await-of for general iteration.
writable.write1writable.write2writable.write3writable.endfor await of readable
bounded vs unbounded
By default all channels are
unbounded but it is possible to set a fixed buffering size when creating a
channel(). When setting a channel size, this will cause a writable to pause at
await when sending values. The
await at the writable will only occur once the channels buffer has filled with values. The writable will remained suspended until such time a receiver starts pulling values from the channel.
The following code demostrates this behavior with channel bound to a buffer of 5.
await writable.write1await writable.write2await writable.write3await writable.write4await writable.write5 // - at capacity, the readable will need to read something.await writable.write6 // suspend <-----+// | - readable.read() dequeues one element from the... // | stream which will cause the writable to resume.// |await readable.read // resume ------+
This library provides a simple channel
select function similar to multi channel select found in the Go programming language. It allows multiple
Reader<T> types to be combined into a singular stream.