feat(streams): Add stream multiplexer #221
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Also tidy up various asynchronous operations in base stream classes. These were causing promises returned from the stream multiplexer to fulfill erratically.
Get rid of the `makeChannels()` contrivance in favor of exposing a singular `addChannel()`. Also expose `start()` function in order to enable either "manually" draining channels or using the `drainAll()` method.
Also make `synchronize()` method of duplex streams optional.
Also fix issues in tests following rebase.
8 tasks
rekmarks
force-pushed
the
rekm/3-stream-multiplexer
branch
from
b0b0d32 to
bac8db2
Compare
sirtimid
approved these changes
Nov 7, 2024
Comment on lines
+172
to
+173
| // eslint-disable-next-line @typescript-eslint/await-thenable | ||
| await null; |
Contributor
There was a problem hiding this comment.
didn't we have a rule to allow this?
Member
Author
There was a problem hiding this comment.
It seems like it didn't get merged, and I think it's fine to disable it when necessary? If someone feels strongly about it (cc: @FUDCo) I'm down to just disable it in our lint config.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Ref: #218
Adds a new class
StreamMultiplexerto@ocap/streams. A multiplexer is not a stream itself, but rather a wrapper around a duplex stream. The multiplexer provides methods for creating "channels" over the underlying stream, which are themselves duplex streams and may have a different message type and validation logic.The multiplexer is constructed in an idle state, and must be explicitly "started" via the
start()ordrainAll()methods. All channels must be added before the multiplexer is started.Starting the multiplexer will synchronize the underlying duplex stream, if it is synchronizable. Therefore, in order to prevent message loss, callers should not synchronize the underlying duplex stream before passing it to the multiplexer. For the same reason, the multiplexer will throw if any channels are added after it has started. We could enable "dynamic" channel creation by synchronization the channels themselves, but only if there's a clear need.
See tests for example usage.