use IReceiver with toSubscriptionIterator

package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "@waku/noise",
-  "version": "0.0.2",
+  "version": "0.0.3",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@waku/noise",
-      "version": "0.0.2",
+      "version": "0.0.3",
       "license": "Apache-2.0 OR MIT",
       "dependencies": {
         "@stablelib/chacha20poly1305": "^1.0.1",

src/pairing.ts

@@ -1,6 +1,6 @@
 import { HMACDRBG } from "@stablelib/hmac-drbg";
 import { randomBytes } from "@stablelib/random";
-import type { IDecoder, ISender } from "@waku/interfaces";
+import type { IDecoder, ISender, IReceiver, IAsyncIterator } from "@waku/interfaces";
 import debug from "debug";
 import { EventEmitter } from "eventemitter3";
 import { pEvent } from "p-event";
@@ -23,32 +23,6 @@ import { QR } from "./qr.js";
 
 const log = debug("waku:noise:pairing");
 
-/**
- * Responder interface than an object must implement so the pairing object can receive noise messages
- */
-export interface Responder {
-  /**
-   * subscribe to receive the messages from a content topic
-   * @param decoder Decoder to use to decrypt the NoiseHandshakeMessages
-   */
-  subscribe(decoder: IDecoder<NoiseHandshakeMessage>): Promise<void>;
-
-  /**
-   * should return messages received in a content topic
-   * messages should be kept in a queue, meaning that nextMessage
-   * will call pop in the queue to remove the oldest message received
-   * (it's important to maintain order of received messages)
-   * @param contentTopic content topic to get the next message from
-   */
-  nextMessage(contentTopic: string): Promise<NoiseHandshakeMessage>;
-
-  /**
-   * Stop the subscription to the content topic
-   * @param contentTopic
-   */
-  stop(contentTopic: string): Promise<void>;
-}
-
 function delay(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
@@ -111,7 +85,7 @@ export class WakuPairing {
    */
   constructor(
     private sender: ISender,
-    private responder: Responder,
+    private responder: IReceiver,
     private myStaticKey: KeyPair,
     pairingParameters: InitiatorParameters | ResponderParameters,
     private myEphemeralKey: KeyPair = generateX25519KeyPair()
@@ -196,8 +170,8 @@ export class WakuPairing {
   }
 
   private async executeReadStepWithNextMessage(
-    contentTopic: string,
-    messageNametag: Uint8Array
+    messageNametag: Uint8Array,
+    iterator: AsyncIterator<NoiseHandshakeMessage>
   ): Promise<HandshakeStepResult> {
     // TODO: create test unit for this function
     let stopLoop = false;
@@ -212,9 +186,14 @@ export class WakuPairing {
 
     while (!stopLoop) {
       try {
-        const hsMessage = await this.responder.nextMessage(contentTopic);
+        const item = await iterator.next();
+
+        if (!item.value) {
+          throw Error("Received no message");
+        }
+
         const step = this.handshake.stepHandshake({
-          readPayloadV2: hsMessage.payloadV2,
+          readPayloadV2: item.value.payloadV2,
           messageNametag,
         });
         return step;
@@ -233,7 +212,7 @@ export class WakuPairing {
   private async initiatorHandshake(): Promise<[NoiseSecureTransferEncoder, NoiseSecureTransferDecoder]> {
     // Subscribe to the contact content topic
     const decoder = new NoiseHandshakeDecoder(this.contentTopic);
-    await this.responder.subscribe(decoder);
+    const subscriptionIterator = await this.responder.toSubscriptionIterator(decoder);
 
     // The handshake initiator writes a Waku2 payload v2 containing the handshake message
     // and the (encrypted) transport message
@@ -263,9 +242,9 @@ export class WakuPairing {
 
     // 2nd step
     // <- sB, eAsB    {r}
-    hsStep = await this.executeReadStepWithNextMessage(this.contentTopic, this.handshake.hs.toMessageNametag());
+    hsStep = await this.executeReadStepWithNextMessage(this.handshake.hs.toMessageNametag(), subscriptionIterator.iterator);
 
-    await this.responder.stop(this.contentTopic);
+    await subscriptionIterator.stop();
 
     if (!this.handshake.hs.rs) throw new Error("invalid handshake state");
 
@@ -299,11 +278,11 @@ export class WakuPairing {
   private async responderHandshake(): Promise<[NoiseSecureTransferEncoder, NoiseSecureTransferDecoder]> {
     // Subscribe to the contact content topic
     const decoder = new NoiseHandshakeDecoder(this.contentTopic);
-    await this.responder.subscribe(decoder);
+    const subscriptionIterator = await this.responder.toSubscriptionIterator(decoder);
 
     // the received reads the initiator's payloads, and returns the (decrypted) transport message the initiator sent
     // Note that the received verifies if the received payloadV2 has the expected messageNametag set
-    let hsStep = await this.executeReadStepWithNextMessage(this.contentTopic, this.qrMessageNameTag);
+    let hsStep = await this.executeReadStepWithNextMessage(this.qrMessageNameTag, subscriptionIterator.iterator);
 
     const initiatorCommittedStaticKey = new Uint8Array(hsStep.transportMessage);
 
@@ -333,9 +312,9 @@ export class WakuPairing {
     // -> sA, sAeB, sAsB  {s}
 
     // The responder reads the initiator's payload sent by the initiator
-    hsStep = await this.executeReadStepWithNextMessage(this.contentTopic, this.handshake.hs.toMessageNametag());
+    hsStep = await this.executeReadStepWithNextMessage(this.handshake.hs.toMessageNametag(), subscriptionIterator.iterator);
 
-    await this.responder.stop(this.contentTopic);
+    await subscriptionIterator.stop();
 
     if (!this.handshake.hs.rs) throw new Error("invalid handshake state");