fix: add an exception catch to the streaming function (#451)

Co-authored-by: droak <[email protected]>

packages/node/src/handlers.ts

@@ -12,13 +12,18 @@ import { deserializeStateMessage, serializeStateMessage } from "./utils.js";
 */
 export async function drpMessagesHandler(node: DRPNode, stream?: Stream, data?: Uint8Array) {
 	let message: NetworkPb.Message;
-	if (stream) {
-		const byteArray = await streamToUint8Array(stream);
-		message = NetworkPb.Message.decode(byteArray);
-	} else if (data) {
-		message = NetworkPb.Message.decode(data);
-	} else {
-		log.error("::messageHandler: Stream and data are undefined");
+	try {
+		if (stream) {
+			const byteArray = await streamToUint8Array(stream);
+			message = NetworkPb.Message.decode(byteArray);
+		} else if (data) {
+			message = NetworkPb.Message.decode(data);
+		} else {
+			log.error("::messageHandler: Stream and data are undefined");
+			return;
+		}
+	} catch (err) {
+		log.error("::messageHandler: Error decoding message", err);
 		return;
 	}
 

packages/node/tests/message_handler.test.ts → packages/node/tests/handlers.test.ts

@@ -1,14 +1,52 @@
-import { Connection, IdentifyResult, Libp2p } from "@libp2p/interface";
+import type { Connection, IdentifyResult, Libp2p, Stream } from "@libp2p/interface";
 import { SetDRP } from "@ts-drp/blueprints";
-import { DRPNetworkNode, DRPNetworkNodeConfig, NetworkPb } from "@ts-drp/network";
+import { DRPNetworkNode, type DRPNetworkNodeConfig, NetworkPb } from "@ts-drp/network";
 import { DrpType } from "@ts-drp/object";
-import { DRPObject, ObjectACL } from "@ts-drp/object";
+import { type DRPObject, ObjectACL } from "@ts-drp/object";
 import { raceEvent } from "race-event";
-import { beforeAll, describe, expect, test, afterAll } from "vitest";
+import { beforeAll, describe, expect, test, afterAll, vi } from "vitest";
 
-import { signGeneratedVertices } from "../src/handlers.js";
+import { drpMessagesHandler, signGeneratedVertices } from "../src/handlers.js";
 import { DRPNode } from "../src/index.js";
 
+describe("drpMessagesHandler inputs", () => {
+	let node: DRPNode;
+	const consoleSpy = vi.spyOn(console, "error");
+
+	beforeAll(async () => {
+		node = new DRPNode();
+	});
+
+	test("normal inputs", async () => {
+		await drpMessagesHandler(node);
+		expect(consoleSpy).toHaveBeenLastCalledWith(
+			"drp::node ::messageHandler: Stream and data are undefined"
+		);
+
+		const msg = NetworkPb.Message.create({
+			sender: node.networkNode.peerId,
+			type: -1,
+			data: new Uint8Array(),
+		});
+		await drpMessagesHandler(node, undefined, msg.data);
+		expect(consoleSpy).toHaveBeenLastCalledWith("drp::node ::messageHandler: Invalid operation");
+
+		await drpMessagesHandler(
+			node,
+			{
+				close: async () => {},
+				closeRead: async () => {},
+				closeWrite: async () => {},
+			} as Stream,
+			undefined
+		);
+		expect(consoleSpy).toHaveBeenLastCalledWith(
+			"drp::node ::messageHandler: Error decoding message",
+			new Error("Empty pipeline")
+		);
+	});
+});
+
 describe("Handle message correctly", () => {
 	const controller = new AbortController();
 	let node1: DRPNode;