Cleanup and refactor WebRTC streaming

packages/rrweb/scripts/stream.js

@@ -10,7 +10,7 @@ import { fileURLToPath } from 'url';
 import { startServer, getServerURL } from './utils.js';
 
 // Turn on devtools for debugging:
-const devtools = false;
+const devtools = true;
 const defaultURL = 'https://webglsamples.org/';
 
 const __filename = fileURLToPath(import.meta.url);
@@ -24,38 +24,39 @@ function getCode() {
 }
 
 async function startRecording(page, serverURL) {
-  page.on('recording - console', (msg) => console.log('PAGE LOG:', msg.text()));
-
-  await page.evaluate((serverURL) => {
-    const el = document.createElement('script');
-    el.addEventListener('load', () => {
-      const win = window;
-      win.__IS_RECORDING__ = true;
-      win.events = [];
-      win.rrweb?.record({
-        emit: (event) => {
-          win.events?.push(event);
-          win._replLog?.(event);
-        },
-        recordCanvas: true,
-        collectFonts: true,
-        inlineImages: true,
-        sampling: {
-          canvas: 'webrtc',
-        },
+  try {
+    await page.evaluate((serverURL) => {
+      const el = document.createElement('script');
+      el.addEventListener('load', () => {
+        const win = window;
+        win.__IS_RECORDING__ = true;
+        win.events = [];
+        window.record = win.rrweb.record;
+        window.record({
+          emit: (event) => {
+            win.events?.push(event);
+            win._captureEvent?.(event);
+          },
+          recordCanvas: true,
+          collectFonts: true,
+          inlineImages: true,
+          sampling: {
+            canvas: 'webrtc',
+          },
+        });
       });
-    });
-    el.src = `${serverURL}/rrweb.js`;
-    document.head.appendChild(el);
-  }, serverURL);
+      el.src = `${serverURL}/rrweb.js`;
+      document.head.appendChild(el);
+    }, serverURL);
+  } catch (error) {
+    console.error(error);
+  }
 }
 
 async function startReplay(page, serverURL, recordedPage) {
-  page.on('replay - console', (msg) => console.log('PAGE LOG:', msg.text()));
-
-  await page.exposeFunction('_signal', (signal) => {
-    recordedPage.evaluate((signal) => {
-      window.p.signal(signal);
+  await page.exposeFunction('_signal', async (signal) => {
+    await recordedPage.evaluate((signal) => {
+      window.record.webRTCSignal(signal);
     }, signal);
   });
 
@@ -65,6 +66,9 @@ async function startReplay(page, serverURL, recordedPage) {
       window.replayer = new rrweb.Replayer([], {
         UNSAFE_replayCanvas: true,
         liveMode: true,
+        webRTCSignalCallback: async (data) => {
+          await _signal(JSON.stringify(data));
+        },
       });
       window.replayer.startLive();
     });
@@ -181,6 +185,12 @@ void (async () => {
     // disables content security policy which enables us to insert rrweb as a script tag
     await recordedPage.setBypassCSP(true);
 
+    replayerPage.on('console', (msg) =>
+      console.log('REPLAY PAGE LOG:', msg.text()),
+    );
+    recordedPage.on('console', (msg) =>
+      console.log('RECORD PAGE LOG:', msg.text()),
+    );
     await startReplay(replayerPage, serverURL, recordedPage);
 
     await Promise.all([
@@ -195,7 +205,7 @@ void (async () => {
 
     if (!replayerPage) throw new Error('No replayer page found');
 
-    await recordedPage.exposeFunction('_replLog', (event) => {
+    await recordedPage.exposeFunction('_captureEvent', (event) => {
       replayerPage.evaluate((event) => {
         window.replayer?.addEvent(event);
       }, event);

packages/rrweb/src/record/index.ts

@@ -40,6 +40,7 @@ function wrapEvent(e: event): eventWithTime {
 let wrappedEmit!: (e: eventWithTime, isCheckout?: boolean) => void;
 
 let takeFullSnapshot!: (isCheckout?: boolean) => void;
+let canvasManager!: CanvasManager;
 
 const mirror = createMirror();
 function record<T = eventWithTime>(
@@ -221,7 +222,7 @@ function record<T = eventWithTime>(
     mutationCb: wrappedMutationEmit,
   });
 
-  const canvasManager = new CanvasManager({
+  canvasManager = new CanvasManager({
     recordCanvas,
     mutationCb: wrappedCanvasMutationEmit,
     win: window,
@@ -529,6 +530,15 @@ record.addCustomEvent = <T>(tag: string, payload: T) => {
   );
 };
 
+/**
+ * Insert response to an WebRTC Offer to complete the WebRTC connection
+ *
+ * @param signal - WebRTC signal supplied by the replayer's `webRTCSignalCallback` function
+ */
+record.webRTCSignal = (signal: RTCSessionDescriptionInit) => {
+  canvasManager.webRTCSignalCallback(signal);
+};
+
 record.freezePage = () => {
   mutationBuffers.forEach((buf) => buf.freeze());
 };

packages/rrweb/src/record/observers/canvas/canvas-manager.ts

@@ -8,6 +8,7 @@ import type {
   IWindow,
   listenerHandler,
   CanvasArg,
+  WebRTCDataChannel,
 } from '../../../types';
 import { CanvasContext } from '../../../types';
 import initCanvas2DMutationObserver from './2d';
@@ -198,31 +199,68 @@ export class CanvasManager {
     };
   }
 
-  // private sdpOffer: RTCSessionDescriptionInit;
+  private sdpOffer: RTCSessionDescriptionInit;
+  private peer: SimplePeer.Instance;
 
-  private setupPeer(id: number, stream: MediaStream) {
-    console.log('setupPeer', id);
-    const p = new SimplePeer({
-      initiator: true,
-      stream,
-      trickle: false, // only send one WebRTC offer per canvas
-    });
+  public webRTCSignalCallback(signal: RTCSessionDescriptionInit) {
+    this.peer.signal(signal);
+  }
 
-    (window as any).p = p;
+  private startStream(id: number, stream: MediaStream) {
+    const data: WebRTCDataChannel = {
+      id,
+    };
+    this.peer.send(JSON.stringify(data));
+    this.peer.addStream(stream);
+    this.peer.send(JSON.stringify(data));
+  }
 
-    p.on('signal', (data: RTCSessionDescriptionInit) => {
-      this.mutationCb({
-        id,
-        type: CanvasContext.WebRTC,
-        msg: data,
-        stream,
+  private peerConnected = false;
+  private idSentSet: Set<number> = new Set();
+  private setupPeer(
+    id: number,
+    stream: MediaStream,
+    streamMap: Map<number, MediaStream>,
+  ) {
+    if (!this.peer) {
+      this.peer = new SimplePeer({
+        initiator: true,
+        // trickle: false, // only create one WebRTC offer per session
       });
-    });
 
-    p.on('connect', () => {
-      // eslint-disable-next-line prefer-rest-params
-      console.log('connected', arguments);
-    });
+      this.peer.on('error', (err) => {
+        console.error('record peer error', err);
+      });
+
+      this.peer.on('signal', (data: RTCSessionDescriptionInit) => {
+        this.sdpOffer = data;
+        this.mutationCb({
+          id,
+          type: CanvasContext.WebRTC,
+          msg: this.sdpOffer,
+          stream,
+        });
+        this.idSentSet.add(id);
+      });
+
+      this.peer.on('connect', () => {
+        this.peerConnected = true;
+        for (const [id, stream] of streamMap) {
+          this.startStream(id, stream);
+          if (!this.idSentSet.has(id)) {
+            this.mutationCb({
+              id,
+              type: CanvasContext.WebRTC,
+              msg: this.sdpOffer,
+              stream,
+            });
+            this.idSentSet.add(id);
+          }
+        }
+      });
+    } else if (this.sdpOffer && this.peerConnected) {
+      this.startStream(id, stream);
+    }
   }
 
   private initCanvasWebRTCObserver(win: IWindow, blockClass: blockClass) {
@@ -235,15 +273,10 @@ export class CanvasManager {
         .querySelectorAll(`canvas:not(.${blockClass as string} *)`)
         .forEach((canvas: HTMLCanvasElement) => {
           const id = this.mirror.getId(canvas);
-          if (streamMap.has(id)) return;
-
-          // streamMap.set(id, (null as unknown) as MediaStream);
-          // setTimeout(() => {
-          console.log('captureStream', canvas, canvas.width, canvas.height);
+          if (id === -1 || streamMap.has(id)) return;
           const stream = canvas.captureStream();
           streamMap.set(id, stream);
-          this.setupPeer(id, stream);
-          // }, 2000);
+          this.setupPeer(id, stream, streamMap);
         });
       rafId = requestAnimationFrame(captureStreams);
     };

packages/rrweb/src/replay/canvas/index.ts

@@ -4,6 +4,7 @@ import {
   canvasMutationCommand,
   canvasMutationData,
   canvasMutationParam,
+  WebRTCDataChannel,
 } from '../../types';
 import webglMutation from './webgl';
 import canvas2DMutation from './2d';
@@ -16,20 +17,29 @@ export default async function canvasMutation({
   imageMap,
   canvasEventMap,
   errorHandler,
+  mirror,
+  webRTCSignalCallback,
 }: {
   event: Parameters<Replayer['applyIncremental']>[0];
   mutation: canvasMutationData;
   target: HTMLCanvasElement;
   imageMap: Replayer['imageMap'];
   canvasEventMap: Replayer['canvasEventMap'];
   errorHandler: Replayer['warnCanvasMutationFailed'];
+  mirror: Replayer['mirror'];
+  webRTCSignalCallback: null | ((signal: RTCSessionDescriptionInit) => void);
 }): Promise<void> {
   try {
+    console.log('canvasMutation', mutation);
     const precomputedMutation: canvasMutationParam =
       canvasEventMap.get(event) || mutation;
 
     if (precomputedMutation.type === CanvasContext.WebRTC) {
-      return setupWebRTC(precomputedMutation, target);
+      if (!webRTCSignalCallback)
+        throw new Error(
+          'webRTCSignalCallback is null, webRTC events will be ignored and Canvas or Video elements will be empty',
+        );
+      return setupWebRTC(precomputedMutation, webRTCSignalCallback, mirror);
     }
 
     const commands: canvasMutationCommand[] =
@@ -125,41 +135,55 @@ function startStream(
   }
 }
 
+let peer: SimplePeer.Instance | null = null;
+let streamId: number;
 function setupWebRTC(
   mutation: canvasMutationParam & {
     type: CanvasContext.WebRTC;
   },
-  target: HTMLCanvasElement | HTMLVideoElement,
-  // mirror: Replayer['mirror'],
+  webRTCSignalCallback: (signal: RTCSessionDescriptionInit) => void,
+  mirror: Replayer['mirror'],
 ) {
-  // eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
-  const p = ((window as any).p = new SimplePeer({
-    initiator: false,
-    trickle: false,
-  }));
-
-  p.on('error', (err: Error) => console.log('error', err));
-
-  p.on('signal', (data: SimplePeer.SignalData) => {
-    console.log('SIGNAL', JSON.stringify(data));
-    // eslint-disable-next-line @typescript-eslint/ban-ts-comment
-    // @ts-ignore
-    // eslint-disable-next-line @typescript-eslint/no-unsafe-call
-    _signal(JSON.stringify(data));
-  });
-
-  p.on('connect', () => {
-    console.log('CONNECT');
-  });
-
-  // p.on('data', (data) => {
-  //   console.log('data: ', data);
-  // });
-
-  p.on('stream', (stream: MediaStream) => {
-    // got remote video stream, now let's show it in a video or canvas element
-    startStream(target, stream);
-  });
-
-  p.signal(mutation.msg);
+  if (!peer) {
+    peer = new SimplePeer({
+      initiator: false,
+      // trickle: false,
+    });
+
+    peer.on('error', (err: Error) => {
+      peer = null;
+      console.log('error', err);
+    });
+
+    peer.on('close', () => {
+      peer = null;
+      console.log('closing');
+    });
+
+    peer.on('signal', (data: RTCSessionDescriptionInit) => {
+      webRTCSignalCallback(data);
+    });
+
+    peer.on('connect', () => {
+      console.log('connect');
+    });
+
+    peer.on('data', (data: SimplePeer.SimplePeerData) => {
+      try {
+        const json = JSON.parse(data as string) as WebRTCDataChannel;
+        streamId = json.id;
+      } catch (error) {
+        console.error('Could not parse data', error);
+      }
+    });
+
+    peer.on('stream', (stream: MediaStream) => {
+      // got remote video stream, now let's show it in a video or canvas element
+      const target = mirror.getNode(streamId) as
+        | HTMLCanvasElement
+        | HTMLVideoElement;
+      startStream(target, stream);
+    });
+  }
+  peer.signal(mutation.msg);
 }