feat(messaging): integration with core Logger dependency
MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
@@ -8,7 +8,6 @@ const fib = (n: number): number =>
(n === 0 || n === 1) ? n : fib(n - 1) + fib(n - 2);
const fibonacci$: MsgEffect = event$ => {
- console.log('Bootstrapped effect: "fibonacci$"');
return event$.pipe(
matchEvent('FIB'),
@@ -23,7 +22,6 @@ const fibonacci$: MsgEffect = event$ => {
};
const buffer$: MsgEffect = event$ => {
- console.log('Bootstrapped effect: "buffer$"');
return event$.pipe(
matchEvent('BUFFER'),
@@ -33,12 +31,11 @@ const buffer$: MsgEffect = event$ => {
};
const timeout$: MsgEffect = event$ => {
- console.log('Bootstrapped effect: "timeout$"');
return event$.pipe(
matchEvent('TIMEOUT'),
mergeMap(event => of(event).pipe(
- delay(40 * 1000),
+ delay(180 * 1000),
mapTo({ ...event, type: 'TIMEOUT_RESULT' } as Event),
)),
);

MODIFY
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
import { tap } from 'rxjs/operators';
-import { useContext } from '@marblejs/core';
-import { provideLogger, LoggerLevel } from '../server/messaging.server.logger';
+import { useContext, LoggerToken, LoggerLevel } from '@marblejs/core';
import { MsgMiddlewareEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
import { TransportLayerToken } from '../server/messaging.server.tokens';
export const inputLogger$: MsgMiddlewareEffect = (event$, ctx) => {
- const logger = provideLogger(ctx.ask);
+ const logger = useContext(LoggerToken)(ctx.ask);
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
return event$.pipe(
tap(event => {
- const channel = transportLayer.config.channel;
+ const tag = transportLayer.config.channel;
const message = `${event.type}, id: ${event.metadata?.correlationId}`;
logger({
- level: LoggerLevel.INFO,
- tag: 'EVENT_IN',
+ tag,
message,
- channel,
- });
+ type: 'EVENT_IN',
+ level: LoggerLevel.INFO,
+ })();
}),
);
};
export const outputLogger$: MsgOutputEffect = (event$, ctx) => {
- const logger = provideLogger(ctx.ask);
+ const logger = useContext(LoggerToken)(ctx.ask);
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
return event$.pipe(
tap(event => {
const { error, metadata, type } = event;
- const channel = transportLayer.config.channel;
+ const tag = transportLayer.config.channel;
const message = error
? `"${error.name}", "${error.message}" for event ${type}`
: metadata?.replyTo
@@ -38,25 +37,30 @@ export const outputLogger$: MsgOutputEffect = (event$, ctx) => {
: `${event.type}, id: ${metadata?.correlationId ?? '-'}`;
logger({
- level: event.error ? LoggerLevel.ERROR : LoggerLevel.INFO,
- tag: 'EVENT_OUT',
+ tag,
message,
- channel,
- });
+ level: event.error ? LoggerLevel.ERROR : LoggerLevel.INFO,
+ type: 'EVENT_OUT',
+ })();
}),
);
};
export const errorLogger$: MsgMiddlewareEffect = (event$, ctx) => {
- const logger = provideLogger(ctx.ask);
+ const logger = useContext(LoggerToken)(ctx.ask);
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
+ const tag = transportLayer.config.channel;
return event$.pipe(
- tap(event => logger({
- tag: 'ERROR',
- message: `"${event.error?.name ?? '-'}: ${event.error?.message ?? '-' }" for event ${event.type}`,
+ tap(event => {
+ const message = `"${event.error?.name ?? '-'}: ${event.error?.message ?? '-' }" for event ${event.type}`;
+
+ return logger({
+ tag,
+ message,
+ type: 'ERROR',
level: LoggerLevel.ERROR,
- channel: transportLayer.config.channel,
- })),
+ })();
+ }),
);
};

MODIFY
packages/messaging/src/middlewares/messaging.statusLogger.middleware.ts
packages/messaging/src/middlewares/messaging.statusLogger.middleware.ts
-import { matchEvent, useContext, combineEffects } from '@marblejs/core';
+import { matchEvent, useContext, combineEffects, LoggerToken, LoggerLevel } from '@marblejs/core';
import { map, distinctUntilChanged, filter, tap } from 'rxjs/operators';
-import { provideLogger, LoggerLevel } from '../server/messaging.server.logger';
import { Transport } from '../transport/transport.interface';
import { AmqpConnectionStatus } from '../transport/strategies/amqp.strategy.interface';
import { RedisConnectionStatus } from '../transport/strategies/redis.strategy.interface';
@@ -20,7 +19,7 @@ const serverStatusMap = {
};
const connect$: MsgServerEffect = (event$, ctx) => {
- const logger = provideLogger(ctx.ask);
+ const logger = useContext(LoggerToken)(ctx.ask);
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
return event$.pipe(
@@ -29,16 +28,16 @@ const connect$: MsgServerEffect = (event$, ctx) => {
distinctUntilChanged((p, c) => p.type === c.type),
filter(({ type }) => type === serverStatusMap.connect[transportLayer.type]),
tap(({ host, channel }) => logger({
- tag: 'CONNECTED',
+ type: 'CONNECTED',
message: `Connected server to host: ${host}`,
level: LoggerLevel.INFO,
- channel,
- })),
+ tag: channel,
+ })()),
);
};
const disconnect$: MsgServerEffect = (event$, ctx) => {
- const logger = provideLogger(ctx.ask);
+ const logger = useContext(LoggerToken)(ctx.ask);
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
return event$.pipe(
@@ -47,27 +46,27 @@ const disconnect$: MsgServerEffect = (event$, ctx) => {
distinctUntilChanged((p, c) => p.type === c.type),
filter(({ type }) => type === serverStatusMap.disconnect[transportLayer.type]),
tap(({ host, channel }) => logger({
- tag: 'DISCONNECTED',
+ type: 'DISCONNECTED',
message: `Disconnected server from host: ${host}`,
level: LoggerLevel.ERROR,
- channel,
- }))
+ tag: channel,
+ })())
);
};
const error$: MsgServerEffect = (event$, ctx) => {
- const logger = provideLogger(ctx.ask);
+ const logger = useContext(LoggerToken)(ctx.ask);
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
return event$.pipe(
matchEvent(ServerEvent.error),
map(event => event.payload),
tap(({ error }) => logger({
- tag: 'ERROR',
+ type: 'ERROR',
message: `${error.name}, ${error.message}`,
level: LoggerLevel.ERROR,
- channel: transportLayer.config.channel,
- }))
+ tag: transportLayer.config.channel,
+ })())
);
};

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -4,10 +4,15 @@ import {
combineEffects,
createEffectContext,
createListener,
+ EffectContext,
+ useContext,
+ LoggerToken,
+ LoggerTag,
+ LoggerLevel,
} from '@marblejs/core';
import { pipe } from 'fp-ts/lib/pipeable';
-import { of, Observable } from 'rxjs';
-import { map, publish, catchError, takeUntil } from 'rxjs/operators';
+import { of, Observable, Subject } from 'rxjs';
+import { map, catchError, takeUntil } from 'rxjs/operators';
import {
TransportMessage,
TransportMessageTransformer,
@@ -47,6 +52,7 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
msgTransformer = jsonTransformer,
} = config ?? {};
+ const logger = useContext(LoggerToken)(ask);
const combinedEffects = combineEffects(...effects);
const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
@@ -73,39 +79,65 @@ 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 eventSubject = new Subject<Event>();
const ctx = createEffectContext({ ask, client: connection });
- const stream = pipe(
- connection.message$.pipe(map(decode)),
+ const incomingEvent$ = pipe(
+ connection.message$,
+ e$ => e$.pipe(map(decode)),
e$ => combinedMiddlewares(e$, ctx),
+ e$ => e$.pipe(catchError(processError$(ctx))),
+ );
+
+ const outgoingEvent$ = pipe(
+ eventSubject.asObservable(),
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)),
- );
- }),
- )
+ e$ => e$.pipe(catchError(processError$(ctx))),
);
- const subscribe = (event$: Observable<Event<unknown, any, string>>) =>
+ const subscribeIncomingEvent = (event$: Observable<Event<unknown, any, string>>) =>
event$
.pipe(takeUntil(connection.close$))
.subscribe(
- send(connection),
+ event => eventSubject.next(event),
+ (error: Error) => {
+ const type = 'ServerListener';
+ const message = `Unexpected error for IncomingEvent stream: "${error.name}", "${error.message}"`;
+ logger({ tag: LoggerTag.MESSAGING, type, message, level: LoggerLevel.ERROR })();
+ subscribeIncomingEvent(event$);
+ },
() => {
- console.error('Unexpected stream error!'); // @TODO: handle unexpected error
- subscribe(event$);
+ subscribeIncomingEvent(event$);
+ },
+ );
+
+ const subscribeOutgoingEvent = (event$: Observable<Event<unknown, any, string>>) =>
+ event$
+ .pipe(takeUntil(connection.close$))
+ .subscribe(
+ event => send(connection)(event),
+ (error: Error) => {
+ const type = 'ServerListener';
+ const message = `Unexpected error OutgoingEvent stream: "${error.name}", "${error.message}"`;
+ logger({ tag: LoggerTag.MESSAGING, type, message, level: LoggerLevel.ERROR })();
+ subscribeOutgoingEvent(event$);
},
() => {
- subscribe(event$);
+ subscribeOutgoingEvent(event$);
},
);
- subscribe(stream);
+ subscribeIncomingEvent(incomingEvent$);
+ subscribeOutgoingEvent(outgoingEvent$);
};
});

