A package that makes it easier to work with subscribable objects #3301
Conversation
🦋 Changeset detectedLatest commit: cef0d13 The changes in this PR will be included in the next version bump. Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
steveluscher
force-pushed
the
10-02-a_package_that_makes_it_easier_to_work_with_subscribable_objects
branch
from
9c45d54
to
74846f3
Compare
steveluscher
force-pushed
the
10-02-a_package_that_makes_it_easier_to_work_with_subscribable_objects
branch
from
74846f3
to
7ea0591
Compare
| // It materializes listener signatures based on the events of the emitter | ||
| { | ||
| const eventEmitter = null as unknown as TypedEventEmitter<{ foo: CustomEvent<'bar'> }>; | ||
| const publisher = getDataPublisherFromEventEmitter(eventEmitter); | ||
| publisher.on('foo', data => { | ||
| data satisfies 'bar'; | ||
| }); | ||
| } |
There was a problem hiding this comment.
Could we add a typetest for non-custom events? I'm guessing, based on the type definition below, that null will be provided as the first argument.
Something like this:
const eventEmitter = null as unknown as TypedEventEmitter<{ foo: SomeEvent }>;
const publisher = getDataPublisherFromEventEmitter(eventEmitter);
publisher.on('foo', (...args) => {
args satisfies readonly [null]; // Is this right?
});There was a problem hiding this comment.
Oof. This change was done poorly and I didn't typecheck it with the rest of the stack.
steveluscher
force-pushed
the
10-02-a_package_that_makes_it_easier_to_work_with_subscribable_objects
branch
from
7ea0591
to
cef0d13
Compare
| it('flushes the queue before vending errors', async () => { | ||
| expect.assertions(4); | ||
| publish('data', 'consumed message'); | ||
| publish('data', 'queued message 1'); | ||
| publish('error', new Error('o no')); | ||
| await expect(iteratorA.next()).resolves.toStrictEqual({ | ||
| done: false, | ||
| value: 'queued message 1', | ||
| }); | ||
| await expect(iteratorB.next()).resolves.toStrictEqual({ | ||
| done: false, | ||
| value: 'queued message 1', | ||
| }); | ||
| await expect(iteratorA.next()).rejects.toThrow(new Error('o no')); | ||
| await expect(iteratorB.next()).rejects.toThrow(new Error('o no')); | ||
| }); |
There was a problem hiding this comment.
One key difference between this implementation and the last is that the last one would nuke the iterators immediately upon receiving an error. This might have caused code that was behind processing notifications to miss notifications that had already arrived on the client.
This implementation continues to vend any messages in the queue before vending the error. Errors are now a queue member like anything else.
| options, | ||
| ); | ||
| return { | ||
| async *[Symbol.asyncIterator]() { |
There was a problem hiding this comment.
This implementation still uses generators instead of the manual approach cancellation (via return()) in #3172. This means that the only way to interrupt this generator is to call the abort signal, which aborts the whole subscription – you can't just call .return() on the iterator. This is probably sufficient for most use cases today, but we should consider rolling in some of the ideas and tests from #3172 anyway.
| // It materializes listener signatures based on the events of the emitter | ||
| { | ||
| const eventEmitter = null as unknown as TypedEventEmitter<{ foo: CustomEvent<'bar'> }>; | ||
| const publisher = getDataPublisherFromEventEmitter(eventEmitter); | ||
| publisher.on('foo', data => { | ||
| data satisfies 'bar'; | ||
| }); | ||
| } |
Merge activity
|
steveluscher
deleted the
10-02-a_package_that_makes_it_easier_to_work_with_subscribable_objects
branch
Summary
At the root of the subscriptions infra is
WebSocketwhich conforms to theEventEmitterinterface. In this new package are utilties for making them easier to work with.getDataPublisherFromEventEmitterlets you convert an emitter into something with anon()method that vends unsubscribe functions. Unsubscribe functions are much easier to work with because you can pass them around as values to unsubscribe from an event without having to hold on to a reference to the original listener function.createAsyncIterableFromDataPublisherlets you create anAsyncIterablefrom aDataPublisher. This iterable handles cancellation, queueing when there is backpressure, and error propagation.Read the
READMEand the tests for more detail.