fix(websockets, messaging): prevent closed stream dead loop
MODIFY
packages/@integration/src/messaging/client.ts
packages/@integration/src/messaging/client.ts
@@ -54,6 +54,19 @@ const timeout$ = r.pipe(
}),
);
+const error$ = r.pipe(
+ r.matchPath('/error'),
+ r.matchType('GET'),
+ r.useEffect((req$, { ask }) => {
+ const client = useContext(ClientToken)(ask);
+
+ return req$.pipe(
+ mergeMap(() => client.send({ type: 'ERROR' })),
+ mapTo(({ body: 'OK' })),
+ );
+ }),
+);
+
const fib$ = r.pipe(
r.matchPath('/fib/:number'),
r.matchType('GET'),
@@ -79,7 +92,7 @@ export const server = createServer({
port,
httpListener: httpListener({
middlewares: [logger$()],
- effects: [buffer$, timeout$, fib$],
+ effects: [buffer$, timeout$, error$, fib$],
}),
dependencies: [
bindEagerlyTo(ClientToken)(client),

MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
@@ -43,6 +43,15 @@ const timeout$: MsgEffect = event$ => {
);
};
+const error$: MsgEffect = event$ => {
+
+ return event$.pipe(
+ matchEvent('ERROR'),
+ tap(() => { throw new Error('test_error'); }),
+ );
+};
+
+
export const microservice = createMicroservice({
transport: Transport.AMQP,
options: {
@@ -51,7 +60,7 @@ export const microservice = createMicroservice({
queueOptions: { durable: false },
},
messagingListener: messagingListener({
- effects: [fibonacci$, buffer$, timeout$],
+ effects: [fibonacci$, buffer$, timeout$, error$],
}),
});

MODIFY
packages/@integration/src/websockets/websockets.server.ts
packages/@integration/src/websockets/websockets.server.ts
@@ -24,6 +24,12 @@ const add$: WsEffect = (event$, ctx) =>
map(payload => ({ type: 'SUM_RESULT', payload })),
);
+const error$: WsEffect = event$ =>
+ event$.pipe(
+ matchEvent('ERROR'),
+ mergeMap(() => throwError(new Error('some_error_message'))),
+ );
+
const connection$: WsConnectionEffect = req$ =>
req$.pipe(
mergeMap(req => iif(
@@ -37,6 +43,6 @@ export const webSocketServer = createWebSocketServer({
connection$,
webSocketListener: webSocketListener({
middlewares: [],
- effects: [add$],
+ effects: [add$, error$],
}),
});

MODIFY
packages/core/src/http/router/http.router.resolver.v2.ts
packages/core/src/http/router/http.router.resolver.v2.ts
@@ -60,14 +60,12 @@ export const resolveRouting = (
stream$.subscribe(
([res, req]) => req.response.send(res),
error => { throw unexpectedErrorWhileSendingOutputFactory(error) },
- () => subscribeOutput(stream$),
);
const subscribeError = (stream$: Observable<[HttpEffectResponse, HttpRequest]>) =>
stream$.subscribe(
([res, req]) => req.response.send(res),
error => { throw unexpectedErrorWhileSendingErrorFactory(error) },
- () => subscribeError(stream$),
);
subscribeOutput(outputFlow$);

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -4,14 +4,13 @@ import {
combineEffects,
createEffectContext,
createListener,
- EffectContext,
useContext,
LoggerToken,
LoggerTag,
LoggerLevel,
} from '@marblejs/core';
import { pipe } from 'fp-ts/lib/pipeable';
-import { of, Observable, Subject } from 'rxjs';
+import { Observable, Subject, defer } from 'rxjs';
import { map, catchError, takeUntil } from 'rxjs/operators';
import {
TransportMessage,
@@ -34,9 +33,6 @@ export interface MessagingListener {
(connection: TransportLayerConnection): void;
}
-const defaultEffect$: MsgEffect = msg$ =>
- msg$;
-
const defaultOutput$: MsgOutputEffect = msg$ =>
msg$;
@@ -52,10 +48,8 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
msgTransformer = jsonTransformer,
} = config ?? {};
- const getEffects = () => effects.length ? effects : [defaultEffect$];
-
const logger = useContext(LoggerToken)(ask);
- const combinedEffects = combineEffects(...getEffects());
+ const combinedEffects = combineEffects(...effects);
const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
const decode = (msg: TransportMessage<Buffer>): Event => ({
@@ -82,14 +76,8 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
}
};
- const processError$ = (ctx: EffectContext<TransportLayerConnection>) => (error: Error) =>
- pipe(
- of(error),
- e$ => error$(e$, ctx),
- e$ => errorLogger$(e$, ctx),
- );
-
return connection => {
+ const errorSubject = new Subject<Error>();
const eventSubject = new Subject<Event>();
const ctx = createEffectContext({ ask, client: connection });
@@ -97,7 +85,7 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
connection.message$,
e$ => e$.pipe(map(decode)),
e$ => combinedMiddlewares(e$, ctx),
- e$ => e$.pipe(catchError(processError$(ctx))),
+ e$ => e$.pipe(catchError(error => defer(() => processError(incomingEvent$)(error)))),
);
const outgoingEvent$ = pipe(
@@ -105,9 +93,20 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
e$ => combinedEffects(e$, ctx),
e$ => output$(e$, ctx),
e$ => outputLogger$(e$, ctx),
- e$ => e$.pipe(catchError(processError$(ctx))),
+ e$ => e$.pipe(catchError(error => defer(() => processError(outgoingEvent$)(error)))),
);
+ const errorEvent$ = pipe(
+ errorSubject.asObservable(),
+ e$ => error$(e$, ctx),
+ e$ => errorLogger$(e$, ctx),
+ );
+
+ const processError = (originStream$: Observable<any>) => (error: Error) => {
+ errorSubject.next(error);
+ return originStream$;
+ };
+
const subscribeIncomingEvent = (event$: Observable<Event<unknown, any, string>>) =>
event$
.pipe(takeUntil(connection.close$))
@@ -119,7 +118,11 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
logger({ tag: LoggerTag.MESSAGING, type, message, level: LoggerLevel.ERROR })();
subscribeIncomingEvent(event$);
},
- () => subscribeOutgoingEvent(event$),
+ () => {
+ const type = 'ServerListener';
+ const message = `IncomingEvent stream completes`;
+ logger({ tag: LoggerTag.MESSAGING, type, message, level: LoggerLevel.DEBUG })();
+ },
);
const subscribeOutgoingEvent = (event$: Observable<Event<unknown, any, string>>) =>
@@ -132,10 +135,15 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
const message = `Unexpected error for OutgoingEvent stream: "${error.name}", "${error.message}"`;
logger({ tag: LoggerTag.MESSAGING, type, message, level: LoggerLevel.ERROR })();
},
- () => subscribeOutgoingEvent(event$),
+ () => {
+ const type = 'ServerListener';
+ const message = `OutgoingEvent stream completes`;
+ logger({ tag: LoggerTag.MESSAGING, type, message, level: LoggerLevel.DEBUG })();
+ },
);
subscribeIncomingEvent(incomingEvent$);
subscribeOutgoingEvent(outgoingEvent$);
+ subscribeOutgoingEvent(errorEvent$);
};
});

MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
import { Event, matchEvent, use, combineEffects } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { Subject, throwError } from 'rxjs';
-import { map, tap, mergeMapTo, delay, bufferCount } from 'rxjs/operators';
+import { map, tap, delay, bufferCount, mergeMap } from 'rxjs/operators';
import { Transport } from '../../transport/transport.interface';
import { MsgEffect, MsgErrorEffect, MsgOutputEffect } from '../../effects/messaging.effects.interface';
import { AmqpStrategyOptions } from '../../transport/strategies/amqp.strategy.interface';
@@ -113,7 +113,7 @@ describe('messagingServer::AMQP', () => {
const event1$: MsgEffect = event$ =>
event$.pipe(
matchEvent(event1.type),
- mergeMapTo(throwError(error)),
+ mergeMap(() => throwError(error)),
);
const event2$: MsgEffect = event$ =>

MODIFY
packages/websockets/src/server/specs/websocket.server.spec.ts
packages/websockets/src/server/specs/websocket.server.spec.ts
@@ -155,7 +155,7 @@ describe('WebSocket server', () => {
const effect$: WsEffect = event$ =>
event$.pipe(
- mergeMapTo(throwError(new Error('test_message'))),
+ mergeMap(() => throwError(new Error('test_message'))),
);
const targetClient = testBed.getClient(0);

MODIFY
packages/websockets/src/server/websocket.server.listener.ts
packages/websockets/src/server/websocket.server.listener.ts
-import { Observable, fromEvent, of, Subject } from 'rxjs';
+import { Observable, fromEvent, Subject, defer } from 'rxjs';
import { map, takeUntil, catchError } from 'rxjs/operators';
import {
combineEffects,
@@ -7,7 +7,6 @@ import {
createListener,
Event,
EventError,
- EffectContext,
LoggerTag,
useContext,
LoggerToken,
@@ -34,7 +33,6 @@ export interface WebSocketListener {
eventTransformer: EventTransformer<any>;
}
-const defaultEffect$: WsEffect = msg$ => msg$;
const defaultOutput$: WsOutputEffect = (out$: Observable<Event>) => out$;
export const webSocketListener = createListener<WebSocketListenerConfig, WebSocketListener>(config => ask => {
@@ -46,20 +44,12 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
eventTransformer = jsonTransformer,
} = config ?? {};
- const getEffects = () => effects.length ? effects : [defaultEffect$];
-
const logger = useContext(LoggerToken)(ask);
const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
- const combinedEffects = combineEffects(...getEffects());
-
- const processError$ = (ctx: EffectContext<WebSocketClientConnection>) => (error: Error) =>
- pipe(
- of(error),
- e$ => error$(e$, ctx),
- e$ => errorLogger$(e$, ctx),
- );
+ const combinedEffects = combineEffects(...effects);
const handle = (client: WebSocketClientConnection) => {
+ const errorSubject = new Subject<Error>();
const eventSubject = new Subject<Event>();
const ctx = createEffectContext({ ask, client });
const close$ = fromEvent(client, 'close');
@@ -78,7 +68,7 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
e$ => e$.pipe(map(msg => eventTransformer.decode(msg.data))),
e$ => e$.pipe(map(applyMetadata)),
e$ => combinedMiddlewares(e$, ctx),
- e$ => e$.pipe(catchError(processError$(ctx))),
+ e$ => e$.pipe(catchError(error => defer(() => processError(incomingEvent$)(error)))),
);
const outgoingEvent$ = pipe(
@@ -89,9 +79,20 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
e$ => outputLogger$(e$, ctx),
// eslint-disable-next-line @typescript-eslint/no-unused-vars
e$ => e$.pipe(map(({ metadata, ...event }) => event)),
- e$ => e$.pipe(catchError(processError$(ctx))),
+ e$ => e$.pipe(catchError(error => defer(() => processError(outgoingEvent$)(error)))),
);
+ const errorEvent$ = pipe(
+ errorSubject.asObservable(),
+ e$ => error$(e$, ctx),
+ e$ => errorLogger$(e$, ctx),
+ );
+
+ const processError = (originStream$: Observable<any>) => (error: Error) => {
+ errorSubject.next(error);
+ return originStream$;
+ };
+
const subscribeIncomingEvent = (input$: Observable<Event>) =>
input$
.pipe(takeUntil(close$))
@@ -103,7 +104,11 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
logger({ tag: LoggerTag.WEBSOCKETS, type, message, level: LoggerLevel.ERROR })();
subscribeIncomingEvent(input$);
},
- () => subscribeOutgoingEvent(input$),
+ () => {
+ const type = 'ServerListener';
+ const message = `OutgoingEvent stream completes`;
+ logger({ tag: LoggerTag.WEBSOCKETS, type, message, level: LoggerLevel.DEBUG })();
+ },
);
const subscribeOutgoingEvent = (input$: Observable<Event>) =>
@@ -117,11 +122,16 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
logger({ tag: LoggerTag.WEBSOCKETS, type, message, level: LoggerLevel.ERROR })();
subscribeOutgoingEvent(input$);
},
- () => subscribeOutgoingEvent(input$),
+ () => {
+ const type = 'ServerListener';
+ const message = `OutgoingEvent stream completes`;
+ logger({ tag: LoggerTag.WEBSOCKETS, type, message, level: LoggerLevel.DEBUG })();
+ },
);
subscribeIncomingEvent(incomingEvent$);
subscribeOutgoingEvent(outgoingEvent$);
+ subscribeOutgoingEvent(errorEvent$);
};
handle.eventTransformer = eventTransformer;

fix(websockets, messaging): prevent closed stream dead loop