Merge branch 'fix/pipe-uncaught-error' into 'dev'

fix(rosenet-node): fix issue in protocol handler pipe Closes #77 See merge request ergo/rosen-bridge/rosenet!37
rosen-bridge · Sep 16, 2024 · 8ceb3a6 · 8ceb3a6
2 parents 7686c5f + 95ead7a
commit 8ceb3a6
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 17 deletions.
diff --git a/.changeset/lovely-hats-join.md b/.changeset/lovely-hats-join.md
@@ -0,0 +1,5 @@
+---
+'@rosen-bridge/rosenet-node': patch
+---
+
+Fix issue of RoseNet direction protocol listener pipe not getting awaited
diff --git a/packages/rosenet-node/lib/rosenet-direct/handleIncomingMessage.ts b/packages/rosenet-node/lib/rosenet-direct/handleIncomingMessage.ts
@@ -23,23 +23,32 @@ const handleIncomingMessageFactory =
             transient: connection.transient,
           },
         );
-        pipe(
-          stream,
-          decode,
-          async function* (source) {
-            for await (const message of source) {
-              RoseNetNodeContext.logger.debug(
-                'message received, calling handler and sending ack',
-                {
-                  message,
-                },
-              );
-              handler(connection.remotePeer.toString(), message);
-              yield Uint8Array.of(ACK_BYTE);
-            }
-          },
-          stream,
-        );
+        try {
+          await pipe(
+            stream,
+            decode,
+            async function* (source) {
+              for await (const message of source) {
+                RoseNetNodeContext.logger.debug(
+                  'message received, calling handler and sending ack',
+                  {
+                    message,
+                  },
+                );
+                handler(connection.remotePeer.toString(), message);
+                yield Uint8Array.of(ACK_BYTE);
+              }
+            },
+            stream,
+          );
+        } catch (error) {
+          RoseNetNodeContext.logger.warn(
+            'An error occurred while reading from stream',
+            {
+              error,
+            },
+          );
+        }
       },
       { runOnTransientConnection: true },
     );