imp(messaging, websockets): bootstrap effects/middlewares only once
MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
@@ -7,8 +7,10 @@ import { createMicroservice, messagingListener, Transport, MsgEffect } from '@ma
const fib = (n: number): number =>
(n === 0 || n === 1) ? n : fib(n - 1) + fib(n - 2);
-const fibonacci$: MsgEffect = event$ =>
- event$.pipe(
+const fibonacci$: MsgEffect = event$ => {
+ console.log('Bootstrapped effect: "fibonacci$"');
+
+ return event$.pipe(
matchEvent('FIB'),
use(eventValidator$(t.number)),
mergeMap(event => of(event).pipe(
@@ -18,13 +20,17 @@ const fibonacci$: MsgEffect = event$ =>
catchError(error => of({ ...event, error: { name: error.name, message: error.message } } as Event)),
)),
);
+};
-const test$: MsgEffect = event$ =>
- event$.pipe(
+const test$: MsgEffect = event$ => {
+ console.log('Bootstrapped effect: "test$"');
+
+ return event$.pipe(
matchEvent('TEST'),
bufferCount(2),
mapTo({ type: 'TEST_RESULT' }),
);
+};
export const microservice = createMicroservice({
transport: Transport.AMQP,

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -5,6 +5,7 @@ import {
createEffectContext,
createListener,
} from '@marblejs/core';
+import { pipe } from 'fp-ts/lib/pipeable';
import { of, Observable } from 'rxjs';
import { map, publish, catchError, takeUntil } from 'rxjs/operators';
import {
@@ -46,10 +47,8 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
msgTransformer = jsonTransformer,
} = config ?? {};
- return connection => {
const combinedEffects = combineEffects(...effects);
const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
- const ctx = createEffectContext({ ask, client: connection });
const decode = (msg: TransportMessage<Buffer>): Event => ({
...msgTransformer.decode(msg.data),
@@ -60,24 +59,7 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
}
});
- const stream = connection.message$.pipe(
- takeUntil(connection.close$),
- map(decode),
- publish(e$ => combinedMiddlewares(e$, ctx)),
- publish(e$ => combinedEffects(e$, ctx)),
- publish(e$ => output$(e$, ctx)),
- publish(e$ => outputLogger$(e$, ctx)),
- catchError(error => {
- const e$ = of(error);
- return error$(e$, ctx).pipe(
- publish(e$ => errorLogger$(e$, ctx)),
- );
- }),
- );
-
- const subscribe = (event$: Observable<Event<unknown, any, string>>) =>
- event$.subscribe(
- event => {
+ const send = (connection: TransportLayerConnection) => (event: Event): void => {
const { metadata, type, payload, error } = event;
if (metadata && metadata.replyTo) {
@@ -89,7 +71,32 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
raw,
});
}
- },
+ };
+
+ return connection => {
+ const ctx = createEffectContext({ ask, client: connection });
+
+ const stream = pipe(
+ connection.message$.pipe(map(decode)),
+ e$ => combinedMiddlewares(e$, ctx),
+ e$ => combinedEffects(e$, ctx),
+ e$ => output$(e$, ctx),
+ e$ => outputLogger$(e$, ctx),
+ e$ => e$.pipe(
+ catchError(error => {
+ const e$ = of(error);
+ return error$(e$, ctx).pipe(
+ publish(e$ => errorLogger$(e$, ctx)),
+ );
+ }),
+ )
+ );
+
+ const subscribe = (event$: Observable<Event<unknown, any, string>>) =>
+ event$
+ .pipe(takeUntil(connection.close$))
+ .subscribe(
+ send(connection),
() => {
console.error('Unexpected stream error!'); // @TODO: handle unexpected error
subscribe(event$);

MODIFY
packages/websockets/src/server/websocket.server.listener.ts
packages/websockets/src/server/websocket.server.listener.ts
import { Observable, fromEvent, of } from 'rxjs';
-import { map, takeUntil, publish } from 'rxjs/operators';
+import { map, takeUntil } from 'rxjs/operators';
import {
combineEffects,
combineMiddlewares,
@@ -8,6 +8,7 @@ import {
Event,
EventError,
} from '@marblejs/core';
+import { pipe } from 'fp-ts/lib/pipeable';
import { WsEffect, WsErrorEffect, WsMiddlewareEffect, WsOutputEffect } from '../effects/websocket.effects.interface';
import { jsonTransformer } from '../transformer/websocket.json.transformer';
import { EventTransformer } from '../transformer/websocket.transformer.interface';
@@ -44,25 +45,29 @@ export const webSocketListener = createListener<WebSocketListenerConfig, WebSock
const handle = (client: WebSocketClientConnection) => {
const ctx = createEffectContext({ ask, client });
+ const close$ = fromEvent(client, 'close');
+ const message$ = fromEvent<MessageEvent>(client, 'message');
- const event$ = fromEvent<MessageEvent>(client, 'message').pipe(
- takeUntil(fromEvent(client, 'close')),
- map(e => eventTransformer.decode(e.data)),
- publish(e$ => middleware$(e$, ctx)),
- publish(e$ => effect$(e$, ctx)),
- publish(e$ => output$(e$, ctx)),
+ const stream = pipe(
+ message$,
+ e$ => e$.pipe(map(e => eventTransformer.decode(e.data))),
+ e$ => middleware$(e$, ctx),
+ e$ => effect$(e$, ctx),
+ e$ => output$(e$, ctx),
);
const subscribe = (input$: Observable<Event>) =>
- input$.subscribe(
+ input$
+ .pipe(takeUntil(close$))
+ .subscribe(
(event: Event) => client.sendResponse(event),
(error: EventError) => {
error$(of({ event: error.event, error }), ctx).subscribe(client.sendResponse);
- subscribe(event$);
+ subscribe(stream);
},
);
- subscribe(event$);
+ subscribe(stream);
};
handle.eventTransformer = eventTransformer;

imp(messaging, websockets): bootstrap effects/middlewares only once