feat(messaging): ackEvent/nackEvent functions
ADD
packages/messaging/src/ack/ack.spec.ts
+import { createEffectContext, contextFactory, lookup, Event, bindTo } from '@marblejs/core';
+import { TransportLayerConnection } from '../transport/transport.interface';
+import { EventTimerStoreToken, EventTimerStore } from '../eventStore/eventTimerStore';
+import { ackEvent, nackEvent, nackAndResendEvent } from './ack';
+
+const prepareEffectContext = async () => {
+ const ctx = await contextFactory(bindTo(EventTimerStoreToken)(EventTimerStore));
+ const ask = lookup(ctx);
+ const client = {
+ ackMessage: jest.fn(),
+ nackMessage: jest.fn(),
+ } as unknown as TransportLayerConnection;
+
+ return createEffectContext({ ask, client });
+};
+
+describe('#ackEvent, #nackEvent, #nackAndRequeueEvent', () => {
+ test('handles event with empty metadata', async () => {
+ // given
+ const ctx = await prepareEffectContext();
+ const event: Event = { type: 'TEST' };
+
+ // when
+ const result = Promise.all([
+ ackEvent(ctx)(event)(),
+ nackEvent(ctx)(event)(),
+ nackAndResendEvent(ctx)(event)(),
+ ]);
+
+ expect(result).resolves.toEqual([true, true, true]);
+ });
+
+ test('handles event with metadata defined', async () => {
+ // given
+ const ctx = await prepareEffectContext();
+ const event: Event = { type: 'TEST', metadata: { correlationId: '123', raw: {} } };
+
+ // when
+ const result = Promise.all([
+ ackEvent(ctx)(event)(),
+ nackEvent(ctx)(event)(),
+ nackAndResendEvent(ctx)(event)(),
+ ]);
+
+ expect(result).resolves.toEqual([true, true, true]);
+ });
+});

ADD
packages/messaging/src/ack/ack.ts
+import { promisify } from 'util';
+import * as T from 'fp-ts/lib/Task';
+import { pipe } from 'fp-ts/lib/pipeable';
+import { constTrue } from 'fp-ts/lib/function';
+import { Event, EffectContext, useContext } from '@marblejs/core';
+import { TransportLayerConnection } from '../transport/transport.interface';
+import { EventTimerStore, EventTimerStoreToken } from '../eventStore/eventTimerStore';
+
+const RESEND = true;
+
+const DONT_RESEND = false;
+
+const timeoutTask = () => promisify(setTimeout)(10);
+
+const unregisterTask = (store: EventTimerStore) => (e: Event) => T.fromIO(store.unregister(e));
+
+const ackTask = (ctx: EffectContext<TransportLayerConnection>) => (e: Event) => T.fromIO(() => ctx.client.ackMessage(e.metadata?.raw));
+
+const nackTask = (ctx: EffectContext<TransportLayerConnection>) => (e: Event) => (resend: boolean) => T.fromIO(() => ctx.client.nackMessage(e.metadata?.raw, resend));
+
+export const ackEvent = (ctx: EffectContext<TransportLayerConnection>) =>
+ pipe(
+ ctx.ask,
+ useContext(EventTimerStoreToken),
+ store => (event: Event): T.Task<boolean> =>
+ pipe(
+ ackTask(ctx)(event),
+ T.chain(_ => unregisterTask(store)(event)),
+ T.chain(_ => timeoutTask),
+ T.map(constTrue),
+ ),
+ );
+
+export const nackEvent = (ctx: EffectContext<TransportLayerConnection>) =>
+ pipe(
+ ctx.ask,
+ useContext(EventTimerStoreToken),
+ store => (event: Event): T.Task<boolean> =>
+ pipe(
+ nackTask(ctx)(event)(DONT_RESEND),
+ T.chain(_ => unregisterTask(store)(event)),
+ T.chain(_ => timeoutTask),
+ T.map(constTrue),
+ ),
+ );
+
+export const nackAndResendEvent = (ctx: EffectContext<TransportLayerConnection>) =>
+ pipe(
+ ctx.ask,
+ useContext(EventTimerStoreToken),
+ store => (event: Event): T.Task<boolean> =>
+ pipe(
+ nackTask(ctx)(event)(RESEND),
+ T.chain(_ => unregisterTask(store)(event)),
+ T.chain(_ => timeoutTask),
+ T.map(constTrue),
+ ),
+ );

feat(messaging): ackEvent/nackEvent functions