feat(messaging): use Context for providing custom logger instance
MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
import { matchEvent, use } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
-import {
- createMicroservice,
- messagingListener,
- status$,
- output$,
- input$,
- Transport,
- MsgEffect,
- MsgServerEffect,
- ServerEvent,
-} from '@marblejs/messaging';
-import { merge } from 'rxjs';
-import { map, tap } from 'rxjs/operators';
+import { createMicroservice, messagingListener, Transport, MsgEffect } from '@marblejs/messaging';
+import { map } from 'rxjs/operators';
-const fibonacci = (n: number): number =>
- n === 0 || n === 1
- ? n
- : fibonacci(n - 1) + fibonacci(n - 2);
+const fib = (n: number): number =>
+ (n === 0 || n === 1) ? n : fib(n - 1) + fib(n - 2);
const fibonacci$: MsgEffect = event$ =>
event$.pipe(
matchEvent('FIB'),
use(eventValidator$(t.number)),
- map(event => fibonacci(event.payload)),
+ map(event => fib(event.payload)),
map(payload => ({ type: 'FIB_RESULT', payload })),
);
-const error$: MsgServerEffect = event$ =>
- event$.pipe(
- matchEvent(ServerEvent.error),
- map(event => event.payload),
- tap(({ error }) => console.error(error)),
- );
-
export const microservice = createMicroservice({
transport: Transport.AMQP,
options: {
@@ -43,14 +23,7 @@ export const microservice = createMicroservice({
},
messagingListener: messagingListener({
effects: [fibonacci$],
- middlewares: [input$()],
- output$: output$(),
}),
- dependencies: [],
- event$: (...args) => merge(
- error$(...args),
- status$()(...args),
- ),
});
const bootstrap = async () => {

MODIFY
packages/core/src/context/context.hook.ts
packages/core/src/context/context.hook.ts
import { pipe } from 'fp-ts/lib/pipeable';
+import { identity } from 'fp-ts/lib/function';
import * as O from 'fp-ts/lib/Option';
import { ContextToken } from './context.token.factory';
import { ContextProvider } from './context.factory';
@@ -20,6 +21,6 @@ export const useContext = <T>(token: ContextToken<T>) => (ask: ContextProvider)
throw error;
},
- dep => dep,
+ identity,
),
);

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
@@ -7,5 +7,4 @@ export * from './server/messaging.server';
export * from './server/messaging.server.interface';
export * from './server/messaging.server.tokens';
export * from './server/messaging.server.events';
-export { status$, input$, output$ } from './server/messaging.server.logger';
export { messagingListener } from './server/messaging.server.listener';

ADD
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
+import chalk from 'chalk';
+import { tap, map } from 'rxjs/operators';
+import { useContext } from '@marblejs/core';
+import { provideLogger, LoggerLevel } from '../server/messaging.server.logger';
+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 transportLayer = useContext(TransportLayerToken)(ctx.ask);
+
+ return event$.pipe(
+ tap(event => {
+ const eventType = chalk.yellow(event.type);
+ const eventCorrelationId = chalk.magenta(event.raw.correlationId);
+ const channel = transportLayer.config.channel;
+ const message = `${eventType}, id: ${eventCorrelationId}`;
+
+ logger({
+ level: LoggerLevel.INFO,
+ tag: 'EVENT_IN',
+ message,
+ channel,
+ });
+ }),
+ );
+};
+
+export const outputLogger$: MsgOutputEffect = (event$, ctx) => {
+ const logger = provideLogger(ctx.ask);
+ const transportLayer = useContext(TransportLayerToken)(ctx.ask);
+
+ return event$.pipe(
+ tap(({ event, initiator }) => {
+ const eventType = chalk.yellow(event.type);
+ const eventCorrelationId = initiator.correlationId && chalk.magenta(initiator.correlationId);
+ const eventReplyTo = initiator.replyTo && chalk.yellow(initiator.replyTo);
+ const channel = transportLayer.config.channel;
+ const message = eventReplyTo
+ ? `${eventType}, id: ${eventCorrelationId} and sent to ${eventReplyTo}`
+ : `${eventType}, id: ${eventCorrelationId}`;
+
+ logger({
+ level: LoggerLevel.INFO,
+ tag: 'EVENT_OUT',
+ message,
+ channel,
+ });
+ }),
+ map(({ event }) => event),
+ );
+};

