-
Notifications
You must be signed in to change notification settings - Fork 154
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
chore (client): use async event emitter in Satellite client (#696)
## Problem
The Satellite client uses an `EventEmitter` to emit events to which the
Satellite process is subscribed. However, the Satellite process
registers asynchronous event listeners but the `EventEmitter` calls them
synchronously. Hence, async listeners for an event may run concurrently
and a new event may be emitted while async listeners of the previous
event are still running.
An example in NodeJS:
```ts
const EventEmitter = require('node:events');
const emitter = new EventEmitter();
emitter.on('transaction', (tx) => {
console.log('listener 1 start: ' + tx);
return new Promise((resolve) => {
setTimeout(() => {
console.log('listener 1 end: ' + tx);
resolve()
}, 10)
})
})
emitter.emit('transaction', 'tx1');
emitter.emit('transaction', 'tx2');
/*
Prints:
listener 1 start: tx1
listener 1 start: tx2
listener 1 end: tx1
listener 1 end: tx2
*/
emitter.on('transaction', (tx) => {
console.log('listener 2 start: ' + tx);
})
emitter.emit('transaction');
/*
Prints:
listener 1 start: tx3
listener 2 start: tx3
listener 1 end: tx3
*/
```
The first example above shows interleaving of async listeners on an
event.
The second example shows interleaving of listeners for distinct events.
## Solution
Implemented an `AsyncEventEmitter` that awaits all listeners before
processing the next event. Modified the Satellite client to use the
`AsyncEventEmitter` instead of the regular `EventEmitter`. In addition,
the `AsyncEventEmitter` is fully typed, inspired by the typings of the
[typed-emitter package](https://www.npmjs.com/package/typed-emitter).
Note that, we start all listeners for an event in order **without
awaiting**. Hence, async listeners registered for an event may run
concurrently. However, we await all listeners before processing the next
event.
With the new `AsyncEventEmitter` we get this behavior:
```ts
const emitter = new AsyncEventEmitter();
emitter.on('transaction', (tx) => {
console.log('listener 1 start: ' + tx);
return new Promise((resolve) => {
setTimeout(() => {
console.log('listener 1 end: ' + tx);
resolve()
}, 10)
})
})
emitter.enqueueEmit('transaction', 'tx1');
emitter.enqueueEmit('transaction', 'tx2');
/*
Prints:
listener 1 start: tx1
listener 1 end: tx1
listener 1 start: tx2
listener 1 end: tx2
*/
emitter.on('transaction', (tx) => {
console.log('listener 2 start: ' + tx);
})
emitter.enqueueEmit('transaction');
/*
Prints:
listener 1 start: tx3
listener 2 start: tx3
listener 1 end: tx3
*/
```
The first example above shows that events are processed in order only
after all listeners have finished.
The second example shows that listeners for a given event are started in
order one after the other without awaiting.
Note: we renamed `emit` to `enqueueEmit` to avoid confusion with the
classical `EventEmitter`'s `emit` function which has different semantics
as it does not await all listeners before processing the next event.- Loading branch information
Showing
9 changed files
with
522 additions
and
58 deletions.
There are no files selected for viewing
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| --- | ||
| "electric-sql": patch | ||
| --- | ||
|
|
||
| Modify Satellite client to use async event emitter. |
This file contains 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
This file contains 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
This file contains 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
Oops, something went wrong.