[Bugifx] Shubhra/ Add buffer to FFI events#508
Conversation
🦋 Changeset detectedLatest commit: 772570f The changes in this PR will be included in the next version bump. This PR includes changesets to release 6 packages
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 |
Shubhrakanti
changed the title
| { | ||
| "compilerOptions": { | ||
| "lib": ["es2015"], | ||
| "lib": ["es2015", "ES2021.WeakRef"], |
There was a problem hiding this comment.
We need this so that linter doesn't throw an error when using FinalizationRegistry
There was a problem hiding this comment.
@lukasIO is there a reason this is es2015 can we just update it to something later.
There was a problem hiding this comment.
Looks like we don't need FinalizationRegistry. But still would be good to upgrade this?
| { | ||
| "compilerOptions": { | ||
| "lib": ["es2015"], | ||
| "lib": ["es2015", "ES2021.WeakRef"], |
There was a problem hiding this comment.
@lukasIO is there a reason this is es2015 can we just update it to something later.
| } | ||
| } | ||
|
|
||
| subscribe(): ReadableStream<T> { |
There was a problem hiding this comment.
I chose this to be the equivalent of the asyncio's Queue we use in python. The main feature we care about is asynchronous reads and waiting for events while the stream is empty.
| remoteParticipants: Map<string, RemoteParticipant> = new Map(); | ||
| localParticipant?: LocalParticipant; | ||
|
|
||
| private static cleanupRegistry = new FinalizationRegistry((cleanup: () => void) => { |
There was a problem hiding this comment.
We use FinalizationRegistry (see docs) to act as an equivalent of python's __del__ method
There was a problem hiding this comment.
There's also Symbol.dispose as defined in a ECMA proposal https://github.com/tc39/proposal-explicit-resource-management but I wasn't sure if this was merged in yet.
| } catch (error) { | ||
| log.debug(error, 'Listen task ended'); | ||
| } finally { | ||
| this.listenTaskPromise = undefined; |
There was a problem hiding this comment.
is it better to type it as | null?
|
@davidzhao this should fix the issues folks were having with inbound sip calls in agents js |
| } catch (error: unknown) { | ||
| log.error(error, 'Error enqueuing item to stream'); | ||
| toRemove.add(controller); | ||
| } | ||
| } | ||
|
|
||
| for (const controller of toRemove) { | ||
| this.subscribers.delete(controller); | ||
| } |
There was a problem hiding this comment.
is this overkill? Just want to make sure there are no memory leaks.
| /** | ||
| * @throws "TypeError: Invalid state: ReadableStream is locked" | ||
| * if the stream is locked, make sure the stream is not being read from | ||
| * before calling this method. | ||
| */ | ||
| unsubscribe(stream: ReadableStream<T>): void { | ||
| stream.cancel(); | ||
| } | ||
| } |
There was a problem hiding this comment.
I don't love this, but it works for now.
| constructor() { | ||
| super(); | ||
| // Register a finalizer to disconnect the room when it's garbage collected | ||
| Room.cleanupRegistry.register(this, () => { |
There was a problem hiding this comment.
I don't think we need this. If the FfiHandle is dropped, the Rust side will already close the room
There was a problem hiding this comment.
Good to know. I saw in the python side so that's why I implemented it - but had a felling we didn't need this.
| }); | ||
|
|
||
| // subscribe before connecting so we don't miss any events | ||
| this.ffiQueue = FfiClient.instance.queue.subscribe(); |
There was a problem hiding this comment.
On JS it may be simpler, just hooking into the FfiEvent using on. We could just append to a list?
There was a problem hiding this comment.
Maybe it's fine, but IIRC that was the only difference. JS using events, but python using queues
There was a problem hiding this comment.
On JS it may be simpler, just hooking into the FfiEvent using on that's exactly what this is doing.
We grab the events and append them to a queue. The only reason it's a stream is so reads are async and we don't block the main event loop.
There was a problem hiding this comment.
Do you mean get rid of FfiClient.instance.queue all together. Replace this with
connect(){
// wire up the event listener first
FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onFfiEvent)
// send the connection request
const res = FfiClient.instance.request<ConnectResponse>({ ...... })
}
//
onEvent(event): {
this.buffer.push(event)
// only start processing the events once we've connected.
if (connected) {
this.proccessEvent(event)
...There was a problem hiding this comment.
As I understand it the order of operations are.
- Create a buffer for all the events coming in
- Connect to the room
- Immediately populate the buffer so we don't drop events
- Start processing events
Right now step 1 is done with this.ffiQueue = FfiClient.instance.queue.subscribe(); if we replace it with FfiClient.instance.on(FfiClientEvent.FfiEvent, this.onFfiEvent) we still need to buffer the events to process until we receive the ConnectResponse?
|
Discussed offline with @theomonnom - we're just going to buffer the events that come in before the connect in |
2 participants
This ports over a fix that was done in the python side, but never implemented in the node sdk (PR).
User reported that the
Roomwould throw an error if there was already a participant inside and they tried to connect. This is because we were processing theparticipant_connectedevent before thetrackSubscribedevent.Before fix:
room.connecttrackSubscribed← Tries to find participant that was never created!After fix:
room.connectparticipants_updatedparticipant_connected← Creates the participanttrack_publishedtrack_subscribed← Works because participant exists