ADD
packages/messaging/src/middlewares/messaging.statusLogger.middleware.ts
+import { matchEvent, useContext, combineEffects } 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';
+import { ServerEvent } from '../server/messaging.server.events';
+import chalk from 'chalk';
+import { MsgServerEffect } from '../effects/messaging.effects.interface';
+import { TransportLayerToken } from '../server/messaging.server.tokens';
+
+const serverStatusMap = {
+ connect: {
+ [Transport.AMQP]: AmqpConnectionStatus.CONNECTED,
+ [Transport.REDIS]: RedisConnectionStatus.CONNECT,
+ },
+ disconnect: {
+ [Transport.AMQP]: AmqpConnectionStatus.CONNECTION_LOST,
+ [Transport.REDIS]: RedisConnectionStatus.RECONNECTING,
+ },
+};
+
+const connect$: MsgServerEffect = (event$, ctx) => {
+ const logger = provideLogger(ctx.ask);
+ const transportLayer = useContext(TransportLayerToken)(ctx.ask);
+
+ return event$.pipe(
+ matchEvent(ServerEvent.status),
+ map(event => event.payload),
+ distinctUntilChanged((p, c) => p.type === c.type),
+ filter(({ type }) => type === serverStatusMap.connect[transportLayer.type]),
+ tap(({ host, channel }) => logger({
+ tag: 'CONNECTED',
+ message: `Connected server to host: ${chalk.yellow(host)}`,
+ level: LoggerLevel.INFO,
+ channel,
+ })),
+ );
+};
+
+const disconnect$: MsgServerEffect = (event$, ctx) => {
+ const logger = provideLogger(ctx.ask);
+ const transportLayer = useContext(TransportLayerToken)(ctx.ask);
+
+ return event$.pipe(
+ matchEvent(ServerEvent.status),
+ map(event => event.payload),
+ distinctUntilChanged((p, c) => p.type === c.type),
+ filter(({ type }) => type === serverStatusMap.disconnect[transportLayer.type]),
+ tap(({ host, channel }) => logger({
+ tag: 'DISCONNECTED',
+ message: `Disconnected server from host: ${chalk.yellow(host)}`,
+ level: LoggerLevel.ERROR,
+ channel,
+ }))
+ );
+};
+
+const error$: MsgServerEffect = (event$, ctx) => {
+ const logger = provideLogger(ctx.ask);
+ const transportLayer = useContext(TransportLayerToken)(ctx.ask);
+
+ return event$.pipe(
+ matchEvent(ServerEvent.error),
+ map(event => event.payload),
+ tap(({ error }) => logger({
+ tag: 'ERROR',
+ message: `${error.name}, ${error.message}`,
+ level: LoggerLevel.ERROR,
+ channel: transportLayer.config.channel,
+ }))
+ );
+};
+
+export const statusLogger$ = combineEffects(
+ connect$,
+ disconnect$,
+ error$,
+);

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -23,6 +23,7 @@ import { jsonTransformer } from '../transport/transport.transformer';
import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
import { TransportLayerToken, ServerEventsToken } from './messaging.server.tokens';
import { AllServerEvents, ServerEvent } from './messaging.server.events';
+import { inputLogger$, outputLogger$ } from '../middlewares/messaging.eventLogger.middleware';
export interface MessagingListenerConfig {
effects?: MsgEffect<any, any>[];
@@ -54,7 +55,8 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const errorSubject = new Subject<Error>();
const combinedEffects = combineEffects(...effects);
- const combinedMiddlewares = combineMiddlewares(...middlewares);
+ const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
+ const combinedOutput = combineEffects(outputLogger$, output$);
const ctx = createEffectContext({ ask, client: conn });
const decode = (msg: TransportMessage<Buffer>): TransportMessage<Event> => ({
@@ -75,7 +77,7 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
publish(msg$ => zip(
- output$(msg$.pipe(map(m => ({ event: m.data, initiator: m }))), ctx),
+ combinedOutput(msg$.pipe(map(m => ({ event: m.data, initiator: m }))), ctx),
msg$,
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),

MODIFY
packages/messaging/src/server/messaging.server.logger.ts
packages/messaging/src/server/messaging.server.logger.ts
import chalk from 'chalk';
-import { matchEvent, useContext, combineEffects } from '@marblejs/core';
-import { distinctUntilChanged, tap, map, filter } from 'rxjs/operators';
-import { Transport} from '../transport/transport.interface';
-import { RedisConnectionStatus } from '../transport/strategies/redis.strategy.interface';
-import { AmqpConnectionStatus } from '../transport/strategies/amqp.strategy.interface';
-import { TransportLayerToken } from './messaging.server.tokens';
-import { ServerEvent } from './messaging.server.events';
-import { MsgServerEffect, MsgMiddlewareEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
-
-type Logger = (msg: string) => any;
-type LogFn = (logger: Logger) => (type: string) => (msg: string) => any;
-
-const BRANDING = '@marblejs/messaging';
-
-const provideLogger = (customLogger?: Logger) =>
- customLogger || console.log;
-
-const log: LogFn = logger => type => msg => {
- const now = new Date().toISOString();
- return logger(`${now}\t${BRANDING}\t${type}\t${msg}`);
-};
-
-const logInfo: LogFn = logger => type => msg => log(logger)(chalk.green(type))(msg);
-
-// eslint-disable-next-line @typescript-eslint/no-unused-vars
-const logWarn: LogFn = logger => type => msg => log(logger)(chalk.yellow(type))(msg);
-
-const logError: LogFn = logger => type => msg => log(logger)(chalk.red(type))(msg);
-
-// log server connected/disconnected status
-export const status$ = (customLogger?: Logger) => {
- const logger = provideLogger(customLogger);
-
- const serverStatusMap = {
- connect: {
- [Transport.AMQP]: AmqpConnectionStatus.CONNECTED,
- [Transport.REDIS]: RedisConnectionStatus.CONNECT,
- },
- disconnect: {
- [Transport.AMQP]: AmqpConnectionStatus.CONNECTION_LOST,
- [Transport.REDIS]: RedisConnectionStatus.RECONNECTING,
- },
- }
-
- const connect$ = (logger: Logger): MsgServerEffect => (event$, ctx) =>
- event$.pipe(
- matchEvent(ServerEvent.status),
- map(event => event.payload),
- distinctUntilChanged((p, c) => p.type === c.type),
- filter(({ type }) => {
- const transportLayer = useContext(TransportLayerToken)(ctx.ask);
- return type === serverStatusMap.connect[transportLayer.type];
- }),
- tap(({ host, channel }) =>
- logInfo(logger)('CONNECTED')(`${chalk.yellow(host)} on channel ${chalk.yellow(channel)}`)),
+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';
+
+export enum LoggerLevel { INFO, WARN, ERROR }
+export type LoggerOptions = { channel: string; message: string; tag: string; level?: LoggerLevel };
+export type Logger = (opts: LoggerOptions) => any;
+
+export const provideLogger = (ask: ContextProvider): Logger => pipe(
+ ask(LoggerToken),
+ getOrElse(() => log),
);
- const disconnect$ = (logger: Logger): MsgServerEffect => (event$, ctx) =>
- event$.pipe(
- matchEvent(ServerEvent.status),
- map(event => event.payload),
- distinctUntilChanged((p, c) => p.type === c.type),
- filter(({ type }) => {
- const transportLayer = useContext(TransportLayerToken)(ctx.ask);
- return type === serverStatusMap.disconnect[transportLayer.type];
- }),
- tap(({ host, channel }) =>
- logError(logger)('DISCONNECTED')(`${chalk.yellow(host)} on channel ${chalk.yellow(channel)}`)),
- );
-
- return combineEffects(
- connect$(logger),
- disconnect$(logger),
- );
-};
-
-// log listener incoming messages
-export const input$ = (customLogger?: Logger): MsgMiddlewareEffect => {
- const logger = provideLogger(customLogger);
-
- return event$ =>
- event$.pipe(
- tap(event => {
- const eventType = chalk.yellow(event.type);
- const eventCorrelationId = chalk.magenta(event.raw.correlationId);
- return logInfo(logger)('EVENT_IN')(`${eventType}, id: ${eventCorrelationId}`);
- }),
- );
-};
-
-// log listener server outgoing messages
-export const output$ = (customLogger?: Logger): MsgOutputEffect => {
- const logger = provideLogger(customLogger);
-
- return event$ =>
- event$.pipe(
- tap(({ event, initiator }) => {
- const eventType = chalk.yellow(event.type);
- const eventCorrelationId = initiator.correlationId && chalk.magenta(initiator.correlationId);
- const eventReplyTo = initiator.replyTo && chalk.yellow(initiator.replyTo);
-
- return eventReplyTo
- ? logInfo(logger)('EVENT_OUT')(`${eventType}, id: ${eventCorrelationId} and sent to ${eventReplyTo}`)
- : logInfo(logger)('EVENT_OUT')(`${eventType}, id: ${eventCorrelationId}`);
- }),
- map(({ event }) => event),
- );
+const log: Logger = ({ channel, message, tag, level = LoggerLevel.INFO }) => {
+ const now = new Date().toISOString();
+ const colorize = {
+ [LoggerLevel.ERROR]: chalk.red,
+ [LoggerLevel.INFO]: chalk.green,
+ [LoggerLevel.WARN]: chalk.yellow,
+ }[level];
+ return console.log(`${now}\t${channel}\t${colorize(tag)}\t${message}`);
};

MODIFY
packages/messaging/src/server/messaging.server.tokens.ts
packages/messaging/src/server/messaging.server.tokens.ts
@@ -2,6 +2,8 @@ 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>();

MODIFY
packages/messaging/src/server/messaging.server.ts
packages/messaging/src/server/messaging.server.ts
-import { createContext, registerAll, bindTo, createEffectContext, lookup } from '@marblejs/core';
+import { createContext, registerAll, bindTo, createEffectContext, lookup, combineEffects } from '@marblejs/core';
import { Subject } from 'rxjs';
import { takeWhile } from 'rxjs/operators';
import { CreateMicroserviceConfig } from './messaging.server.interface';
import { provideTransportLayer } from '../transport/transport.provider';
import { TransportLayerToken, ServerEventsToken } from './messaging.server.tokens';
import { AllServerEvents, isCloseEvent } from './messaging.server.events';
+import { statusLogger$ } from '../middlewares/messaging.statusLogger.middleware';
export const createMicroservice = (config: CreateMicroserviceConfig) => {
const {
@@ -22,11 +23,11 @@ export const createMicroservice = (config: CreateMicroserviceConfig) => {
const context = registerAll([ boundTransportLayer, boundServerEvents, ...dependencies ])(createContext());
const listenerWithContext = messagingListener(context);
- if (event$) {
const serverEvent$ = serverEventSubject.asObservable().pipe(takeWhile(e => !isCloseEvent(e)));
const ctx = createEffectContext({ ask: lookup(context), client: undefined });
- event$(serverEvent$, ctx).subscribe();
- }
+ const combinedEvents = event$ ? combineEffects(statusLogger$, event$) : statusLogger$;
+
+ combinedEvents(serverEvent$, ctx).subscribe();
return listenerWithContext.listen;
};

feat(messaging): use Context for providing custom logger instance