feat(messaging): eventTimerStore - auto remove entry after timeout
ADD
packages/messaging/src/eventStore/eventTimerStore.spec.ts
+import { createContext } from '@marblejs/core';
+import { createUuid, wait } from '@marblejs/core/dist/+internal/utils';
+import { EventTimerStore } from './eventTimerStore';
+
+describe('EventTimerStore', () => {
+ const ctx = createContext();
+
+ test('registers event timeout for event with "correlationId"', () => {
+ // given
+ const timeout = 10 * 1000;
+ const timeoutHandler = jest.fn();
+ const event = { type: 'TEST', metadata: { correlationId: createUuid() } };
+ const eventTimerStore = EventTimerStore(ctx);
+
+ // when
+ eventTimerStore.register(timeout)(timeoutHandler)(event)();
+
+ // then
+ const entry = eventTimerStore.store.get(event.metadata.correlationId);
+
+ expect(entry).toBeDefined();
+ expect(eventTimerStore.store.size).toEqual(1);
+
+ if (entry) clearTimeout(entry);
+ });
+
+ test('doesn\'t register event timeout when event "correlationId" is not defined', () => {
+ // given
+ const timeout = 10 * 1000;
+ const timeoutHandler = jest.fn();
+ const event = { type: 'TEST' };
+ const eventTimerStore = EventTimerStore(ctx);
+
+ // when
+ eventTimerStore.register(timeout)(timeoutHandler)(event)();
+
+ // then
+ expect(eventTimerStore.store.size).toEqual(0);
+ });
+
+ test('timeouts event after specified period and automatically removes entry from store', async () => {
+ // given
+ const timeout = 500;
+ const timeoutHandler = jest.fn();
+ const event = { type: 'TEST', metadata: { correlationId: createUuid() } };
+ const eventTimerStore = EventTimerStore(ctx);
+
+ // when
+ eventTimerStore.register(timeout)(timeoutHandler)(event)();
+
+ // then
+ expect(eventTimerStore.store.size).toEqual(1);
+
+ await wait(1);
+
+ expect(timeoutHandler).toHaveBeenCalled();
+ expect(eventTimerStore.store.size).toEqual(0);
+ });
+
+ test('manually unregisters timeout', async () => {
+ // given
+ const timeout = 5 * 1000;
+ const timeoutHandler = jest.fn();
+ const event = { type: 'TEST', metadata: { correlationId: createUuid() } };
+ const eventTimerStore = EventTimerStore(ctx);
+
+ // when, then
+ eventTimerStore.register(timeout)(timeoutHandler)(event)();
+
+ expect(eventTimerStore.store.size).toEqual(1);
+ expect(timeoutHandler).not.toHaveBeenCalled();
+
+ eventTimerStore.unregister(event)();
+
+ expect(eventTimerStore.store.size).toEqual(0);
+ expect(timeoutHandler).not.toHaveBeenCalled();
+ });
+
+ test('if timeout doesn\'t exist manual unregister doesn\'t fail', async () => {
+ // given
+ const event = { type: 'TEST', metadata: { correlationId: createUuid() } };
+ const eventTimerStore = EventTimerStore(ctx);
+
+ // when, then
+ expect(eventTimerStore.store.size).toEqual(0);
+
+ eventTimerStore.unregister(event)();
+
+ expect(eventTimerStore.store.size).toEqual(0);
+ });
+});

MODIFY
packages/messaging/src/eventStore/eventTimerStore.ts
packages/messaging/src/eventStore/eventTimerStore.ts
import * as IO from 'fp-ts/lib/IO';
import * as O from 'fp-ts/lib/Option';
import { pipe } from 'fp-ts/lib/pipeable';
+import { constant } from 'fp-ts/lib/function';
import { createReader, createContextToken, Event } from '@marblejs/core';
-const createStore = () => new Map<string, NodeJS.Timeout>();
+type StoreKey = string;
+type Store = Map<StoreKey, NodeJS.Timeout>;
-const getTimer = (store: Map<string, NodeJS.Timeout>) => (event: Event): readonly [string, NodeJS.Timeout] | undefined =>
+const createStore = (): Store => new Map();
+
+const setTimer = (store: Store) => (key: StoreKey) => (timeout: number) => (handler: IO.IO<any>): IO.IO<NodeJS.Timeout> => () =>
+ pipe(
+ pipe(handler, IO.chain(() => deleteTimerEntry(store)(key))),
+ handler => setTimeout(handler, timeout),
+ );
+
+const clearTimer = (timer: NodeJS.Timeout): IO.IO<void> => () =>
+ clearTimeout(timer);
+
+const setTimerEntry = (store: Store) => (key: StoreKey) => (timer: NodeJS.Timeout): IO.IO<Store> => () =>
+ pipe(
+ store.set(key, timer),
+ constant(store),
+ );
+
+const deleteTimerEntry = (store: Store) => (key: StoreKey): IO.IO<Store> => () =>
+ pipe(
+ store.delete(key),
+ constant(store),
+ );
+
+const getTimer = (store: Store) => (event: Event): readonly [StoreKey, NodeJS.Timeout] | undefined =>
pipe(
O.fromNullable(event.metadata?.correlationId),
- O.chain(eventId => pipe(
- O.fromNullable(store.get(eventId)),
- O.map(timer => ([eventId, timer] as const)),
+ O.chain(eventkey => pipe(
+ O.fromNullable(store.get(eventkey)),
+ O.map(timer => ([eventkey, timer] as const)),
)),
O.toUndefined,
);
-const register = (store: Map<string, NodeJS.Timeout>) => (timeout: number) => (handler: (() => any)) => (event: Event): IO.IO<void> =>
+/**
+ * Registers timeout handler for given event
+ * @since v3.3.0
+ */
+const register = (store: Store) => (timeout: number) => <T>(handler: IO.IO<T>) => (event: Event): IO.IO<void> =>
pipe(
IO.of(event.metadata?.correlationId),
- IO.chain(id => () => {
- id && pipe(
- setTimeout(handler, timeout),
- timer => store.set(id, timer),
- );
- }),
+ IO.chain(key => key
+ ? pipe(setTimer(store)(key)(timeout)(handler), IO.chain(setTimerEntry(store)(key)))
+ : IO.of(undefined)),
);
-const unregister = (store: Map<string, NodeJS.Timeout>) => (event: Event): IO.IO<void> =>
+/**
+ * Unregisters timeout handler for given event
+ * @since v3.3.0
+ */
+const unregister = (store: Store) => (event: Event): IO.IO<void> =>
pipe(
IO.of(event),
IO.map(getTimer(store)),
- IO.chain(data => () => {
- data && clearTimeout(data[1]);
- data && store.delete(data[0]);
- }),
+ IO.chain(entry => entry
+ ? pipe(clearTimer(entry[1]), IO.chain(() => deleteTimerEntry(store)(entry[0])))
+ : IO.of(undefined))
);
/**
@@ -44,6 +73,7 @@ const unregister = (store: Map<string, NodeJS.Timeout>) => (event: Event): IO.IO
export const EventTimerStore = createReader(_ => pipe(
createStore(),
store => ({
+ store,
register: register(store),
unregister: unregister(store),
}),

feat(messaging): eventTimerStore - auto remove entry after timeout