Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 39 additions & 2 deletions packages/api/src/EmbeddedChatApi.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@ export default class EmbeddedChatApi {
auth: RocketChatAuth;
private _connectPromise: Promise<void> | null = null;
private _activeSubscriptions: { stop: () => void }[] = [];
// Incremented by close() so that a connect() which is still awaiting
// can detect that it has been superseded or cancelled.
private _connectionGeneration = 0;
private _authListener: ((user: any) => void) | null = null;

constructor(
Expand Down Expand Up @@ -231,6 +234,16 @@ export default class EmbeddedChatApi {
} catch (err: any) {
console.error(err instanceof Error ? err.message : err);
return err;
} finally {
// Tear down the realtime connection even if the logout request
// failed, otherwise the socket stays subscribed to the room and
// keeps receiving messages for a session that is no longer
// authenticated.
try {
await this.close();
Comment thread
rohanesor marked this conversation as resolved.
Resolved
} catch (err: any) {
console.error(err instanceof Error ? err.message : err);
}
}
}

Expand Down Expand Up @@ -273,16 +286,27 @@ export default class EmbeddedChatApi {
}

private async _doConnect(isChannelPrivate = false) {
// Take a snapshot of the generation: every close() invalidates it, so
// a connect() which is interrupted while awaiting (for example by a
// logout happening during joinRoom()) can stop before subscribing.
// Without this the subscriptions would be created after the socket
// was torn down, and the DDP client would replay them on reconnect.
this.close();
const generation = this._connectionGeneration;
const isStale = () => generation !== this._connectionGeneration;
try {
this.close(); // before connection, all previous subscriptions should be cancelled
await this.sdk.connection.connect();
if (isStale()) return;

const currentUser = (await this.auth.getCurrentUser()) as any;
if (isStale()) return;
const token = currentUser?.authToken || currentUser?.data?.authToken;
if (token) {
this._applyCredentials(currentUser);
await this.sdk.account.loginWithToken(token);
if (isStale()) return;
await this.joinRoom(isChannelPrivate);
if (isStale()) return;
}

// Subscribe to room messages
Expand Down Expand Up @@ -359,6 +383,12 @@ export default class EmbeddedChatApi {
);
this._activeSubscriptions.push(notifyUserUiSub);
}

// A close() may have happened while the subscriptions above were
// being registered, so drop them if this connect is no longer current.
if (isStale()) {
this._stopSubscriptions();
}
} catch (err) {
this.close();
}
Expand Down Expand Up @@ -551,9 +581,16 @@ export default class EmbeddedChatApi {
}
}

async close() {
private _stopSubscriptions() {
this._activeSubscriptions.forEach((sub) => sub.stop());
this._activeSubscriptions = [];
}

async close() {
// Invalidate any connect() which is currently in flight, so that it
// does not register subscriptions after this teardown.
this._connectionGeneration += 1;
this._stopSubscriptions();
this.sdk.connection.close();
}

Expand Down