feat(messaging): rejectUnhandled$ middleware
MODIFY
packages/messaging/src/eventStore/eventTimerStore.ts
packages/messaging/src/eventStore/eventTimerStore.ts
import * as IO from 'fp-ts/lib/IO';
-import * as M from 'fp-ts/lib/Map';
import * as O from 'fp-ts/lib/Option';
-import { ordString } from 'fp-ts/lib/Ord';
import { pipe } from 'fp-ts/lib/pipeable';
import { createReader, createContextToken, Event } from '@marblejs/core';
@@ -11,7 +9,7 @@ const getTimer = (store: Map<string, NodeJS.Timeout>) => (event: Event): readonl
pipe(
O.fromNullable(event.metadata?.correlationId),
O.chain(eventId => pipe(
- M.lookup(ordString)(eventId, store),
+ O.fromNullable(store.get(eventId)),
O.map(timer => ([eventId, timer] as const)),
)),
O.toUndefined,
@@ -20,10 +18,12 @@ const getTimer = (store: Map<string, NodeJS.Timeout>) => (event: Event): readonl
const register = (store: Map<string, NodeJS.Timeout>) => (timeout: number) => (handler: (() => any)) => (event: Event): IO.IO<void> =>
pipe(
IO.of(event.metadata?.correlationId),
- IO.chain(id => () => id && pipe(
+ IO.chain(id => () => {
+ id && pipe(
setTimeout(handler, timeout),
- timer => M.insertAt(ordString)(id, timer)(store),
- )),
+ timer => store.set(id, timer),
+ );
+ }),
);
const unregister = (store: Map<string, NodeJS.Timeout>) => (event: Event): IO.IO<void> =>

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
@@ -8,11 +8,13 @@ export { AmqpStrategyOptions, AmqpConnectionStatus } from './transport/strategie
export { RedisStrategyOptions, RedisConnectionStatus } from './transport/strategies/redis.strategy.interface';
export { LocalStrategyOptions, EVENT_BUS_CHANNEL } from './transport/strategies/local.strategy.interface';
-// effects
+// effects, middlewares
export * from './effects/messaging.effects.interface';
+export * from './middlewares/messaging.ack.middleware';
// handy functions
export { reply } from './reply/reply';
+export { ackEvent, nackEvent, nackAndResendEvent } from './ack/ack';
// server
export * from './server/messaging.server';
@@ -23,3 +25,4 @@ export { messagingListener } from './server/messaging.server.listener';
// readers
export * from './eventbus/messaging.eventBus.reader';
+export * from './eventStore/eventTimerStore';

ADD
packages/messaging/src/middlewares/messaging.ack.middleware.ts
+import { useContext } from '@marblejs/core';
+import { tap } from 'rxjs/operators';
+import { MsgEffect } from '../effects/messaging.effects.interface';
+import { EventTimerStoreToken } from '../eventStore/eventTimerStore';
+import { Transport } from '../transport/transport.interface';
+import { nackEvent } from '../ack/ack';
+
+const ACK_TRANSPORTS = [Transport.AMQP];
+
+export const rejectUnhandled$: MsgEffect = (event$, ctx) => {
+ const eventTimerStore = useContext(EventTimerStoreToken)(ctx.ask);
+ const isAckTransport = ACK_TRANSPORTS.includes(ctx.client.type);
+ const timeout = ctx.client.config.timeout;
+
+ return event$.pipe(
+ tap(event => {
+ if (isAckTransport) {
+ const handler = nackEvent(ctx)(event);
+ eventTimerStore.register(timeout)(handler)(event)();
+ }
+ }),
+ );
+};

MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
@@ -5,9 +5,11 @@ import { forkJoin, TimeoutError } from 'rxjs';
import { map, tap, delay, mapTo } from 'rxjs/operators';
import { TransportLayerConnection } from '../../transport/transport.interface';
import { MsgEffect } from '../../effects/messaging.effects.interface';
+import { rejectUnhandled$ } from '../../middlewares/messaging.ack.middleware';
import { MessagingClient } from '../../client/messaging.client.interface';
import { reply } from '../../reply/reply';
import * as Util from '../../util/messaging.test.util';
+import { ackEvent } from '../../ack/ack';
describe('messagingServer::AMQP', () => {
let client: MessagingClient;
@@ -121,11 +123,11 @@ describe('messagingServer::AMQP', () => {
const optionsMicroservice = Util.createAmqpOptions({ expectAck: true });
const optionsClient = Util.createAmqpOptions({ queue: optionsMicroservice.queue });
- const ack$: MsgEffect = (event$, { client }) =>
+ const ack$: MsgEffect = (event$, ctx) =>
event$.pipe(
matchEvent('ACK'),
delay(50),
- tap(event => client.ackMessage(event.metadata?.raw)),
+ tap(event => ackEvent(ctx)(event)()),
map(event => ({ type: 'ACK_RESPONSE', payload: event.payload })),
);
@@ -136,13 +138,39 @@ describe('messagingServer::AMQP', () => {
)(done);
// when
- microservice = await Util.createAmqpMicroservice(optionsMicroservice)({ effects: [ack$], output$ });
+ microservice = await Util.createAmqpMicroservice(optionsMicroservice)({ middlewares: [rejectUnhandled$], effects: [ack$], output$ });
client = await Util.createAmqpClient(optionsClient);
await client.emit({ type: 'ACK', payload: 1 });
await client.emit({ type: 'ACK', payload: 2 });
});
+ test('rejects unhandled event events and doesn\'t block the consumer', async done => {
+ // given
+ const optionsMicroservice = Util.createAmqpOptions({ expectAck: true, timeout: 100 });
+ const optionsClient = Util.createAmqpOptions({ queue: optionsMicroservice.queue });
+
+ const ack$: MsgEffect = (event$, ctx) =>
+ event$.pipe(
+ matchEvent('ACK'),
+ delay(50),
+ tap(event => ackEvent(ctx)(event)()),
+ map(event => ({ type: 'ACK_RESPONSE', payload: event.payload })),
+ );
+
+ // then
+ const output$ = Util.assertOutputEvent(
+ { type: 'ACK_RESPONSE' },
+ )(done);
+
+ // when
+ microservice = await Util.createAmqpMicroservice(optionsMicroservice)({ middlewares: [rejectUnhandled$], effects: [ack$], output$ });
+ client = await Util.createAmqpClient(optionsClient);
+
+ await client.emit({ type: 'TEST' });
+ await client.emit({ type: 'ACK', payload: 1 });
+ });
+
test('chains events by sending back to origin channel when no reply is defined', async done => {
// given
const options = Util.createAmqpOptions();

MODIFY
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
@@ -8,6 +8,7 @@ import { MsgEffect } from '../../effects/messaging.effects.interface';
import { MessagingClient } from '../../client/messaging.client.interface';
import { reply } from '../../reply/reply';
import * as Util from '../../util/messaging.test.util';
+import { ackEvent } from '../../ack/ack';
describe('messagingServer::Redis', () => {
let client: MessagingClient;
@@ -126,7 +127,7 @@ describe('messagingServer::Redis', () => {
const test$: MsgEffect = (event$, ctx) =>
event$.pipe(
matchEvent('TEST'),
- tap(event => ctx.client.ackMessage(event.metadata?.raw)),
+ tap(event => ackEvent(ctx)(event)()),
);
const error$ = Util.assertError({

feat(messaging): rejectUnhandled$ middleware