refactor(messaging): process server related events in factory function instead of server listener
MODIFY
packages/messaging/src/server/messaging.server.interface.ts
packages/messaging/src/server/messaging.server.interface.ts
-import { BoundDependency } from '@marblejs/core';
+import { BoundDependency, ListenerServer } from '@marblejs/core';
import { MsgServerEffect } from '../effects/messaging.effects.interface';
import { TransportStrategy, TransportLayerConnection } from '../transport/transport.interface';
import { messagingListener } from './messaging.server.listener';
@@ -14,6 +14,4 @@ export type CreateMicroserviceConfig =
& ConfigurationBase
;
-export interface Microservice {
- (): Promise<TransportLayerConnection>;
-}
+export interface Microservice extends ListenerServer<TransportLayerConnection> {}

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -3,15 +3,12 @@ import {
combineMiddlewares,
combineEffects,
createEffectContext,
- useContext,
EventError,
createListener,
} from '@marblejs/core';
-import * as O from 'fp-ts/lib/Option';
-import { pipe } from 'fp-ts/lib/pipeable';
import { flow } from 'fp-ts/lib/function';
import { Observable, Subscription, Subject, of, zip, OperatorFunction} from 'rxjs';
-import { map, publish, takeUntil, catchError, take, mergeMapTo } from 'rxjs/operators';
+import { map, publish, takeUntil, catchError, mergeMapTo } from 'rxjs/operators';
import {
TransportMessage,
TransportMessageTransformer,
@@ -20,8 +17,6 @@ import {
import { jsonTransformer } from '../transport/transport.transformer';
import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
import { inputLogger$, outputLogger$, errorLogger$ } from '../middlewares/messaging.eventLogger.middleware';
-import { TransportLayerToken, ServerEventsToken } from './messaging.server.tokens';
-import { AllServerEvents, ServerEvent } from './messaging.server.events';
type ProcessOperator = OperatorFunction<TransportMessage<any>, TransportMessage<any>>;
@@ -34,7 +29,7 @@ export interface MessagingListenerConfig {
}
export interface MessagingListener {
- listen: () => Promise<TransportLayerConnection>;
+ (connection: TransportLayerConnection): void;
}
const defaultOutput$: MsgOutputEffect = msg$ => msg$.pipe(map(m => m.event));
@@ -42,8 +37,6 @@ const defaultError$: MsgErrorEffect = msg$ => msg$;
const defaultEffect$: MsgEffect = msg$ => msg$;
export const messagingListener = createListener<MessagingListenerConfig, MessagingListener>(config => ask => {
- const transportLayer = useContext(TransportLayerToken)(ask);
-
const {
effects = [defaultEffect$],
middlewares = [],
@@ -52,13 +45,13 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
msgTransformer = jsonTransformer,
} = config ?? {};
- const handleConnection = (conn: TransportLayerConnection, serverEventsSubject: Subject<AllServerEvents>) => {
+ return connection => {
let effectsSub: Subscription;
const errorSubject = new Subject<Error>();
const combinedEffects = combineEffects(...effects);
const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
- const ctx = createEffectContext({ ask, client: conn });
+ const ctx = createEffectContext({ ask, client: connection });
const toUnhandledErrorEvent = ({ name, message }: Error): Observable<Event> =>
of({ type: 'UNHANDLED_ERROR', error: { name, message } });
@@ -105,7 +98,7 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
);
- const message$ = conn.message$.pipe(
+ const message$ = connection.message$.pipe(
map(decode),
processMiddlewares,
processEffects,
@@ -139,37 +132,13 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
}
const subscribeEffects = (input$: Observable<TransportMessage<any>>) => input$
- .pipe(takeUntil(conn.close$))
+ .pipe(takeUntil(connection.close$))
.subscribe(
- onSubscribeEffectsOutput(conn),
+ onSubscribeEffectsOutput(connection),
onSubscribeEffectsError(errorSubject),
onSubscribeEffectsClose,
);
- conn.close$
- .pipe(take(1))
- .subscribe(() => serverEventsSubject.next(ServerEvent.close()));
-
- conn.error$
- .pipe(takeUntil(conn.close$))
- .subscribe(error => serverEventsSubject.next(ServerEvent.error(error)));
-
effectsSub = subscribeEffects(message$);
};
-
- const listen = async () => {
- const { host, channel } = transportLayer.config;
- const serverEventsSubject = pipe(ask(ServerEventsToken), O.getOrElse(() => undefined as unknown as Subject<AllServerEvents>));
- const connection = await transportLayer.connect({ isConsumer: true });
-
- handleConnection(connection, serverEventsSubject);
-
- connection.status$
- .pipe(takeUntil(connection.close$))
- .subscribe(type => serverEventsSubject.next(ServerEvent.status(host,channel, type)));
-
- return connection;
- };
-
- return { listen };
});

MODIFY
packages/messaging/src/server/messaging.server.ts
packages/messaging/src/server/messaging.server.ts
import { Subject } from 'rxjs';
-import { takeWhile } from 'rxjs/operators';
+import { takeWhile, takeUntil, take } from 'rxjs/operators';
import { createContext, registerAll, bindTo, createEffectContext, lookup, combineEffects } from '@marblejs/core';
import { provideTransportLayer } from '../transport/transport.provider';
import { statusLogger$ } from '../middlewares/messaging.statusLogger.middleware';
import { CreateMicroserviceConfig, Microservice } from './messaging.server.interface';
import { TransportLayerToken, ServerEventsToken } from './messaging.server.tokens';
-import { AllServerEvents, isCloseEvent } from './messaging.server.events';
+import { AllServerEvents, isCloseEvent, ServerEvent } from './messaging.server.events';
export const createMicroservice = (config: CreateMicroserviceConfig): Microservice => {
const {
@@ -16,18 +16,41 @@ export const createMicroservice = (config: CreateMicroserviceConfig): Microservi
messagingListener,
} = config;
- const serverEventSubject = new Subject<AllServerEvents>();
+ const serverEventsSubject = new Subject<AllServerEvents>();
const transportLayer = provideTransportLayer(transport, options);
const boundTransportLayer = bindTo(TransportLayerToken)(() => transportLayer);
- const boundServerEvents = bindTo(ServerEventsToken)(() => serverEventSubject);
+ const boundServerEvents = bindTo(ServerEventsToken)(() => serverEventsSubject);
const context = registerAll([ boundTransportLayer, boundServerEvents, ...dependencies ])(createContext());
- const listenerWithContext = messagingListener(context);
+ const listener = messagingListener(context);
- const serverEvent$ = serverEventSubject.asObservable().pipe(takeWhile(e => !isCloseEvent(e)));
+ const serverEvent$ = serverEventsSubject.asObservable().pipe(takeWhile(e => !isCloseEvent(e)));
const ctx = createEffectContext({ ask: lookup(context), client: undefined });
const combinedEvents = event$ ? combineEffects(statusLogger$, event$) : statusLogger$;
combinedEvents(serverEvent$, ctx).subscribe();
- return listenerWithContext.listen;
+ const listen = async () => {
+ const { host, channel } = transportLayer.config;
+ const connection = await transportLayer.connect({ isConsumer: true });
+
+ listener(connection);
+
+ connection.status$
+ .pipe(takeUntil(connection.close$))
+ .subscribe(type => serverEventsSubject.next(ServerEvent.status(host, channel, type)));
+
+ connection.close$
+ .pipe(take(1))
+ .subscribe(() => serverEventsSubject.next(ServerEvent.close()));
+
+ connection.error$
+ .pipe(takeUntil(connection.close$))
+ .subscribe(error => serverEventsSubject.next(ServerEvent.error(error)));
+
+ return connection;
+ };
+
+ listen.context = context;
+
+ return listen;
};

refactor(messaging): process server related events in factory function instead of server listener