feat(messaging): default input/output/status loggers
MODIFY
packages/messaging/package.json
packages/messaging/package.json
"rxjs": "^6.5.3"
},
"dependencies": {
+ "chalk": "~2.4.1"
},
"devDependencies": {
"@marblejs/core": "^2.3.1",

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
@@ -7,4 +7,5 @@ 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';

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -59,7 +59,7 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const decode = (msg: TransportMessage<Buffer>): TransportMessage<Event> => ({
...msg,
- data: { ...msgTransformer.decode(msg.data), raw: msg.raw },
+ data: { ...msgTransformer.decode(msg.data), raw: msg },
});
const message$ = conn.message$.pipe(

ADD
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)}`)),
+ );
+
+ 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),
+ );
+};

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
@@ -2,7 +2,7 @@ import { Subject, fromEvent, merge, from } from 'rxjs';
import { map, filter, take, mapTo, first, mergeMap } from 'rxjs/operators';
import { Channel, ConsumeMessage } from 'amqplib';
import { AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager';
-import { TransportLayer, TransportMessage, TransportLayerConnection } from '../transport.interface';
+import { TransportLayer, TransportMessage, TransportLayerConnection, Transport } from '../transport.interface';
import { AmqpStrategyOptions, AmqpConnectionStatus } from './amqp.strategy.interface';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
@@ -106,15 +106,15 @@ class AmqpStrategyConnection implements TransportLayerConnection {
return true;
};
- ackMessage = (msg: ConsumeMessage | undefined) => {
+ ackMessage = (msg: TransportMessage | undefined) => {
if (msg) {
- this.channelWrapper.ack(msg);
+ this.channelWrapper.ack(msg.raw);
}
}
- nackMessage = (msg: ConsumeMessage | undefined, resend = true) => {
+ nackMessage = (msg: TransportMessage | undefined, resend = true) => {
if (msg) {
- this.channelWrapper.nack(msg, false , resend);
+ this.channelWrapper.nack(msg.raw, false , resend);
}
}
@@ -130,6 +130,10 @@ class AmqpStrategyConnection implements TransportLayerConnection {
class AmqpStrategy implements TransportLayer {
constructor(private options: AmqpStrategyOptions) {}
+ get type() {
+ return Transport.AMQP;
+ }
+
get config() {
return ({
host: this.options.host,

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.ts
packages/messaging/src/transport/strategies/redis.strategy.ts
@@ -2,7 +2,7 @@ import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { Subject, fromEvent, merge } from 'rxjs';
import { map, mapTo, take } from 'rxjs/operators';
import { RedisClient, ClientOpts } from 'redis';
-import { TransportLayer, TransportLayerConnection, TransportMessage } from '../transport.interface';
+import { TransportLayer, TransportLayerConnection, TransportMessage, Transport } from '../transport.interface';
import { RedisStrategyOptions, RedisConnectionStatus } from './redis.strategy.interface';
import { throwUnsupportedError } from '../transport.error';
import * as RedisHelper from './redis.strategy.helper';
@@ -109,6 +109,10 @@ class RedisStrategyConnection implements TransportLayerConnection {
class RedisStrategy implements TransportLayer {
constructor(private options: RedisStrategyOptions) {}
+ get type() {
+ return Transport.REDIS;
+ }
+
get config() {
return ({
host: this.options.host,

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -21,6 +21,7 @@ export type TransportStrategy =
export interface TransportLayer {
connect: (opts: { isConsumer: boolean }) => Promise<TransportLayerConnection>;
+ type: Transport;
config: TransportLayerConfig;
}
@@ -32,8 +33,8 @@ export interface TransportLayerConfig {
export interface TransportLayerConnection {
sendMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<TransportMessage<Buffer>>;
emitMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<boolean>;
- ackMessage: (message: any | undefined) => void;
- nackMessage: (message: any | undefined, resend?: boolean) => void;
+ ackMessage: (message: TransportMessage | undefined) => void;
+ nackMessage: (message: TransportMessage | undefined, resend?: boolean) => void;
close: () => Promise<any>;
getChannel(): string;
message$: Observable<TransportMessage<Buffer>>;

feat(messaging): default input/output/status loggers