fix(messaging): RPC event missmach
MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -11,8 +11,8 @@ import {
import * as O from 'fp-ts/lib/Option';
import * as R from 'fp-ts/lib/Reader';
import { pipe } from 'fp-ts/lib/pipeable';
-import { Observable, Subscription, Subject, of } from 'rxjs';
-import { map, publish, withLatestFrom, takeUntil, catchError, take } from 'rxjs/operators';
+import { Observable, Subscription, Subject, of, zip } from 'rxjs';
+import { map, publish, takeUntil, catchError, take } from 'rxjs/operators';
import {
TransportMessage,
TransportMessageTransformer,
@@ -63,16 +63,19 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const message$ = conn.message$.pipe(
map(decode),
- publish(msg$ => combinedMiddlewares(msg$.pipe(map(m => m.data)), ctx).pipe(
- withLatestFrom(msg$),
+ publish(msg$ => zip(
+ combinedMiddlewares(msg$.pipe(map(m => m.data)), ctx),
+ msg$,
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- publish(msg$ => combinedEffects(msg$.pipe(map(m => m.data)), ctx).pipe(
- withLatestFrom(msg$),
+ publish(msg$ => zip(
+ combinedEffects(msg$.pipe(map(m => m.data)), ctx),
+ msg$,
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- publish(msg$ => output$(msg$.pipe(map(m => ({ event: m.data, initiator: m }))), ctx).pipe(
- withLatestFrom(msg$),
+ publish(msg$ => zip(
+ output$(msg$.pipe(map(m => ({ event: m.data, initiator: m }))), ctx),
+ msg$,
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
catchError((error: EventError) => error$(of({ event: error.event, error }), ctx).pipe(

MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
import { Event, matchEvent, use, EventError, createEvent } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
-import { map, tap, mergeMapTo } from 'rxjs/operators';
+import { map, tap, mergeMapTo, delay } from 'rxjs/operators';
import { Transport } from '../../transport/transport.interface';
import { MsgEffect, MsgErrorEffect } from '../../effects/messaging.effects.interface';
import { Subject, throwError } from 'rxjs';
@@ -19,6 +19,7 @@ describe('messagingServer::AMQP', () => {
const rpc$: MsgEffect = event$ =>
event$.pipe(
matchEvent('RPC_TEST'),
+ delay(250),
use(eventValidator$(t.number)),
map(event => event.payload),
map(payload => ({ type: 'RPC_TEST_RESULT', payload: payload + 1 })),
@@ -27,12 +28,17 @@ describe('messagingServer::AMQP', () => {
const options = createOptions({ expectAck: false });
const client = await runClient(Transport.AMQP, options);
const server = await runServer(Transport.AMQP, options)(rpc$);
- const message = createMessage({ type: 'RPC_TEST', payload: 1 });
- const result = await client.sendMessage(options.queue, message);
- const parsedResult = JSON.parse(result.data.toString());
+ const result = await Promise.all([
+ client.sendMessage(options.queue, createMessage({ type: 'RPC_TEST', payload: 1 })),
+ client.sendMessage(options.queue, createMessage({ type: 'RPC_TEST', payload: 10 })),
+ ]);
- expect(parsedResult).toEqual({ type: 'RPC_TEST_RESULT', payload: 2 });
+ const parsedResult0 = JSON.parse(result[0].data.toString());
+ const parsedResult1 = JSON.parse(result[1].data.toString());
+
+ expect(parsedResult0).toEqual({ type: 'RPC_TEST_RESULT', payload: 2 });
+ expect(parsedResult1).toEqual({ type: 'RPC_TEST_RESULT', payload: 11 });
await server.close();
});

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
-import { Subject, fromEvent, from, merge } from 'rxjs';
-import { map, filter, take, mergeMap, mapTo, first } from 'rxjs/operators';
+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';

fix(messaging): RPC event missmach