Add ping pong heartbeat for signal connection

.changeset/fresh-terms-train.md

@@ -0,0 +1,5 @@
+---
+'livekit-client': patch
+---
+
+Add ping pong heartbeat for signal connection

src/api/SignalClient.ts

@@ -123,6 +123,14 @@ export class SignalClient {
 
  ws?: WebSocket;
 
+ private pingTimeout: ReturnType<typeof setTimeout> | undefined;
+
+ private pingTimeoutDuration: number | undefined;
+
+ private pingIntervalDuration: number | undefined;
+
+ private pingInterval: ReturnType<typeof setInterval> | undefined;
+
  constructor(useJSON: boolean = false) {
  this.isConnected = false;
  this.isReconnecting = false;
@@ -155,6 +163,8 @@ export class SignalClient {
 
  async reconnect(url: string, token: string): Promise<void> {
  this.isReconnecting = true;
+ // clear ping interval and restart it once reconnected
+ this.clearPingInterval();
  await this.connect(url, token, {
  reconnect: true,
  });
@@ -215,29 +225,41 @@ export class SignalClient {
  if (opts.reconnect) {
  // upon reconnection, there will not be additional handshake
  this.isConnected = true;
+ // restart ping interval as it's cleared for reconnection
+ this.startPingInterval();
  resolve();
  }
  };
 
  ws.onmessage = async (ev: MessageEvent) => {
  // not considered connected until JoinResponse is received
- let msg: SignalResponse;
+ let resp: SignalResponse;
  if (typeof ev.data === 'string') {
  const json = JSON.parse(ev.data);
- msg = SignalResponse.fromJSON(json);
+ resp = SignalResponse.fromJSON(json);
  } else if (ev.data instanceof ArrayBuffer) {
- msg = SignalResponse.decode(new Uint8Array(ev.data));
+ resp = SignalResponse.decode(new Uint8Array(ev.data));
  } else {
  log.error(`could not decode websocket message: ${typeof ev.data}`);
  return;
  }
 
  if (!this.isConnected) {
  // handle join message only
- if (msg.message?.$case === 'join') {
+ if (resp.message?.$case === 'join') {
  this.isConnected = true;
  abortSignal?.removeEventListener('abort', abortHandler);
- resolve(msg.message.join);
+ this.pingTimeoutDuration = resp.message.join.pingTimeout;
+ this.pingIntervalDuration = resp.message.join.pingInterval;
+
+ if (this.pingTimeoutDuration && this.pingTimeoutDuration > 0) {
+ log.debug('ping config', {
+ timeout: this.pingTimeoutDuration,
+ interval: this.pingIntervalDuration,
+ });
+ this.startPingInterval();
+ }
+ resolve(resp.message.join);
  } else {
  reject(new ConnectionError('did not receive join response'));
  }
@@ -247,7 +269,7 @@ export class SignalClient {
  if (this.signalLatency) {
  await sleep(this.signalLatency);
  }
- this.handleSignalResponse(msg);
+ this.handleSignalResponse(resp);
  };
 
  ws.onclose = (ev: CloseEvent) => {
@@ -268,6 +290,7 @@ export class SignalClient {
  if (this.ws) this.ws.onclose = null;
  this.ws?.close();
  this.ws = undefined;
+ this.clearPingInterval();
  }
 
  // initial offer after joining
@@ -364,6 +387,13 @@ export class SignalClient {
  });
  }
 
+ sendPing() {
+ this.sendRequest({
+ $case: 'ping',
+ ping: Date.now(),
+ });
+ }
+
  async sendLeave() {
  await this.sendRequest({
  $case: 'leave',
@@ -475,6 +505,8 @@ export class SignalClient {
  if (this.onLocalTrackUnpublished) {
  this.onLocalTrackUnpublished(msg.trackUnpublished);
  }
+ } else if (msg.$case === 'pong') {
+ this.resetPingTimeout();
  } else {
  log.debug('unsupported message', msg);
  }
@@ -493,6 +525,51 @@ export class SignalClient {
  private handleWSError(ev: Event) {
  log.error('websocket error', ev);
  }
+
+ private resetPingTimeout() {
+ this.clearPingTimeout();
+ if (!this.pingTimeoutDuration) {
+ log.warn('ping timeout duration not set');
+ return;
+ }
+ this.pingTimeout = setTimeout(() => {
+ log.warn(
+ `ping timeout triggered. last pong received at: ${new Date(
+ Date.now() - this.pingTimeoutDuration! * 1000,
+ ).toUTCString()}`,
+ );
+ if (this.onClose) {
+ this.onClose('ping timeout');
+ }
+ }, this.pingTimeoutDuration * 1000);
+ }
+
+ private clearPingTimeout() {
+ if (this.pingTimeout) {
+ clearTimeout(this.pingTimeout);
+ }
+ }
+
+ private startPingInterval() {
+ this.clearPingInterval();
+ this.resetPingTimeout();
+ if (!this.pingIntervalDuration) {
+ log.warn('ping interval duration not set');
+ return;
+ }
+ log.debug('start ping interval');
+ this.pingInterval = setInterval(() => {
+ this.sendPing();
+ }, this.pingIntervalDuration * 1000);
+ }
+
+ private clearPingInterval() {
+ log.debug('clearing ping interval');
+ this.clearPingTimeout();
+ if (this.pingInterval) {
+ clearInterval(this.pingInterval);
+ }
+ }
 }
 
 function fromProtoSessionDescription(sd: SessionDescription): RTCSessionDescriptionInit {

src/proto/livekit_models.ts

@@ -226,6 +226,7 @@ export enum DisconnectReason {
  PARTICIPANT_REMOVED = 4,
  ROOM_DELETED = 5,
  STATE_MISMATCH = 6,
+ JOIN_FAILURE = 7,
  UNRECOGNIZED = -1,
 }
 
@@ -252,6 +253,9 @@ export function disconnectReasonFromJSON(object: any): DisconnectReason {
  case 6:
  case 'STATE_MISMATCH':
  return DisconnectReason.STATE_MISMATCH;
+ case 7:
+ case 'JOIN_FAILURE':
+ return DisconnectReason.JOIN_FAILURE;
  case -1:
  case 'UNRECOGNIZED':
  default:
@@ -275,6 +279,8 @@ export function disconnectReasonToJSON(object: DisconnectReason): string {
  return 'ROOM_DELETED';
  case DisconnectReason.STATE_MISMATCH:
  return 'STATE_MISMATCH';
+ case DisconnectReason.JOIN_FAILURE:
+ return 'JOIN_FAILURE';
  case DisconnectReason.UNRECOGNIZED:
  default:
  return 'UNRECOGNIZED';
@@ -627,6 +633,11 @@ export interface RTPStats_GapHistogramEntry {
  value: number;
 }
 
+export interface TimedVersion {
+ unixMicro: number;
+ ticks: number;
+}
+
 function createBaseRoom(): Room {
  return {
  sid: '',
@@ -2654,6 +2665,64 @@ export const RTPStats_GapHistogramEntry = {
  },
 };
 
+function createBaseTimedVersion(): TimedVersion {
+ return { unixMicro: 0, ticks: 0 };
+}
+
+export const TimedVersion = {
+ encode(message: TimedVersion, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer {
+ if (message.unixMicro !== 0) {
+ writer.uint32(8).int64(message.unixMicro);
+ }
+ if (message.ticks !== 0) {
+ writer.uint32(16).int32(message.ticks);
+ }
+ return writer;
+ },
+
+ decode(input: _m0.Reader | Uint8Array, length?: number): TimedVersion {
+ const reader = input instanceof _m0.Reader ? input : new _m0.Reader(input);
+ let end = length === undefined ? reader.len : reader.pos + length;
+ const message = createBaseTimedVersion();
+ while (reader.pos < end) {
+ const tag = reader.uint32();
+ switch (tag >>> 3) {
+ case 1:
+ message.unixMicro = longToNumber(reader.int64() as Long);
+ break;
+ case 2:
+ message.ticks = reader.int32();
+ break;
+ default:
+ reader.skipType(tag & 7);
+ break;
+ }
+ }
+ return message;
+ },
+
+ fromJSON(object: any): TimedVersion {
+ return {
+ unixMicro: isSet(object.unixMicro) ? Number(object.unixMicro) : 0,
+ ticks: isSet(object.ticks) ? Number(object.ticks) : 0,
+ };
+ },
+
+ toJSON(message: TimedVersion): unknown {
+ const obj: any = {};
+ message.unixMicro !== undefined && (obj.unixMicro = Math.round(message.unixMicro));
+ message.ticks !== undefined && (obj.ticks = Math.round(message.ticks));
+ return obj;
+ },
+
+ fromPartial<I extends Exact<DeepPartial<TimedVersion>, I>>(object: I): TimedVersion {
+ const message = createBaseTimedVersion();
+ message.unixMicro = object.unixMicro ?? 0;
+ message.ticks = object.ticks ?? 0;
+ return message;
+ },
+};
+
 declare var self: any | undefined;
 declare var window: any | undefined;
 declare var global: any | undefined;