refactor to new interface

packages/interfaces/src/index.ts

@@ -18,3 +18,4 @@ export * from "./constants.js";
 export * from "./sharding.js";
 export * from "./health_status.js";
 export * from "./discovery.js";
+export * from "./webrtc.js";

packages/interfaces/src/waku.ts

@@ -15,6 +15,7 @@ import type { Protocols } from "./protocols.js";
 import type { IRelay } from "./relay.js";
 import type { ShardId } from "./sharding.js";
 import type { IStore } from "./store.js";
+import type { IWebRTC } from "./webrtc.js";
 
 export type CreateDecoderParams = {
   contentTopic: string;
@@ -62,6 +63,7 @@ export interface IWaku {
   store?: IStore;
   filter?: IFilter;
   lightPush?: ILightPush;
+  webRTC?: IWebRTC;
 
   /**
    * Emits events related to the Waku node.
@@ -272,6 +274,7 @@ export interface LightNode extends IWaku {
   store: IStore;
   filter: IFilter;
   lightPush: ILightPush;
+  webRTC: IWebRTC;
 }
 
 export interface RelayNode extends IWaku {

packages/interfaces/src/webrtc.ts

@@ -0,0 +1,93 @@
+import type { PeerId, TypedEventEmitter } from "@libp2p/interface";
+
+export enum WebRTCEvent {
+  InboundRequest = "webRTC:inbound-request",
+  Connected = "webRTC:connected",
+  Closed = "webRTC:closed",
+  Rejected = "webRTC:rejected"
+}
+
+export interface IWebRTCEvents {
+  /**
+   * Used to listen to incoming WebRTC connection request.
+   *
+   * @example
+   * ```typescript
+   * waku.addEventListener(WebRTCEvent.Inbound, (event) => {
+   *   const requesterPeerId = event.detail;
+   *
+   *   if (requesterPeerId.equals(expectedPeerId)) {
+   *     waku.webRTC.accept(requesterPeerId);
+   *   } else {
+   *     waku.webRTC.hangUp(requesterPeerId);
+   *   }
+   * });
+   */
+  [WebRTCEvent.InboundRequest]: CustomEvent<PeerId>;
+
+  /**
+   * Used to listen to get notified when a WebRTC connection is established.
+   *
+   * @example
+   * ```typescript
+   * waku.addEventListener(WebRTCEvent.Connected, (event) => {
+   *   const connection = event.detail; // RTCPeerConnection
+   * });
+   * ```
+   */
+  [WebRTCEvent.Connected]: CustomEvent<RTCPeerConnection>;
+}
+
+export type PeerIdOrString = PeerId | string;
+
+export type WebRTCDialOptions = {
+  peerId: PeerIdOrString;
+  timeoutMs?: number;
+};
+
+export interface IWebRTC {
+  /**
+   * Used to listen to incoming WebRTC connection request or progress of established connections.
+   */
+  events: TypedEventEmitter<IWebRTCEvents>;
+
+  /**
+   * Starts the listening to incoming WebRTC connection requests.
+   */
+  start(): Promise<void>;
+
+  /**
+   * Stops the listening to incoming WebRTC connection requests.
+   */
+  stop(): Promise<void>;
+
+  /**
+   * Dials a peer using Waku WebRTC protocol.
+   */
+  dial(options: WebRTCDialOptions): Promise<void>;
+
+  /**
+   * Accepts a WebRTC connection request from a peer.
+   */
+  accept(peerId: PeerIdOrString): void;
+
+  /**
+   * Hang up a WebRTC connection to a peer or incoming connection request.
+   */
+  hangUp(peerId: PeerIdOrString): void;
+
+  /**
+   * Checks if a WebRTC connection is established to a peer.
+   */
+  isConnected(peerId: PeerIdOrString): boolean;
+
+  /**
+   * Gets the list of connected peers using Waku WebRTC protocol.
+   */
+  getConnectedPeers(): PeerId[];
+
+  /**
+   * Gets map of WebRTC connections by peer ID.
+   */
+  getConnections(): Record<string, RTCPeerConnection>;
+}

packages/sdk/src/waku/waku.ts

@@ -20,6 +20,7 @@ import type {
   IStore,
   IWaku,
   IWakuEventEmitter,
+  IWebRTC,
   Libp2p,
   NetworkConfig
 } from "@waku/interfaces";
@@ -35,6 +36,7 @@ import { HealthIndicator } from "../health_indicator/index.js";
 import { LightPush } from "../light_push/index.js";
 import { PeerManager } from "../peer_manager/index.js";
 import { Store } from "../store/index.js";
+import { WebRTC } from "../webrtc/index.js";
 
 import { waitForRemotePeer } from "./wait_for_remote_peer.js";
 
@@ -52,6 +54,7 @@ export class WakuNode implements IWaku {
   public store?: IStore;
   public filter?: IFilter;
   public lightPush?: ILightPush;
+  public webRTC?: IWebRTC;
 
   public readonly events: IWakuEventEmitter = new TypedEventEmitter();
 
@@ -126,6 +129,21 @@ export class WakuNode implements IWaku {
       });
     }
 
