feat: use the shared sfu call object in react native

.yarn/patches/react-native-webrtc-npm-106.0.0-beta.6-795b82ce91.patch

@@ -0,0 +1,12 @@
+diff --git a/src/RTCPeerConnection.ts b/src/RTCPeerConnection.ts
+index 20de0069ce6137c1a10f65c2164084cf124c7ffe..46d89f399f65af135789b23ecd51258ce9df94f7 100644
+--- a/src/RTCPeerConnection.ts
++++ b/src/RTCPeerConnection.ts
+@@ -497,6 +497,7 @@ export default class RTCPeerConnection extends defineCustomEventTarget(...PEER_C
+
+             if (oldTransceiver) {
+                 transceiver = oldTransceiver;
++                transceiver._receiver._track =  new MediaStreamTrack(transceiver._receiver._track);
+                 track = transceiver._receiver._track;
+                 transceiver._mid = ev.transceiver.mid;
+                 transceiver._currentDirection = ev.transceiver.currentDirection;

package.json

@@ -44,6 +44,7 @@
   },
   "resolutions": {
     "react-native-pure-jwt@^3.0.1": "patch:react-native-pure-jwt@npm%3A3.0.1#./.yarn/patches/react-native-pure-jwt-npm-3.0.1-3e5e2c9059.patch",
-    "react-native-callkeep@^4.3.3": "patch:react-native-callkeep@npm%3A4.3.3#./.yarn/patches/react-native-callkeep-npm-4.3.3-71db80f4ef.patch"
+    "react-native-callkeep@^4.3.3": "patch:react-native-callkeep@npm%3A4.3.3#./.yarn/patches/react-native-callkeep-npm-4.3.3-71db80f4ef.patch",
+    "react-native-webrtc@106.0.0-beta.6": "patch:react-native-webrtc@npm%3A106.0.0-beta.6#./.yarn/patches/react-native-webrtc-npm-106.0.0-beta.6-795b82ce91.patch"
   }
 }

packages/client/src/StreamVideoClient.test.ts

@@ -17,7 +17,9 @@ describe('StreamVideoClient', () => {
         createSocketConnection: vi.fn(),
       };
     });
