fix: updated moleculer-timeout-middleware with streams

src/middlewares/timeout.js

@@ -6,6 +6,7 @@
 
 "use strict";
 
+const { Stream } = require("stream");
 const { RequestTimeoutError } = require("../errors");
 const { METRIC } = require("../metrics");
 
@@ -38,6 +39,11 @@ module.exports = function (broker) {
 							nodeID,
 							timeout: ctx.options.timeout
 						});
+
+						if (ctx.params instanceof Stream) {
+							ctx.params.emit('moleculer-timeout-middleware', ctx.options.timeout)
+						}
+
 						err = new RequestTimeoutError({ action: actionName, nodeID });
 
 						broker.metrics.increment(METRIC.MOLECULER_REQUEST_TIMEOUT_TOTAL, {

src/transit.js

@@ -550,6 +550,13 @@ class Transit {
 			pass.$pool = new Map();
 
 			this.pendingReqStreams.set(payload.id, { sender: payload.sender, stream: pass });
+
+			pass.on('moleculer-timeout-middleware', (timeout) => {
+				setTimeout(() => {
+					this.pendingReqStreams.delete(payload.id);
+					this._destroyStreamIfPossible(pass, `Pending request stream ${payload.id} have been closed by timeout ${timeout} ms`)
+				}, 1000);
+			})
 		}
 
 		if (payload.seq > pass.$prevSeq + 1) {
@@ -864,6 +871,14 @@ class Transit {
 		// Add to pendings
 		this.pendingRequests.set(ctx.id, request);
 
+		if (request.stream) {
+			const pass = request.ctx.params
+
+			pass.on('moleculer-timeout-middleware', (timeout) => {
+				this._destroyStreamIfPossible(pass, `Request stream ${ctx.id} have been closed by timeout ${timeout} ms`)
+			})
+		}
+
 		// Publish request
 		return this.publish(packet)
 			.then(() => {
@@ -1103,6 +1118,7 @@ class Transit {
 	 */
 	_destroyStreamIfPossible(stream, errorMessage) {
 		if (!stream.destroyed && stream.destroy) {
+			stream.on('error', (err) => this.logger.error(err.message))
 			stream.destroy(new Error(errorMessage));
 		}
 	}