+    if (this.lightPush && this.filter) {
+      const webRtcContentTopic = WebRTC.buildContentTopic(this.libp2p.peerId);
+
+      this.webRTC = new WebRTC({
+        lightPush: this.lightPush,
+        filter: this.filter,
+        encoder: this.createEncoder({
+          contentTopic: webRtcContentTopic
+        }),
+        decoder: this.createDecoder({
+          contentTopic: webRtcContentTopic
+        })
+      });
+    }
+
     log.info(
       "Waku node created",
       peerId,

packages/sdk/src/webrtc/index.ts

@@ -1 +1 @@
-export WebRTC from "./webrtc.js";
+export { WebRTC } from "./webrtc.js";

packages/sdk/src/webrtc/webrtc.ts

@@ -1,236 +1,121 @@
-/* eslint-disable eslint-comments/no-unlimited-disable */
+import { PeerId, TypedEventEmitter } from "@libp2p/interface";
-/* eslint-disable */
+import type {
-import type { IDecodedMessage, IDecoder, IEncoder, LightNode } from "@waku/sdk";
+  IDecodedMessage,
-import { createDecoder, createEncoder, bytesToUtf8, utf8ToBytes } from "@waku/sdk";
+  IDecoder,
-import type { MediaStreams } from "./media";
+  IEncoder,
-import { AudioSignal, SignalType } from "./audiosignal";
+  IFilter,
+  ILightPush,
+  IWebRTC,
+  IWebRTCEvents,
+  PeerIdOrString,
+  WebRTCDialOptions
+} from "@waku/interfaces";
 
-type WakuRTCParams = {
+type WebRTCConstructorOptions = {
-  node: LightNode;
+  lightPush: ILightPush;
-  config?: RTCConfiguration;
+  filter: IFilter;
+  decoder: IDecoder<IDecodedMessage>;
+  encoder: IEncoder;
 };
 
-const DEFAULT_STUN = "stun:stun.l.google.com:19302";
+export class WebRTC implements IWebRTC {
-const DEFAULT_CONTENT_TOPIC = "/waku-phone/1/sig01/proto";
+  private readonly lightPush: ILightPush;
+  private readonly filter: IFilter;
 
-export class WakuRTC {
+  private readonly decoder: IDecoder<IDecodedMessage>;
-  private started = false;
-  private inProgress = false;
-
-  private readonly node: LightNode;
   private readonly encoder: IEncoder;
-  private readonly decoder: IDecoder<any>;
 
-  public readonly rtcConnection: RTCPeerConnection;
+  public readonly events: TypedEventEmitter<IWebRTCEvents> =
-  private iceCandidates: RTCIceCandidate[] = [];
+    new TypedEventEmitter();
 
-  private inboundChannel: RTCDataChannel | undefined;
+  private isStarted = false;
-  private readonly outboundChannel: RTCDataChannel;
 
-  private filterUnsubscribe: undefined | (() => void);
+  public static buildContentTopic(peerId: PeerId): string {
-  public mediaStreams: MediaStreams | undefined;
+    return `/js-waku-webrtc/1/${peerId.toString()}/proto`;
-  public audioSignal: AudioSignal | undefined;
+  }
-  public isFree: boolean = true;
-  private inCallwith: string= '';
 
-  public constructor(params: WakuRTCParams) {
+  public constructor(options: WebRTCConstructorOptions) {
-    this.node = params.node;
+    this.lightPush = options.lightPush;
+    this.filter = options.filter;
 
-    this.encoder = createEncoder({
+    this.decoder = options.decoder;
-      contentTopic: DEFAULT_CONTENT_TOPIC,
+    this.encoder = options.encoder;
-      pubsubTopicShardInfo: {
-        clusterId: 42,
-        shard: 0
-      }
-    });
-    this.decoder = createDecoder(DEFAULT_CONTENT_TOPIC, { clusterId: 42 , shard: 0});
 
-    this.rtcConnection = new RTCPeerConnection({
+    this.handleInboundRequest = this.handleInboundRequest.bind(this);
-      iceServers: [{ urls: DEFAULT_STUN }],
-      ...(params.config || {})
-    });
-    this.outboundChannel = this.rtcConnection.createDataChannel("outbound");
-    this.onICECandidate = this.onICECandidate.bind(this);
-    this.onInboundChannel = this.onInboundChannel.bind(this);
   }
 
   public async start(): Promise<void> {
-    if (this.started || this.inProgress) {
+    if (this.isStarted) {
       return;
     }
 
-    this.inProgress = true;
+    this.isStarted = true;
-
+    await this.subscribeToInboundRequests();
-    this.rtcConnection.addEventListener("datachannel", this.onInboundChannel);
-    this.rtcConnection.addEventListener("icecandidate", this.onICECandidate);
-
-    try {
-      this.filterUnsubscribe = await this.node.filter.subscribeWithUnsubscribe(this.decoder, this.onWakuMessage.bind(this));
-    } catch(e) {
-      console.error("Error while Filter subscribe:", e);
-    }
-
-    this.inProgress = false;
-    this.started = true;
   }
 
   public async stop(): Promise<void> {
-    if (!this.started || this.inProgress) {
+    if (!this.isStarted) {
       return;
     }
 
-    this.inProgress = true;
+    this.isStarted = false;
+    await this.unsubscribeFromInboundRequests();
+  }
 
-    this.rtcConnection.removeEventListener("datachannel", this.onInboundChannel);
+  public async dial(options: WebRTCDialOptions): Promise<void> {
-    this.rtcConnection.removeEventListener("icecandidate", this.onICECandidate);
+    // TODO: implement
+  }
 
-    try {
+  public accept(peerId: PeerIdOrString): void {
-      this?.filterUnsubscribe?.();
+    // TODO: implement
-    } catch(e) {
+  }
-      console.error("Error while Filter unsubscribe:", e);
+
+  public hangUp(peerId: PeerIdOrString): void {
+    // TODO: implement
+  }
+
+  public isConnected(peerId: PeerIdOrString): boolean {
+    // TODO: implement
+    return false;
+  }
+
+  public getConnectedPeers(): PeerId[] {
+    // TODO: implement
+    return [];
+  }
+
+  public getConnections(): Record<string, RTCPeerConnection> {
+    // TODO: implement
+    return {};
+  }
+
+  private async subscribeToInboundRequests(): Promise<void> {
+    await this.filter.subscribe(this.decoder, this.handleInboundRequest);
+  }
+
+  private async unsubscribeFromInboundRequests(): Promise<void> {
+    await this.filter.unsubscribe(this.decoder);
+  }
+
+  private handleInboundRequest(message: IDecodedMessage): void {
+    /*
+    const decryptedMessage = decrypt(message.payload, this.privateKey);
+    switch (decryptedMessage.type) {
+      case "dial":
+        break;
+      case "ack":
+        break;
+      case "answer":
+        break;
+      case "reject":
+        break;
+      case "candidate":
+        break;
+      case "close":
+        break;
+      default:
+        break;
     }
-
+    */
-    this.inProgress = false;
-    this.started = false;
   }
-
+}
-  public async initiateConnection(peerId: string): Promise<void> {
-    this.audioSignal?.playSignal(SignalType.RINGING);
-
-    this.inCallwith = peerId;
-    await this.sendWakuMessage("call", '');
-  }
-
-  public async hangupCall(): Promise<void> {
-    await this.sendWakuMessage("bye", "");
-    this.inCallwith = '';
-    this.isFree = true;
-  }
-
-  private onInboundChannel(event: RTCDataChannelEvent): void {
-    this.inboundChannel = event.channel;
-    this.inboundChannel.addEventListener("message", (event) => {
-      console.log("Received message:", event.data);
-    });
-  }
-
-  private async onICECandidate(event: RTCPeerConnectionIceEvent): Promise<void> {
-    if (!event.candidate) {
-      return;
-    }
-
-    this.iceCandidates.push(event.candidate);
-    await this.sendWakuMessage("candidate", this.iceCandidates);
-  }
-
-  private async onWakuMessage(message: IDecodedMessage): Promise<void> {
-    const payload = bytesToUtf8(message.payload);
-    const data = JSON.parse(payload);
-
-    if (data.receiver !== this.node.peerId.toString() ||
-        data.sender === this.node.peerId.toString()) {
-      return;
-    }
-    console.log("received a waku message with payload:", data);
-
-    if (data.type === "call") {
-      await this.onConnectionRequestMessage(data.receiver, data.sender);
-    } else if (data.type === "candidate") {
-      await this.onCandidateMessage(data.payload);
-    } else if (data.type === "offer") {
-      await this.onOfferMessage(data.payload);
-    } else if (data.type === "answer") {
-      await this.onAnswerMessage(data.payload);
-    } else if (data.type === "ready") {
-      this.onReadyMessage();
-    }else if (data.type === "bye"){
-      this.onByeMessage();
-    } else if (data.type === "busy"){
-      this.onBusyMessage();
-    }
-  }
-
-  private async onBusyMessage() {
-    this.audioSignal?.playSignal(SignalType.BUSY, 5000);
-    this.isFree = true;
-    this.inCallwith = '';
-    this.rtcConnection.close();
-  }
-
-  private async onByeMessage() {
-    if (this.isFree){
-      return;
-    }
-    this.isFree = true;
-    this.inCallwith = '';
-    this.rtcConnection.close();
-    this.mediaStreams?.stopStreams();
-  }
-
-  private async onCandidateMessage(candidates: RTCIceCandidate[]): Promise<void> {
-    for (const candidate of candidates) {
-      await this.rtcConnection.addIceCandidate(
-        new RTCIceCandidate(candidate)
-      );
-    }
-  }
-
-  private async onOfferMessage(offer: RTCSessionDescriptionInit): Promise<void> {
-    await this.rtcConnection.setRemoteDescription(
-      new RTCSessionDescription(offer)
-    );
-
-    const answer = await this.rtcConnection.createAnswer();
-    this.rtcConnection.setLocalDescription(answer);
-
-    await this.sendWakuMessage("answer", answer);
-    this.audioSignal?.stopSignal();
-  }
-
-  private async onAnswerMessage(answer: RTCSessionDescriptionInit) {
-    await this.rtcConnection.setRemoteDescription(
-      new RTCSessionDescription(answer)
-    );
-  }
-
-  private async onConnectionRequestMessage(peerId: string, remotePeerId: string): Promise<void> {
-    if(!this.isFree) {
-      await this.sendWakuMessage("busy",'', remotePeerId);
-      return;
-    }
-    //this.mediaStreams?.setupLocalStream();
-    //this.mediaStreams?.setupRemoteStream();
-    this.isFree = false;
-    this.inCallwith = remotePeerId;
-    const offer = await this.rtcConnection.createOffer();
-    await this.rtcConnection.setLocalDescription(offer);
-
-    await this.sendWakuMessage("offer", offer);
-    
-    if (this.iceCandidates.length) {
-      await this.sendWakuMessage("candidate", this.iceCandidates);
-    }
-  }
-
-  private async onReadyMessage() {
-    console.log("RTC: partner is ready");
-  }
-
-  private async sendWakuMessage(type: string, payload: any, remotePeerId:string=this.inCallwith): Promise<void> {
-    const response = await this.node.lightPush.send(this.encoder, {
-      payload: utf8ToBytes(JSON.stringify({
-        type,
-        payload,
-        sender: this.node.peerId.toString(),
-        receiver: remotePeerId
-      }))
-    });
-
-    console.log(`sendWakuMessage of type:${type}, with ${response} , receiver ${remotePeerId}`);
-  }
-
-  public sendChatMessage(message: string): void {
-    this.outboundChannel.send(message);
-  }
-
-}