fix(messaging): mismatched events
MODIFY
packages/messaging/src/effects/messaging.effects.interface.ts
packages/messaging/src/effects/messaging.effects.interface.ts
@@ -11,7 +11,7 @@ export interface MsgMiddlewareEffect<
export interface MsgErrorEffect<
I = Event,
Err extends Error = Error,
-> extends MsgEffect<{ event?: I; error: Err }, any, MsgClient> {}
+> extends MsgEffect<{ event: I; error: Err }, Event, MsgClient> {}
export interface MsgEffect<
I = Event,

MODIFY
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
@@ -60,5 +60,6 @@ export const errorLogger$: MsgErrorEffect = (event$, ctx) => {
level: LoggerLevel.ERROR,
channel: transportLayer.config.channel,
})),
+ map(({ event }) => event),
);
};

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -3,12 +3,10 @@ import {
combineMiddlewares,
combineEffects,
createEffectContext,
- EventError,
createListener,
} from '@marblejs/core';
-import { flow } from 'fp-ts/lib/function';
-import { Observable, Subscription, Subject, of, zip, OperatorFunction} from 'rxjs';
-import { map, publish, takeUntil, catchError, mergeMapTo } from 'rxjs/operators';
+import { Subject, of} from 'rxjs';
+import { map, publish, takeUntil, catchError, mergeMapTo, mergeMap, tap } from 'rxjs/operators';
import {
TransportMessage,
TransportMessageTransformer,
@@ -18,8 +16,6 @@ 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';
-type ProcessOperator = OperatorFunction<TransportMessage<any>, TransportMessage<any>>;
-
export interface MessagingListenerConfig {
effects?: MsgEffect<any, any>[];
middlewares?: MsgMiddlewareEffect<any, any>[];
@@ -32,9 +28,18 @@ export interface MessagingListener {
(connection: TransportLayerConnection): void;
}
-const defaultOutput$: MsgOutputEffect = msg$ => msg$.pipe(map(m => m.event));
-const defaultError$: MsgErrorEffect = msg$ => msg$;
-const defaultEffect$: MsgEffect = msg$ => msg$;
+const defaultEffect$: MsgEffect = msg$ =>
+ msg$;
+
+const defaultOutput$: MsgOutputEffect = msg$ =>
+ msg$.pipe(map(m => m.event));
+
+const defaultError$: MsgErrorEffect = msg$ =>
+ msg$.pipe(map(({ event, error }) => ({
+ type: event.type,
+ payload: event.payload,
+ error: { name: error.name, message: error.message },
+ } as Event)));
export const messagingListener = createListener<MessagingListenerConfig, MessagingListener>(config => ask => {
const {
@@ -46,99 +51,51 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
} = config ?? {};
return connection => {
- let effectsSub: Subscription;
-
- const errorSubject = new Subject<Error>();
const combinedEffects = combineEffects(...effects);
const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
const ctx = createEffectContext({ ask, client: connection });
- const toUnhandledErrorEvent = ({ name, message }: Error): Observable<Event> =>
- of({ type: 'UNHANDLED_ERROR', error: { name, message } });
-
const decode = (msg: TransportMessage<Buffer>): TransportMessage<Event> => ({
...msg,
data: { ...msgTransformer.decode(msg.data), raw: msg },
});
- const processMiddlewares: ProcessOperator = flow(
- publish(msg$ => zip(
- combinedMiddlewares(msg$.pipe(map(m => m.data)), ctx).pipe(
- catchError(toUnhandledErrorEvent),
- ),
- msg$,
- )),
- map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- );
-
- const processEffects: ProcessOperator = flow(
- publish(msg$ => zip(
- combinedEffects(msg$.pipe(map(m => m.data)), ctx).pipe(
- catchError(toUnhandledErrorEvent),
- ),
- msg$,
- )),
- map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- );
-
- const processOutput: ProcessOperator = flow(
- publish(msg$ => zip(
- outputLogger$(msg$.pipe(map(m => ({ event: m.data, initiator: m }))), ctx).pipe(
- catchError(toUnhandledErrorEvent),
- ),
- msg$,
- )),
- map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- publish(msg$ => zip(
- output$(msg$.pipe(map(m => ({ event: m.data, initiator: m }))), ctx).pipe(
- catchError(toUnhandledErrorEvent),
- ),
- msg$,
- )),
- map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- );
+ const subject = new Subject<TransportMessage<Buffer>>();
- const message$ = connection.message$.pipe(
+ subject
+ .pipe(
+ takeUntil(connection.close$),
map(decode),
- processMiddlewares,
- processEffects,
- processOutput,
- catchError((error: EventError) => {
- const e$ = of({ event: error.event, error });
+ mergeMap(initiator => of(initiator.data).pipe(
+ publish(e$ => combinedMiddlewares(e$, ctx)),
+ publish(e$ => combinedEffects(e$, ctx)),
+ publish(e$ => outputLogger$(e$.pipe(map(event => ({ initiator, event }))), ctx)),
+ publish(e$ => output$(e$.pipe(map(event => ({ initiator, event }))), ctx)),
+ catchError(error => {
+ const e$ = of({ event: initiator.data, error });
return errorLogger$(e$, ctx).pipe(
mergeMapTo(error$(e$, ctx)),
- map(data => ({ data } as TransportMessage<any>)),
);
}),
- );
-
- const onSubscribeEffectsOutput = (conn: TransportLayerConnection) => (msg: TransportMessage<any>) => {
- if (msg.replyTo) {
- conn.emitMessage(msg.replyTo, {
- data: msgTransformer.encode(msg.data),
- correlationId: msg.correlationId,
- raw: msg.raw,
+ tap(async event => {
+ if (initiator.replyTo) {
+ const { replyTo, correlationId, raw } = initiator;
+ const { type, payload, error } = event;
+ return connection.emitMessage(replyTo, {
+ data: msgTransformer.encode({ type, payload, error }),
+ correlationId,
+ raw,
});
}
- }
-
- const onSubscribeEffectsError = (errorSubject: Subject<Error>) => (error: Error) => {
- errorSubject.next(error);
- if (effectsSub.closed) { effectsSub = subscribeEffects(message$); }
- }
- const onSubscribeEffectsClose = () => {
- effectsSub = subscribeEffects(message$);
- }
+ return true;
+ }),
+ )),
+ )
+ .subscribe();
- const subscribeEffects = (input$: Observable<TransportMessage<any>>) => input$
+ connection.message$
.pipe(takeUntil(connection.close$))
- .subscribe(
- onSubscribeEffectsOutput(connection),
- onSubscribeEffectsError(errorSubject),
- onSubscribeEffectsClose,
- );
-
- effectsSub = subscribeEffects(message$);
+ .subscribe(msg => subject.next(msg));
};
});

MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
@@ -3,7 +3,7 @@ import { eventValidator$, t } from '@marblejs/middleware-io';
import { Subject, throwError } from 'rxjs';
import { map, tap, mergeMapTo, delay, bufferCount } from 'rxjs/operators';
import { Transport } from '../../transport/transport.interface';
-import { MsgEffect, MsgOutputEffect } from '../../effects/messaging.effects.interface';
+import { MsgEffect, MsgErrorEffect, MsgOutputEffect } from '../../effects/messaging.effects.interface';
import { AmqpStrategyOptions } from '../../transport/strategies/amqp.strategy.interface';
import { runClient, runServer, createMessage } from '../../util/messaging.test.util';
@@ -14,7 +14,7 @@ const createOptions = (config: { expectAck?: boolean; queue?: string } = {}): Am
queueOptions: { durable: false },
});
-describe('messagingServer::AMQP', () => {
+describe.only('messagingServer::AMQP', () => {
test('handles RPC event', async () => {
const rpc$: MsgEffect = event$ =>
event$.pipe(
@@ -97,20 +97,19 @@ describe('messagingServer::AMQP', () => {
await client.emitMessage(options.queue, message);
});
- test('reacts to thrown Error and doesn\'t crash internal messages stream', async done => {
+ test.only('reacts to thrown Error and doesn\'t crash internal messages stream', async done => {
const event1: Event = { type: 'EVENT_TEST_1' };
const event2: Event = { type: 'EVENT_TEST_2' };
const error = new Error('test_error');
- const expectedError = { error: { message: error.message, name: error.name }, type: 'UNHANDLED_ERROR' };
- const outputSubject = new Subject<Event>();
+ const outputSubject = new Subject<[Event, Error | undefined]>();
outputSubject
.pipe(bufferCount(2))
.subscribe(data => {
- expect(data[0].type).toEqual(expectedError.type);
- expect(data[0].error).toEqual(expectedError.error);
- expect(data[1].type).toEqual(event2.type);
- expect(data[1].payload).toEqual(event2.payload);
+ expect(data[0][0].type).toEqual(event1.type);
+ expect(data[0][1]).toEqual(error);
+ expect(data[1][0].type).toEqual(event2.type);
+ expect(data[1][1]).toBeUndefined();
setTimeout(() => server.close().then(done), 1000);
});
@@ -125,15 +124,21 @@ describe('messagingServer::AMQP', () => {
matchEvent(event2.type),
);
+ const error$: MsgErrorEffect = event$ =>
+ event$.pipe(
+ tap(({ event, error }) => outputSubject.next([event, error])),
+ map(({ event }) => event),
+ );
+
const output$: MsgOutputEffect = event$ =>
event$.pipe(
- tap(({ event }) => outputSubject.next(event)),
+ tap(({ event }) => outputSubject.next([event, undefined])),
map(({ event }) => event),
);
const options = createOptions({ expectAck: false });
const client = await runClient(Transport.AMQP, options);
- const server = await runServer(Transport.AMQP, options)(combineEffects(event1$, event2$), output$);
+ const server = await runServer(Transport.AMQP, options)(combineEffects(event1$, event2$), output$, error$);
await client.emitMessage(options.queue, createMessage(event1));
await client.emitMessage(options.queue, createMessage(event2));

MODIFY
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
@@ -2,7 +2,7 @@ import { matchEvent, use } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { map, tap } from 'rxjs/operators';
import { Transport } from '../../transport/transport.interface';
-import { MsgEffect, MsgOutputEffect } from '../../effects/messaging.effects.interface';
+import { MsgEffect, MsgErrorEffect } from '../../effects/messaging.effects.interface';
import { RedisStrategyOptions } from '../../transport/strategies/redis.strategy.interface';
import { runClient, runServer, createMessage, wait } from '../../util/messaging.test.util';
@@ -74,7 +74,6 @@ describe('messagingServer::Redis', () => {
});
test('throws an UnsupportedError for unsupported "ackMessage/nackMessage"', async done => {
- const expectedEventType = 'UNHANDLED_ERROR';
const expectedEventError = {
name: 'UnsupportedError',
message: 'Unsupported operation. Method \"ackMessage\" is unsupported for Redis transport layer.',
@@ -86,11 +85,12 @@ describe('messagingServer::Redis', () => {
tap(event => ctx.client.ackMessage(event.raw)),
);
- const output$: MsgOutputEffect = event$ =>
+ const error$: MsgErrorEffect = event$ =>
event$.pipe(
- tap(async ({ event }) => {
- expect(event.type).toEqual(expectedEventType);
- expect(event.error).toEqual(expectedEventError);
+ map(({ event, error}) => ({ event, error: { name: error.name, message: error.message }})),
+ tap(async ({ event, error }) => {
+ expect(event.type).toEqual('EVENT_TEST');
+ expect(error).toEqual(expectedEventError);
await wait();
await server.close();
await client.close();
@@ -101,7 +101,7 @@ describe('messagingServer::Redis', () => {
const options = createOptions();
const client = await runClient(Transport.REDIS, options);
- const server = await runServer(Transport.REDIS, options)(event$, output$);
+ const server = await runServer(Transport.REDIS, options)(event$, undefined, error$);
const message = createMessage({ type: 'EVENT_TEST' });
const emitResult = await client.emitMessage(options.channel, message);

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
import { Subject, fromEvent, merge, from } from 'rxjs';
-import { map, filter, take, mapTo, first, mergeMap } from 'rxjs/operators';
+import { map, filter, take, mapTo, first, mergeMap, share } from 'rxjs/operators';
import { Channel, ConsumeMessage } from 'amqplib';
import { AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
@@ -42,6 +42,7 @@ class AmqpStrategyConnection implements TransportLayerConnection {
get message$() {
return this.msgSubject$.asObservable().pipe(
+ share(),
map(raw => ({
data: raw.content,
replyTo: raw.properties.replyTo,

MODIFY
packages/messaging/src/util/messaging.test.util.ts
packages/messaging/src/util/messaging.test.util.ts
import * as uuid from 'uuid/v4';
import { createMicroservice } from '../server/messaging.server';
import { TransportMessage, Transport } from '../transport/transport.interface';
-import { MsgEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
+import { MsgEffect, MsgOutputEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
import { messagingListener } from '../server/messaging.server.listener';
import { provideTransportLayer } from '../transport/transport.provider';
-export const runServer = (transport: any, options: any) => (effect$?: MsgEffect, output$?: MsgOutputEffect) =>
+export const runServer = (transport: any, options: any) => (effect$?: MsgEffect, output$?: MsgOutputEffect, error$?: MsgErrorEffect) =>
createMicroservice({
options,
transport,
- messagingListener: messagingListener(effect$ ? { effects: [effect$], output$ } : undefined),
+ messagingListener: messagingListener(effect$ ? { effects: [effect$], output$, error$ } : undefined),
}).then(app => app());
export const runClient = (transport: Transport, transportOptions: any) =>

fix(messaging): mismatched events