DELETE
packages/messaging/src/server/messaging.server.logger.ts
-import chalk from 'chalk';
-import { pipe } from 'fp-ts/lib/pipeable';
-import { getOrElse } from 'fp-ts/lib/Option';
-import { ContextProvider } from '@marblejs/core';
-import { LoggerToken } from './messaging.server.tokens';
-
-const IS_TEST_ENV = process.env.NODE_ENV === 'test';
-
-export type Logger = (opts: LoggerOptions) => any;
-export enum LoggerLevel { INFO, WARN, ERROR }
-export type LoggerOptions = {
- channel: string;
- message: string;
- tag: string;
- level?: LoggerLevel;
-};
-
-export const provideLogger = (ask: ContextProvider): Logger => pipe(
- ask(LoggerToken),
- getOrElse(() => !IS_TEST_ENV ? defaultLogger : () => undefined),
-);
-
-const defaultLogger: Logger = ({ channel, message, tag, level = LoggerLevel.INFO }) => {
- const now = new Date().toISOString();
- const colorizeText = {
- [LoggerLevel.ERROR]: chalk.red,
- [LoggerLevel.INFO]: chalk.green,
- [LoggerLevel.WARN]: chalk.yellow,
- }[level];
- return console.log(now.padEnd(30) + channel.padEnd(15) + colorizeText(tag.padEnd(15)) + message);
-};