-    client = new StreamVideoClient('123', { token: 'abc' });
+    client = new StreamVideoClient('123', {
+      token: 'abc',
+    });
   });
 
   it('should connect', async () => {

packages/client/src/StreamVideoClient.ts

@@ -128,22 +128,6 @@ export class StreamVideoClient {
     return callEnvelope;
   };
 
-  // TODO: remove this method (it's only used in react-native for now until the sfu-Call object is merged)
-  joinCallRaw = async (data: JoinCallRequest, sessionId?: string) => {
-    const { response } = await this.client.joinCall({
-      ...data,
-      // FIXME: OL this needs to come from somewhere
-      datacenterId: 'amsterdam',
-    });
-    if (response.call && response.call.call && response.edges) {
-      const edge = await this.getCallEdgeServer(
-        response.call.call,
-        response.edges,
-      );
-      return { response, edge };
-    }
-  };
-
   joinCall = async (data: JoinCallRequest, sessionId?: string) => {
     const { response } = await this.client.joinCall({
       ...data,

packages/client/src/rpc/latency.ts

@@ -17,13 +17,16 @@ export const measureResourceLoadLatencyTo = async (
       .fill(undefined)
       .map(async () => {
         const start = Date.now();
+        const src = new URL(endpoint);
+        src.searchParams.set('rand', `react_${Math.random() * 10000000}`);
+        const controller = new AbortController();
+        const timeoutId = setTimeout(() => controller.abort(), 3000); // 3 second timeout:
         try {
-          const src = new URL(endpoint);
-          src.searchParams.set('rand', `react_${Math.random() * 10000000}`);
-          await fetch(src.toString()).then((response) => response.blob());
+          await fetch(src.toString(), { signal: controller.signal });
         } catch (e) {
-          console.warn(`failed to measure latency to ${endpoint}`, e);
+          console.warn(`failed to measure latency to ${src}`, e);
         }
+        clearTimeout(timeoutId); // clear timeout incase fetch completes before timeout
         const latency = Date.now() - start;
         measurements.push(toSeconds(latency));
       }),

packages/client/src/rtc/Call.ts

@@ -1,4 +1,4 @@
-import { createSubscriber } from './subscriber';
+import { addSubscriberListeners } from './subscriber';
 import {
   defaultVideoLayers,
   findOptimalVideoLayers,
@@ -11,7 +11,7 @@ import {
   getReceiverCodecs,
   getSenderCodecs,
 } from './codecs';
-import { createPublisher } from './publisher';
+import { addPublisherListeners } from './publisher';
 import { CallState, VideoDimension } from '../gen/video/sfu/models/models';
 import { registerEventHandlers } from './callEventHandlers';
 import { SfuRequest } from '../gen/video/sfu/event/events';
@@ -24,6 +24,8 @@ export type CallOptions = {
   connectionConfig: RTCConfiguration | undefined;
 };
 
+type MediaDeviceKind = 'audioinput' | 'audiooutput' | 'videoinput';
+
 export class Call {
   /**@deprecated use store for this data */
   currentUserId: string;
@@ -46,19 +48,19 @@ export class Call {
       stateStore.connectedUserSubject,
     )!.name;
     const { dispatcher, iceTrickleBuffer } = this.client;
-    this.subscriber = createSubscriber({
+    this.subscriber = new RTCPeerConnection(this.options.connectionConfig);
+    addSubscriberListeners({
       rpcClient: this.client,
-
       // FIXME: don't do this
       dispatcher: dispatcher,
-      connectionConfig: this.options.connectionConfig,
+      subscriber: this.subscriber,
       onTrack: this.handleOnTrack,
       candidates: iceTrickleBuffer.subscriberCandidates,
     });
-
-    this.publisher = createPublisher({
+    this.publisher = new RTCPeerConnection(this.options.connectionConfig);
+    addPublisherListeners({
       rpcClient: this.client,
-      connectionConfig: this.options.connectionConfig,
+      publisher: this.publisher,
       candidates: iceTrickleBuffer.publisherCandidates,
     });
 
@@ -100,6 +102,83 @@ export class Call {
     this.stateStore.activeCallSubject.next(undefined);
   };
 
+  // FIXME: remove this, it is temporary for RN implementation
+  joinWithCombinedStream = async (stream?: MediaStream) => {
+    await this.client.signalReady;
+
+    if (this.joinResponseReady) {
+      throw new Error(`Illegal State: Already joined.`);
+    }
+    this.joinResponseReady = new Promise<CallState | undefined>(
+      async (resolve) => {
+        const [audioEncode, audioDecode, videoEncode, videoDecode] =
+          await Promise.all([
+            getSenderCodecs('audio'),
+            getReceiverCodecs('audio', this.subscriber),
+            getSenderCodecs('video'),
+            getReceiverCodecs('video', this.subscriber),
+          ]);
+
+        this.videoLayers = stream
+          ? await findOptimalVideoLayers(stream)
+          : defaultVideoLayers;
+
+        this.client.dispatcher.on('joinResponse', (event) => {
+          if (event.eventPayload.oneofKind !== 'joinResponse') return;
+          const callState = event.eventPayload.joinResponse.callState;
+          this.stateStore.activeCallSubject.next(this);
+          this.participants.push(...(callState?.participants || []));
+          const ownParticipant = this.localParticipant;
+          if (ownParticipant) {
+            ownParticipant.isLoggedInUser = true;
+            ownParticipant.audioTrack = stream;
+            ownParticipant.videoTrack = stream;
+          }
+          this.stateStore.activeCallParticipantsSubject.next([
+            ...this.participants,
+          ]);
+          this.client.keepAlive();
+          resolve(callState); // expose call state
+        });
+
+        this.client.send(
+          SfuRequest.create({
+            requestPayload: {
+              oneofKind: 'joinRequest',
+              joinRequest: {
+                sessionId: this.client.sessionId,
+                token: this.client.token,
+                publish: true,
+                // FIXME OL: encode parameters and video layers
+                // should be announced when initiating "publish" operation
+                codecSettings: {
+                  audio: {
+                    encodes: audioEncode,
+                    decodes: audioDecode,
+                  },
+                  video: {
+                    encodes: videoEncode,
+                    decodes: videoDecode,
+                  },
+                  layers: this.videoLayers.map((layer) => ({
+                    rid: layer.rid!,
+                    bitrate: layer.maxBitrate!,
+                    videoDimension: {
+                      width: layer.width,
+                      height: layer.height,
+                    },
+                  })),
+                },
+              },
+            },
+          }),
+        );
+      },
+    );
+
+    return this.joinResponseReady;
+  };
+
   join = async (videoStream?: MediaStream, audioStream?: MediaStream) => {
     await this.client.signalReady;
 
@@ -235,6 +314,65 @@ export class Call {
     }
   };
 
+  // FIXME: remove this, it is temporary for RN implementation
+  publishWithCombinedStream = async (stream?: MediaStream) => {
+    if (!this.joinResponseReady) {
+      throw new Error(
+        `Illegal State: Can't publish. Please join the call first`,
+      );
+    }
+
+    // wait until we get a JoinResponse from the SFU, otherwise we risk
+    // breaking the ICETrickle flow.
+    await this.joinResponseReady;
+
+    if (stream) {
+      const videoEncodings: RTCRtpEncodingParameters[] =
+        this.videoLayers && this.videoLayers.length > 0
+          ? this.videoLayers
+          : defaultVideoPublishEncodings;
+
+      const [videoTrack] = stream.getVideoTracks();
+      if (videoTrack) {
+        const videoTransceiver = this.publisher?.addTransceiver(videoTrack, {
+          direction: 'sendonly',
+          streams: [stream],
+          sendEncodings: videoEncodings,
+        });
+
+        const codecPreferences = getPreferredCodecs('video', 'vp8');
+        // @ts-ignore
+        if ('setCodecPreferences' in videoTransceiver && codecPreferences) {
+          console.log(`set codec preferences`, codecPreferences);
+          videoTransceiver.setCodecPreferences(codecPreferences);
+        }
+      }
+
+      if (this.localParticipant) {
+        this.localParticipant.videoTrack = stream;
+        this.stateStore.activeCallParticipantsSubject.next([
+          ...this.participants,
+        ]);
+      }
+    }
+
+    if (stream) {
+      const [audioTrack] = stream.getAudioTracks();
+      if (audioTrack) {
+        this.publisher?.addTransceiver(audioTrack, {
+          direction: 'sendonly',
+        });
+      }
+
+      if (this.localParticipant) {
+        this.localParticipant.audioTrack = stream;
+        this.stateStore.activeCallParticipantsSubject.next([
+          ...this.participants,
+        ]);
+      }
+    }
+  };
+
   /**
    * Update track subscription configuration for one or more participants. You have to create a subscription for each participant you want to receive any kind of track.
    * @param changes
@@ -372,6 +510,7 @@ export class Call {
 
     const params = await videoSender.getParameters();
     let changed = false;
+    // @ts-ignore
     params.encodings.forEach((enc) => {
       console.log(enc.rid, enc.active);
       // flip 'active' flag only when necessary
@@ -382,6 +521,7 @@ export class Call {
       }
     });
     if (changed) {
+      // @ts-ignore
       if (params.encodings.length === 0) {
         console.warn('No suitable video encoding quality found');
       }

packages/client/src/rtc/codecs.ts

@@ -5,19 +5,19 @@ export const defaultVideoPublishEncodings: RTCRtpEncodingParameters[] = [
   {
     rid: 'f',
     active: true,
-    maxBitrate: 1_280_000,
+    maxBitrate: 1280000,
   },
   {
     rid: 'h',
     active: true,
     scaleResolutionDownBy: 2.0,
-    maxBitrate: 768_000,
+    maxBitrate: 768000,
   },
   {
     rid: 'q',
     active: true,
     scaleResolutionDownBy: 4.0,
-    maxBitrate: 384_000,
+    maxBitrate: 384000,
   },
 ];
 

packages/client/src/rtc/helpers/iceCandidate.ts

@@ -0,0 +1,16 @@
+import { ICETrickle } from '../../gen/video/sfu/models/models';
+
+export function getIceCandidate(
+  candidate: RTCIceCandidate,
+): ICETrickle['iceCandidate'] {
+  if (!candidate.usernameFragment) {
+    // react-native-webrtc doesn't include usernameFragment in the candidate
+    const splittedCandidate = candidate.candidate.split(' ');
+    const ufragIndex =
+      splittedCandidate.findIndex((s: string) => s === 'ufrag') + 1;
+    const usernameFragment = splittedCandidate[ufragIndex];
+    return JSON.stringify({ ...candidate, usernameFragment });
+  } else {
+    return JSON.stringify(candidate.toJSON());
+  }
+}

packages/client/src/rtc/publisher.ts

@@ -1,19 +1,21 @@
 import { StreamSfuClient } from '../StreamSfuClient';
 import { ICETrickle, PeerType } from '../gen/video/sfu/models/models';
 import { ReplaySubject } from 'rxjs';
+import { getIceCandidate } from './helpers/iceCandidate';
 
-export type PublisherOpts = {
+export type PublisherOpts<RTCPeerConnectionType extends RTCPeerConnection> = {
   rpcClient: StreamSfuClient;
-  connectionConfig?: RTCConfiguration;
+  publisher: RTCPeerConnectionType;
   candidates: ReplaySubject<ICETrickle>;
 };
 
-export const createPublisher = ({
-  connectionConfig,
+export const addPublisherListeners = <
+  RTCPeerConnectionType extends RTCPeerConnection,
+>({
+  publisher,
   rpcClient,
   candidates,
-}: PublisherOpts) => {
-  const publisher = new RTCPeerConnection(connectionConfig);
+}: PublisherOpts<RTCPeerConnectionType>) => {
   publisher.addEventListener('icecandidate', async (e) => {
     const { candidate } = e;
     if (!candidate) {
@@ -22,7 +24,7 @@ export const createPublisher = ({
     }
     await rpcClient.rpc.iceTrickle({
       sessionId: rpcClient.sessionId,
-      iceCandidate: JSON.stringify(candidate.toJSON()),
+      iceCandidate: getIceCandidate(candidate),
       peerType: PeerType.PUBLISHER_UNSPECIFIED,
     });
   });
@@ -69,6 +71,4 @@ export const createPublisher = ({
       }
     });
   });
-
-  return publisher;
 };