chore: update API

waku-org · Apr 30, 2024 · b9b2ef0 · b9b2ef0
1 parent 2da0e44
commit b9b2ef0
Show file tree

Hide file tree

Showing 7 changed files with 42 additions and 12 deletions.
diff --git a/packages/core/src/lib/base_protocol.ts b/packages/core/src/lib/base_protocol.ts
@@ -47,8 +47,11 @@ export class BaseProtocol implements IBaseProtocolCore {
       this.addLibp2pEventListener
     );
   }
-  protected async getStream(peer: Peer): Promise<Stream> {
-    return this.streamManager.getStream(peer);
+  protected async getStream(peer: Peer): Promise<Stream | null> {
+    const stream = await this.streamManager.getStream(peer);
+
+    if (!stream) return null;
+    return stream;
   }
 
   public get peerStore(): PeerStore {

diff --git a/packages/core/src/lib/filter/index.ts b/packages/core/src/lib/filter/index.ts
@@ -92,6 +92,9 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
     contentTopics: ContentTopic[]
   ): Promise<void> {
     const stream = await this.getStream(peer);
+    if (!stream) {
+      throw new Error(`Failed to get stream for peer ${peer.id.toString()}`);
+    }
 
     const request = FilterSubscribeRpc.createSubscribeRequest(
       pubsubTopic,
@@ -128,6 +131,10 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
     contentTopics: ContentTopic[]
   ): Promise<void> {
     const stream = await this.getStream(peer);
+    if (!stream) {
+      throw new Error(`Failed to get stream for peer ${peer.id.toString()}`);
+    }
+
     const unsubscribeRequest = FilterSubscribeRpc.createUnsubscribeRequest(
       pubsubTopic,
       contentTopics
@@ -138,6 +145,9 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
 
   async unsubscribeAll(pubsubTopic: PubsubTopic, peer: Peer): Promise<void> {
     const stream = await this.getStream(peer);
+    if (!stream) {
+      throw new Error(`Failed to get stream for peer ${peer.id.toString()}`);
+    }
 
     const request = FilterSubscribeRpc.createUnsubscribeAllRequest(pubsubTopic);
 
@@ -167,6 +177,9 @@ export class FilterCore extends BaseProtocol implements IBaseProtocolCore {
 
   async ping(peer: Peer): Promise<void> {
     const stream = await this.getStream(peer);
+    if (!stream) {
+      throw new Error(`Failed to get stream for peer ${peer.id.toString()}`);
+    }
 
     const request = FilterSubscribeRpc.createSubscriberPingRequest();
 

diff --git a/packages/core/src/lib/light_push/index.ts b/packages/core/src/lib/light_push/index.ts
@@ -1,4 +1,4 @@
-import type { Peer, PeerId, Stream } from "@libp2p/interface";
+import type { Peer, PeerId } from "@libp2p/interface";
 import {
   Failure,
   IBaseProtocolCore,
@@ -100,18 +100,12 @@ export class LightPushCore extends BaseProtocol implements IBaseProtocolCore {
       };
     }
 
-    let stream: Stream | undefined;
-    try {
-      stream = await this.getStream(peer);
-    } catch (err) {
-      log.error(
-        `Failed to get a stream for remote peer${peer.id.toString()}`,
-        err
-      );
+    const stream = await this.getStream(peer);
+    if (!stream) {
       return {
         success: null,
         failure: {
-          error: ProtocolError.REMOTE_PEER_FAULT,
+          error: ProtocolError.NO_STREAM_AVAILABLE,
           peerId: peer.id
         }
       };

diff --git a/packages/core/src/lib/metadata/index.ts b/packages/core/src/lib/metadata/index.ts
@@ -86,6 +86,12 @@ class Metadata extends BaseProtocol implements IMetadata {
     }
 
     const stream = await this.getStream(peer);
+    if (!stream) {
+      return {
+        shardInfo: null,
+        error: ProtocolError.NO_STREAM_AVAILABLE
+      };
+    }
 
     const encodedResponse = await pipe(
       [request],

diff --git a/packages/core/src/lib/store/index.ts b/packages/core/src/lib/store/index.ts
@@ -93,6 +93,9 @@ export class StoreCore extends BaseProtocol implements IStoreCore {
       const historyRpcQuery = HistoryRpc.createQuery(queryOpts);
 
       const stream = await this.getStream(peer);
+      if (!stream) {
+        throw new Error(`Failed to get stream to peer ${peer.id.toString()}`);
+      }
 
       const res = await pipe(
         [historyRpcQuery.encode()],

diff --git a/packages/discovery/src/peer-exchange/waku_peer_exchange.ts b/packages/discovery/src/peer-exchange/waku_peer_exchange.ts
@@ -50,6 +50,12 @@ export class WakuPeerExchange extends BaseProtocol implements IPeerExchange {
     }
 
     const stream = await this.getStream(peer);
+    if (!stream) {
+      return {
+        error: ProtocolError.NO_STREAM_AVAILABLE,
+        peerInfos: null
+      };
+    }
 
     const res = await pipe(
       [rpcQuery.encode()],

diff --git a/packages/interfaces/src/protocols.ts b/packages/interfaces/src/protocols.ts
@@ -163,6 +163,11 @@ export enum ProtocolError {
    * on the connection manager before retrying.
    */
   NO_PEER_AVAILABLE = "No peer available",
+  /**
+   * Failure to find a stream to the peer. This may be because the connection with the peer is not still alive.
+   * Mitigation can be: retrying after a given time period, or mitigation for `NO_PEER_AVAILABLE` can be used.
+   */
+  NO_STREAM_AVAILABLE = "No stream available",
   /**
    * The remote peer did not behave as expected. Mitigation for `NO_PEER_AVAILABLE`
    * or `DECODE_FAILED` can be used.