MODIFY
packages/messaging/src/server/messaging.server.tokens.ts
packages/messaging/src/server/messaging.server.tokens.ts
@@ -2,8 +2,6 @@ import { createContextToken } from '@marblejs/core';
import { Subject } from 'rxjs';
import { TransportLayer } from '../transport/transport.interface';
import { AllServerEvents } from './messaging.server.events';
-import { Logger } from './messaging.server.logger';
-export const TransportLayerToken = createContextToken<TransportLayer>();
-export const ServerEventsToken = createContextToken<Subject<AllServerEvents>>();
-export const LoggerToken = createContextToken<Logger>();
+export const TransportLayerToken = createContextToken<TransportLayer>('TransportLayerToken');
+export const ServerEventsToken = createContextToken<Subject<AllServerEvents>>('ServerEventsToken');

MODIFY
packages/messaging/src/server/messaging.server.ts
packages/messaging/src/server/messaging.server.ts
import { Subject } from 'rxjs';
import { takeWhile, takeUntil, take } from 'rxjs/operators';
import { flow } from 'fp-ts/lib/function';
-import { createContext, registerAll, bindTo, createEffectContext, lookup, combineEffects, resolve, ServerIO } from '@marblejs/core';
+import {
+ bindTo,
+ createContext,
+ createEffectContext,
+ combineEffects,
+ registerAll,
+ resolve,
+ ServerIO,
+ mockLogger,
+ LoggerToken,
+ lookup,
+ logContext,
+ logger,
+ LoggerTag,
+} from '@marblejs/core';
+import { isTestEnv } from '@marblejs/core/dist/+internal/utils';
import { provideTransportLayer } from '../transport/transport.provider';
import { statusLogger$ } from '../middlewares/messaging.statusLogger.middleware';
import { TransportLayerConnection } from '../transport/transport.interface';
@@ -20,15 +35,18 @@ export const createMicroservice = async (config: CreateMicroserviceConfig) => {
const serverEventsSubject = new Subject<AllServerEvents>();
const transportLayer = provideTransportLayer(transport, options);
+ const boundLogger = bindTo(LoggerToken)(isTestEnv() ? mockLogger : logger);
const boundTransportLayer = bindTo(TransportLayerToken)(() => transportLayer);
const boundServerEvents = bindTo(ServerEventsToken)(() => serverEventsSubject);
const context = await flow(
registerAll([
+ boundLogger,
boundTransportLayer,
boundServerEvents,
...dependencies,
]),
+ logContext(LoggerTag.MESSAGING),
resolve,
)(createContext());

feat(messaging): integration with core Logger dependency