feat(messaging): automatic event reply for all messaging strategies
ADD
packages/messaging/src/middlewares/messaging.eventRouter.middleware.ts
+import { useContext } from '@marblejs/core';
+import { createUuid } from '@marblejs/core/dist/+internal/utils';
+import { map } from 'rxjs/operators';
+import { MsgOutputEffect } from '../effects/messaging.effects.interface';
+import { TransportLayerToken } from '../server/messaging.server.tokens';
+
+export const outputRouter$: MsgOutputEffect = (event$, ctx) => {
+ const transportLayer = useContext(TransportLayerToken)(ctx.ask);
+ const originChannel = transportLayer.config.channel;
+
+ return event$.pipe(
+ map(event => ({
+ ...event,
+ metadata: {
+ ...event.metadata,
+ correlationId: event.metadata?.correlationId ?? createUuid(),
+ replyTo: event.metadata?.replyTo ?? originChannel
+ },
+ }))
+ );
+};

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -12,14 +12,11 @@ import {
import { pipe } from 'fp-ts/lib/pipeable';
import { Observable, Subject, defer } from 'rxjs';
import { map, catchError, takeUntil, filter } from 'rxjs/operators';
-import {
- TransportMessage,
- TransportMessageTransformer,
- TransportLayerConnection,
-} from '../transport/transport.interface';
+import { TransportMessage, TransportMessageTransformer, TransportLayerConnection } from '../transport/transport.interface';
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';
+import { outputRouter$ } from '../middlewares/messaging.eventRouter.middleware';
export interface MessagingListenerConfig {
effects?: MsgEffect<any, any>[];
@@ -88,6 +85,7 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
const outgoingEvent$ = pipe(
eventSubject.asObservable(),
e$ => combinedEffects(e$, ctx),
+ e$ => outputRouter$(e$, ctx),
e$ => output$(e$, ctx),
e$ => outputLogger$(e$, ctx),
e$ => defer(() => processError(e$)),

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, combineEffects } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { Subject, throwError } from 'rxjs';
-import { map, tap, delay, bufferCount, mergeMap } from 'rxjs/operators';
+import { map, tap, delay, bufferCount, mergeMap, mapTo } from 'rxjs/operators';
import { Transport } from '../../transport/transport.interface';
import { MsgEffect, MsgErrorEffect, MsgOutputEffect } from '../../effects/messaging.effects.interface';
import { AmqpStrategyOptions } from '../../transport/strategies/amqp.strategy.interface';
@@ -79,7 +79,7 @@ describe('messagingServer::AMQP', () => {
eventSubject.subscribe(event => {
expect(event).toEqual({ type: 'EVENT_TEST_RESPONSE', payload: 2 });
setTimeout(async () => {
- await microservice.close();
+ await consumer.close();
await client.close();
done();
}, 1000);
@@ -96,17 +96,19 @@ describe('messagingServer::AMQP', () => {
tap(event => eventSubject.next(event)),
);
- const options = createOptions({ expectAck: true, queue: 'test_ack_queue_server' });
- const client = await runMicroserviceClient(Transport.AMQP, options);
- const microservice = await runMicroservice(Transport.AMQP, options)(event$);
+ const optionsClient = createOptions({ queue: 'test_ack_queue_server' });
+ const optionsConsumer = createOptions({ expectAck: true, queue: 'test_ack_queue_server' });
+ const client = await runMicroserviceClient(Transport.AMQP, optionsClient);
+ const consumer = await runMicroservice(Transport.AMQP, optionsConsumer)(event$);
const message = createMessage({ type: 'EVENT_TEST', payload: 1 });
- await client.emitMessage(options.queue, message);
+ await client.emitMessage(optionsClient.queue, message);
});
test('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 event2Outgoing: Event = { type: `${event2.type}__processed` };
const error = new Error('test_error');
const outputSubject = new Subject<[Event | undefined, Error | undefined]>();
@@ -115,7 +117,7 @@ describe('messagingServer::AMQP', () => {
.subscribe(data => {
expect(data[0][0]).toBeUndefined();
expect(data[0][1]).toEqual(error);
- expect(data[1][0]).toEqual(event2);
+ expect(data[1][0]).toEqual(event2Outgoing);
expect(data[1][1]).toBeUndefined();
setTimeout(async () => {
await microservice.close();
@@ -133,7 +135,7 @@ describe('messagingServer::AMQP', () => {
const event2$: MsgEffect = event$ =>
event$.pipe(
matchEvent(event2.type),
- map(event => ({ type: event.type })),
+ mapTo(event2Outgoing),
);
const error$: MsgErrorEffect = event$ =>

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.interface.ts
packages/messaging/src/transport/strategies/amqp.strategy.interface.ts
import { Options } from 'amqplib';
+import { NamedError } from '@marblejs/core/dist/+internal/utils';
import { Transport } from '../transport.interface';
export interface AmqpStrategy {
@@ -21,3 +22,16 @@ export enum AmqpConnectionStatus {
CONNECTION_LOST = 'CONNECTION_LOST',
CHANNEL_CONNECTION_LOST = 'CHANNEL_CONNECTION_LOST',
};
+
+export enum AmqpErrorType {
+ CANNOT_SET_ACK_FOR_NON_CONSUMER_CONNECTION = 'AmqpCannotSetExpectAckForNonConsumerConnection',
+}
+
+export class AmqpCannotSetExpectAckForNonConsumerConnection extends NamedError {
+ constructor() {
+ super(
+ AmqpErrorType.CANNOT_SET_ACK_FOR_NON_CONSUMER_CONNECTION,
+ `Non consumer connections cannot set "expectAck" attribute`,
+ );
+ }
+}

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
@@ -4,7 +4,7 @@ import { Channel, ConsumeMessage, Replies } from 'amqplib';
import { AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { TransportLayer, TransportMessage, TransportLayerConnection, Transport, DEFAULT_TIMEOUT } from '../transport.interface';
-import { AmqpStrategyOptions, AmqpConnectionStatus } from './amqp.strategy.interface';
+import { AmqpStrategyOptions, AmqpConnectionStatus, AmqpCannotSetExpectAckForNonConsumerConnection } from './amqp.strategy.interface';
class AmqpStrategyConnection implements TransportLayerConnection {
private statusSubject$ = new Subject<AmqpConnectionStatus>();
@@ -110,6 +110,10 @@ class AmqpStrategyConnection implements TransportLayerConnection {
emitMessage = async (queue: string, msg: TransportMessage<Buffer>) => {
const { correlationId, data, replyTo } = msg;
+ // don't try to emit the message to the same channel if "ack" mode is enabled
+ if (this.options.expectAck && queue === this.getChannel())
+ return false;
+
await this.channelWrapper.sendToQueue(queue, data, {
replyTo,
correlationId,
@@ -155,9 +159,12 @@ class AmqpStrategy implements TransportLayer {
}
async connect(opts?: { isConsumer: boolean }) {
- const { host, queue, queueOptions, prefetchCount } = this.options;
+ const { host, queue, queueOptions, prefetchCount, expectAck } = this.options;
const msgSubject$ = new Subject<ConsumeMessage>();
+ if (!opts?.isConsumer && expectAck)
+ throw new AmqpCannotSetExpectAckForNonConsumerConnection();
+
await import('amqplib');
const amqplib = await import('amqp-connection-manager');

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.ts
packages/messaging/src/transport/strategies/redis.strategy.ts
@@ -66,7 +66,7 @@ class RedisStrategyConnection implements TransportLayerConnection {
return this.consumerSubject.asObservable().pipe(
map(msg => RedisHelper.decodeMessage(msg.content)),
);
- }z
+ }
emitMessage = async (channel: string, message: TransportMessage<Buffer>) => {
const replyChannel = message.correlationId;

feat(messaging): automatic event reply for all messaging strategies