refactor(messaging): local strategy transport layer provider
+ customizable event bus timeouts
+ refactore transport layer provider layer
+ event bus no longer resides in global scope
MODIFY
packages/core/src/+internal/utils/error.util.ts
packages/core/src/+internal/utils/error.util.ts
@@ -17,3 +17,5 @@ export const encodeError = (error: any) =>
acc[key] = error[key];
return acc;
}, Object.create(null));
+
+export const throwException = (error: any) => { throw error };

MODIFY
packages/core/src/context/specs/context.spec.ts
packages/core/src/context/specs/context.spec.ts
@@ -21,6 +21,7 @@ import {
} from '../context';
import { createContextToken } from '../context.token.factory';
import { contextFactory } from '../context.helper';
+import { wait } from '../../+internal/utils';
describe('#bindTo', () => {
test('binds lazy reader to token', () => {
@@ -287,4 +288,53 @@ describe('#reader', () => {
expect(lookup(context)(dependencyToken_3)).toEqual(O.some('test_3'));
expect(lookup(context)(unknownToken)).toEqual(O.none);
});
+
+ test('asks context for dependant asynchronous dependencies', async () => {
+ // given
+ const executionOrder: number[] = [];
+ const spy1 = jest.fn(() => executionOrder.push(1));
+ const spy2 = jest.fn(() => executionOrder.push(2));
+ const spy3 = jest.fn(() => executionOrder.push(3));
+ const token1 = createContextToken<string>();
+ const token2 = createContextToken<string>();
+ const token3 = createContextToken<string>();
+
+ const dependency1 = pipe(reader, R.map(async () => {
+ spy1();
+ wait(0.5);
+ return 'test';
+ }));
+
+ const dependency2 = pipe(reader, R.map(ask => {
+ spy2();
+ return pipe(ask(token1), O.map(async v => {
+ await wait(0.5);
+ return v + '_1';
+ }), O.getOrElse(() => Promise.resolve('')));
+ }));
+
+ const dependency3 = pipe(reader, R.map(ask => {
+ spy3();
+ return pipe(ask(token2), O.map(async v => {
+ await wait(0.5);
+ return v + '_2';
+ }), O.getOrElse(() => Promise.resolve('')));
+ }));
+
+ // when
+ const context = await flow(
+ registerAll([
+ bindEagerlyTo(token1)(dependency1),
+ bindEagerlyTo(token2)(dependency2),
+ bindEagerlyTo(token3)(dependency3),
+ ]),
+ resolve,
+ )(createContext());
+
+ // then
+ expect(lookup(context)(token1)).toEqual(O.some('test'));
+ expect(lookup(context)(token2)).toEqual(O.some('test_1'));
+ expect(lookup(context)(token3)).toEqual(O.some('test_1_2'));
+ expect(executionOrder).toEqual([1, 2, 3]);
+ });
});

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
+import * as E from 'fp-ts/lib/Either';
import * as O from 'fp-ts/lib/Option';
+import * as R from 'fp-ts/lib/Reader';
import { pipe } from 'fp-ts/lib/pipeable';
import {
Event,
@@ -12,11 +14,13 @@ import {
LoggerLevel,
isEventError,
EventError,
- createReader,
+ lookup,
+ Context,
} from '@marblejs/core';
-import { isNamedError, NamedError } from '@marblejs/core/dist/+internal/utils';
+import { isNamedError, NamedError, throwException } from '@marblejs/core/dist/+internal/utils';
import { from, Observable, EMPTY, throwError, of, defer } from 'rxjs';
import { mergeMap, take, map, timeout } from 'rxjs/operators';
+import { identity } from 'fp-ts/lib/function';
import { TransportMessage, DEFAULT_TIMEOUT } from '../transport/transport.interface';
import { provideTransportLayer } from '../transport/transport.provider';
import { jsonTransformer } from '../transport/transport.transformer';
@@ -29,9 +33,15 @@ export const messagingClient = (config: MessagingClientConfig) => {
msgTransformer = jsonTransformer,
} = config;
- return createReader<Promise<MessagingClient>>(async ask => {
+ return pipe(
+ R.ask<Context>(),
+ R.map<Context, Promise<MessagingClient>>(async context => {
+ const ask = lookup(context);
const logger = useContext(LoggerToken)(ask);
- const transportLayer = provideTransportLayer(transport, options);
+ const transportLayer = pipe(
+ provideTransportLayer(transport, options)(context),
+ E.fold(throwException, identity));
+
const connection = await transportLayer.connect({ isConsumer: false });
const emit = async <I extends Event>(event: I): Promise<void> => {
@@ -71,7 +81,7 @@ export const messagingClient = (config: MessagingClientConfig) => {
}
return defer(() => from(connection.sendMessage(channel, message)).pipe(
- timeout(config.options.timeout || DEFAULT_TIMEOUT),
+ timeout(config.options.timeout ?? connection.config.timeout ?? DEFAULT_TIMEOUT),
map(m => msgTransformer.decode<O>(m.data)),
mergeMap(catchErrorEvent),
take(1),
@@ -100,5 +110,5 @@ export const messagingClient = (config: MessagingClientConfig) => {
send,
close,
};
- });
+ }));
};

MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
-import { Event, matchEvent, combineEffects, act, EventError } from '@marblejs/core';
+import { Event, matchEvent, combineEffects, act, EventError, contextFactory, bindEagerlyTo, lookup, useContext } from '@marblejs/core';
import { wait, NamedError } from '@marblejs/core/dist/+internal/utils';
import { eventValidator$, t } from '@marblejs/middleware-io';
-import { throwError } from 'rxjs';
+import { throwError, of, TimeoutError } from 'rxjs';
import { delay, map, tap, ignoreElements } from 'rxjs/operators';
import { flow } from 'fp-ts/lib/function';
+import { pipe } from 'fp-ts/lib/pipeable';
import { MsgEffect } from '../effects/messaging.effects.interface';
-import { runEventBus, runEventBusClient } from '../util/messaging.test.util';
import { reply } from '../reply/reply';
+import { messagingListener } from '../server/messaging.server.listener';
+import { createEventBusTestBed } from '../util/messaging.test.util';
+import { EventBusToken, eventBus } from './messaging.eventBus.reader';
+import { EventBusClientToken, eventBusClient } from './messaging.eventBusClient.reader';
describe('#eventBus', () => {
test('handles RPC event', async () => {
@@ -20,8 +24,8 @@ describe('#eventBus', () => {
))
);
- const eventBus = await runEventBus({ effects: [rpc$] });
- const eventBusClient = await runEventBusClient();
+ const [eventBus, eventBusClient] = await createEventBusTestBed({ effects: [rpc$] });
+
const event: Event = { type: 'RPC_TEST', payload: 1 };
const result = await eventBusClient.send(event).toPromise();
@@ -53,8 +57,7 @@ describe('#eventBus', () => {
))
);
- const eventBus = await runEventBus({ effects: [combineEffects(rpc1$, rpc2$)] });
- const eventBusClient = await runEventBusClient();
+ const [eventBus, eventBusClient] = await createEventBusTestBed({ effects: [combineEffects(rpc1$, rpc2$)] });
const event1: Event = { type: 'RPC_1_TEST', payload: 1 };
const event2: Event = { type: 'RPC_2_TEST', payload: 2 };
@@ -89,8 +92,7 @@ describe('#eventBus', () => {
)),
);
- const eventBus = await runEventBus({ effects: [rpc1$] });
- const eventBusClient = await runEventBusClient();
+ const [eventBus, eventBusClient] = await createEventBusTestBed({ effects: [rpc1$] });
const event1: Event = { type: 'RPC_TEST', payload: 1 };
const event2: Event = { type: 'RPC_TEST', payload: '2' };
@@ -129,8 +131,7 @@ describe('#eventBus', () => {
ignoreElements(),
);
- const eventBus = await runEventBus({ effects: [foo$] });
- const eventBusClient = await runEventBusClient();
+ const [eventBus, eventBusClient] = await createEventBusTestBed({ effects: [foo$] });
const event: Event = { type: 'TEST', payload: 1 };
await eventBusClient.emit(event);
@@ -151,8 +152,7 @@ describe('#eventBus', () => {
),
);
- const eventBus = await runEventBus({ effects: [rpc$] });
- const eventBusClient = await runEventBusClient();
+ const [eventBus, eventBusClient] = await createEventBusTestBed({ effects: [rpc$] });
const event: Event = { type: 'RPC_TEST' };
const result = eventBusClient.send(event).toPromise();
@@ -177,8 +177,7 @@ describe('#eventBus', () => {
),
);
- const eventBus = await runEventBus({ effects: [rpc$] });
- const eventBusClient = await runEventBusClient();
+ const [eventBus, eventBusClient] = await createEventBusTestBed({ effects: [rpc$] });
const event: Event = { type: 'RPC_TEST' };
const result = eventBusClient.send(event).toPromise();
@@ -197,8 +196,7 @@ describe('#eventBus', () => {
act(() => throwError(error)),
);
- const eventBus = await runEventBus({ effects: [rpc$] });
- const eventBusClient = await runEventBusClient();
+ const [eventBus, eventBusClient] = await createEventBusTestBed({ effects: [rpc$] });
const event: Event = { type: 'RPC_TEST' };
const result = eventBusClient.send(event).toPromise();
@@ -207,4 +205,49 @@ describe('#eventBus', () => {
await eventBus.close();
await eventBusClient.close();
});
+
+ test('allows to set custom timeout', async () => {
+ const timeout = 1;
+
+ const rpc$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('RPC_TEST'),
+ act(event => pipe(
+ of(reply(event)({ type: 'RPC_TEST_RESULT' })),
+ delay(100),
+ )),
+ );
+
+ const event: Event = { type: 'RPC_TEST' };
+ const listener = messagingListener({ effects: [rpc$] });
+
+ const ask = pipe(
+ await contextFactory(
+ bindEagerlyTo(EventBusToken)(eventBus({ listener, timeout })),
+ bindEagerlyTo(EventBusClientToken)(eventBusClient),
+ ),
+ lookup,
+ );
+
+ const eventBusInstance = useContext(EventBusToken)(ask);
+ const eventBusClientInstance = useContext(EventBusClientToken)(ask);
+
+ expect(eventBusInstance.config.timeout).toEqual(timeout);
+
+ const result = eventBusClientInstance.send(event).toPromise();
+
+ await expect(result).rejects.toEqual(new TimeoutError());
+
+ await eventBusInstance.close();
+ await eventBusClientInstance.close();
+ });
+
+ test('fails if event bus client is registered before main EventBus reader', async () => {
+ const context = contextFactory(
+ bindEagerlyTo(EventBusClientToken)(eventBusClient),
+ bindEagerlyTo(EventBusToken)(eventBus({ listener: messagingListener() })),
+ );
+
+ await expect(context).rejects.toEqual(new Error('Cannot provide non-evaluated EventBus transport layer'));
+ });
});

MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
@@ -3,25 +3,22 @@ import * as R from 'fp-ts/lib/Reader';
import { Context, bindTo, createContextToken, contextFactory, bindEagerlyTo, DerivedContextToken, logContext, LoggerTag } from '@marblejs/core';
import { TransportLayerToken } from '../server/messaging.server.tokens';
import { Transport, TransportLayerConnection } from '../transport/transport.interface';
-import { provideTransportLayer } from '../transport/transport.provider';
import { messagingListener } from '../server/messaging.server.listener';
-import { MessagingClient } from '../client/messaging.client.interface';
-import { messagingClient } from '../client/messaging.client';
import { EventTimerStoreToken, EventTimerStore } from '../eventStore/eventTimerStore';
+import { createLocalStrategy } from '../transport/strategies/local.strategy';
export interface EventBusConfig {
listener: ReturnType<typeof messagingListener>;
timeout?: number;
}
-export const EventBusToken = createContextToken<TransportLayerConnection>('EventBusToken');
-export const EventBusClientToken = createContextToken<MessagingClient>('EventBusClientToken');
+export const EventBusToken = createContextToken<TransportLayerConnection<Transport.LOCAL>>('EventBus');
export const eventBus = (config: EventBusConfig) => pipe(
R.ask<Context>(),
R.map(async derivedContext => {
const { listener } = config;
- const transportLayer = provideTransportLayer(Transport.LOCAL, { timeout: config.timeout });
+ const transportLayer = createLocalStrategy(config);
const transportLayerConnection = await transportLayer.connect();
const context = await contextFactory(
@@ -37,11 +34,3 @@ export const eventBus = (config: EventBusConfig) => pipe(
return transportLayerConnection;
}),
);
-
-export const eventBusClient = pipe(
- R.ask<Context>(),
- R.map(messagingClient({
- transport: Transport.LOCAL,
- options: {},
- })),
-);

ADD
packages/messaging/src/eventbus/messaging.eventBusClient.reader.ts
+import { pipe } from 'fp-ts/lib/pipeable';
+import * as R from 'fp-ts/lib/Reader';
+import { Context, createContextToken } from '@marblejs/core';
+import { Transport } from '../transport/transport.interface';
+import { MessagingClient } from '../client/messaging.client.interface';
+import { messagingClient } from '../client/messaging.client';
+
+export const EventBusClientToken = createContextToken<MessagingClient>('EventBusClient');
+
+export const eventBusClient = pipe(
+ R.ask<Context>(),
+ R.map(messagingClient({
+ transport: Transport.LOCAL,
+ options: {},
+ })),
+);

MODIFY
packages/messaging/src/index.spec.ts
packages/messaging/src/index.spec.ts
@@ -8,6 +8,10 @@ describe('@marblejs/messaging', () => {
expect(API.createMicroservice).toBeDefined();
expect(API.messagingClient).toBeDefined();
expect(API.messagingListener).toBeDefined();
+ expect(API.eventBus).toBeDefined();
+ expect(API.eventBusClient).toBeDefined();
+ expect(API.EventBusClientToken).toBeDefined();
+ expect(API.EventBusToken).toBeDefined();
expect(API.reply).toBeDefined();
expect(API.ackEvent).toBeDefined();
expect(API.nackEvent).toBeDefined();

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
@@ -24,4 +24,5 @@ export { messagingListener } from './server/messaging.server.listener';
// readers
export * from './eventbus/messaging.eventBus.reader';
+export * from './eventbus/messaging.eventBusClient.reader';
export * from './eventStore/eventTimerStore';

MODIFY
packages/messaging/src/server/messaging.server.ts
packages/messaging/src/server/messaging.server.ts
+import * as E from 'fp-ts/lib/Either';
+import { pipe } from 'fp-ts/lib/pipeable';
+import { identity } from 'fp-ts/lib/function';
import { Subject } from 'rxjs';
import { takeWhile, takeUntil, take } from 'rxjs/operators';
import { bindTo, createEffectContext, combineEffects, ServerIO, lookup, logContext, LoggerTag, contextFactory } from '@marblejs/core';
+import { throwException } from '@marblejs/core/dist/+internal/utils';
import { provideTransportLayer } from '../transport/transport.provider';
import { statusLogger$ } from '../middlewares/messaging.statusLogger.middleware';
import { TransportLayerConnection } from '../transport/transport.interface';
@@ -18,8 +22,11 @@ export const createMicroservice = async (config: CreateMicroserviceConfig) => {
listener,
} = config;
+ const transportLayer = pipe(
+ provideTransportLayer(transport, options)(),
+ E.fold(throwException, identity));
+
const serverEventsSubject = new Subject<AllServerEvents>();
- const transportLayer = provideTransportLayer(transport, options);
const boundEventTimerStore = bindTo(EventTimerStoreToken)(EventTimerStore);
const boundTransportLayer = bindTo(TransportLayerToken)(() => transportLayer);
const boundServerEvents = bindTo(ServerEventsToken)(() => serverEventsSubject);

DELETE
packages/messaging/src/transport/strategies/local.strategy.provider.ts
-import * as O from 'fp-ts/lib/Option';
-import { pipe } from 'fp-ts/lib/pipeable';
-import { TransportLayer, TransportLayerConnection, Transport } from '../transport.interface';
-import { createLocalStrategy } from './local.strategy';
-import { LocalStrategyOptions } from './local.strategy.interface';
-
-export class LocalStrategyProvider implements TransportLayer<Transport.LOCAL> {
- private static instance: LocalStrategyProvider;
- private readonly strategy: TransportLayer<Transport.LOCAL>;
- private readonly strategyConnection: Promise<TransportLayerConnection<Transport.LOCAL>>;
-
- private constructor(options: LocalStrategyOptions) {
- this.strategy = createLocalStrategy(options);
- this.strategyConnection = this.strategy.connect();
- }
-
- static getDefault = (options: LocalStrategyOptions) => pipe(
- O.fromNullable(LocalStrategyProvider.instance),
- O.getOrElse(() => LocalStrategyProvider.instance = new LocalStrategyProvider(options)),
- );
-
- async connect() {
- return this.strategyConnection;
- }
-
- get type() {
- return this.strategy.type;
- }
-
- get config() {
- return this.strategy.config;
- }
-}
-
-export const provideLocalStrategy = (options: LocalStrategyOptions) => LocalStrategyProvider.getDefault(options);

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -62,3 +62,8 @@ export interface TransportMessage<T = Event> {
replyTo?: string;
correlationId?: string;
}
+
+export const isTransportLayerConnection = <T extends Transport>(data: any): data is TransportLayerConnection<T> => {
+ const conn = data as TransportLayerConnection;
+ return Boolean(conn.status$ && conn.emitMessage && conn.sendMessage);
+}

MODIFY
packages/messaging/src/transport/transport.provider.spec.ts
packages/messaging/src/transport/transport.provider.spec.ts
+import * as E from 'fp-ts/lib/Either';
+import { pipe } from 'fp-ts/lib/pipeable';
+import { createContext, contextFactory, bindEagerlyTo, bindTo, register } from '@marblejs/core';
+import { EventBusToken, eventBus } from '../eventbus/messaging.eventBus.reader';
+import { messagingListener } from '../server/messaging.server.listener';
import { provideTransportLayer } from './transport.provider';
import { Transport } from './transport.interface';
describe('#provideTransportLayer', () => {
- test('provides requested transport layer', () => {
- const provide = () => provideTransportLayer(Transport.AMQP, {});
- expect(provide).toBeDefined();
+ test('provides AMQP transport layer', () => {
+ pipe(
+ createContext(),
+ provideTransportLayer(Transport.AMQP),
+ E.fold(fail, transportLayer => expect(transportLayer.type).toEqual(Transport.AMQP)),
+ );
});
- test('throws an exception if requested transport layer is not supported', () => {
- const provide = () => provideTransportLayer(Transport.GRPC, {});
- expect(provide).toThrowError('Unsupported transport type');
+ test('provides REDIS transport layer', () => {
+ pipe(
+ createContext(),
+ provideTransportLayer(Transport.REDIS),
+ E.fold(fail, transportLayer => expect(transportLayer.type).toEqual(Transport.REDIS)),
+ );
+ });
+
+ test('provides LOCAL (EventBus) transport layer', async () => {
+ pipe(
+ await contextFactory(bindEagerlyTo(EventBusToken)(eventBus({ listener: messagingListener() }))),
+ provideTransportLayer(Transport.LOCAL),
+ E.fold(fail, transportLayer => expect(transportLayer.type).toEqual(Transport.LOCAL)),
+ );
+ });
+
+ test('returns error if EventBus is not registered in the context', () => {
+ pipe(
+ createContext(),
+ provideTransportLayer(Transport.LOCAL),
+ E.fold(error => expect(error.message).toEqual('Cannot provide EventBus transport layer if it is not registered'), fail),
+ );
+ });
+
+ test('returns error if EventBus is not evaluated in the context', () => {
+ pipe(
+ createContext(),
+ register(bindTo(EventBusToken)(eventBus({ listener: messagingListener() }))),
+ provideTransportLayer(Transport.LOCAL),
+ E.fold(error => expect(error.message).toEqual('Cannot provide non-evaluated EventBus transport layer'), fail),
+ );
+ });
+
+ test('returns error if unsupported transport layer type is requested', () => {
+ pipe(
+ createContext(),
+ provideTransportLayer(Transport.MQTT),
+ E.fold( error => expect(error.message).toEqual(`Unsupported transport type: "${Transport.MQTT}"`), fail),
+ );
});
});

MODIFY
packages/messaging/src/transport/transport.provider.ts
packages/messaging/src/transport/transport.provider.ts
-import { Transport } from './transport.interface';
+import * as E from 'fp-ts/lib/Either';
+import * as O from 'fp-ts/lib/Option';
+import * as M from 'fp-ts/lib/Map';
+import { pipe } from 'fp-ts/lib/pipeable';
+import { Context, ordContextToken } from '@marblejs/core';
+import { EventBusToken } from '../eventbus/messaging.eventBus.reader';
+import { Transport, TransportLayer, isTransportLayerConnection } from './transport.interface';
import { createAmqpStrategy } from './strategies/amqp.strategy';
import { createRedisStrategy } from './strategies/redis.strategy';
-import { provideLocalStrategy } from './strategies/local.strategy.provider';
-export const provideTransportLayer = <T extends Transport>(transport: T, transportOptions: any = {}) => {
+export const provideLocalTransportLayer = (context?: Context): E.Either<Error, TransportLayer<Transport>> =>
+ pipe(
+ O.fromNullable(context),
+ O.chain(ctx => M.lookup(ordContextToken)(EventBusToken, ctx)),
+ E.fromOption(() => new Error('Cannot provide EventBus transport layer if it is not registered')),
+ E.chain(dependency => !isTransportLayerConnection(dependency)
+ ? E.left(new Error('Cannot provide non-evaluated EventBus transport layer'))
+ : E.right({
+ connect: () => Promise.resolve(dependency),
+ config: dependency.config.raw,
+ type: dependency.type,
+ }),
+ ));
+
+export const provideTransportLayer = <T extends Transport>(transport: T, transportOptions: any = {}) => (context?: Context): E.Either<Error, TransportLayer<T>> => {
switch (transport) {
case Transport.AMQP:
- return createAmqpStrategy(transportOptions);
+ return E.right(createAmqpStrategy(transportOptions) as TransportLayer<T>);
case Transport.REDIS:
- return createRedisStrategy(transportOptions);
+ return E.right(createRedisStrategy(transportOptions) as TransportLayer<T>);
case Transport.LOCAL:
- return provideLocalStrategy(transportOptions);
+ return provideLocalTransportLayer(context) as E.Either<Error, TransportLayer<T>>;
default:
- throw new Error(`Unsupported transport type: ${transport}`);
+ return E.left(new Error(`Unsupported transport type: "${transport}"`));
}
};

MODIFY
packages/messaging/src/util/messaging.test.util.ts
packages/messaging/src/util/messaging.test.util.ts
import { v4 as uuid } from 'uuid';
-import { bindTo, Event, LoggerToken, mockLogger, contextFactory, Context } from '@marblejs/core';
+import { bindTo, Event, LoggerToken, mockLogger, contextFactory, Context, bindEagerlyTo, lookup, useContext } from '@marblejs/core';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { pipe } from 'fp-ts/lib/pipeable';
import { merge } from 'rxjs';
import { tap, catchError, take, toArray, filter, ignoreElements, map } from 'rxjs/operators';
import { createMicroservice } from '../server/messaging.server';
-import { TransportMessage, Transport, TransportLayerConnection } from '../transport/transport.interface';
+import { TransportMessage, Transport } from '../transport/transport.interface';
import { MsgOutputEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
import { messagingListener, MessagingListenerConfig } from '../server/messaging.server.listener';
-import { eventBus, eventBusClient } from '../eventbus/messaging.eventBus.reader';
+import { EventBusToken, eventBus } from '../eventbus/messaging.eventBus.reader';
+import { EventBusClientToken, eventBusClient } from '../eventbus/messaging.eventBusClient.reader';
import { messagingClient } from '../client/messaging.client';
import { AmqpStrategyOptions } from '../transport/strategies/amqp.strategy.interface';
import { RedisStrategyOptions } from '../transport/strategies/redis.strategy.interface';
import { MessagingClient } from '../client/messaging.client.interface';
// event bus
-export const runEventBus = (config: MessagingListenerConfig): Promise<TransportLayerConnection> =>
+export const createEventBusTestBed = async (config: MessagingListenerConfig) =>
pipe(
- createTestContext(),
- async ctx => eventBus({ listener: messagingListener(config) })(await ctx),
- );
-
-export const runEventBusClient = (): Promise<MessagingClient> =>
- pipe(
- createTestContext(),
- async ctx => eventBusClient(await ctx),
+ await contextFactory(
+ bindEagerlyTo(EventBusToken)(eventBus({ listener: messagingListener(config) })),
+ bindEagerlyTo(EventBusClientToken)(eventBusClient),
+ ),
+ lookup,
+ ask => ([useContext(EventBusToken)(ask), useContext(EventBusClientToken)(ask)] as const),
);
// microservice

MODIFY
packages/testing/src/testBed/testBedContainer.spec.ts
packages/testing/src/testBed/testBedContainer.spec.ts
@@ -12,8 +12,8 @@ describe('#createTestBedSetup cleanup container', () => {
});
const dependencies = [
- bindEagerlyTo(EventBusClientToken)(eventBusClient),
bindEagerlyTo(EventBusToken)(eventBus({ listener: messagingListener() })),
+ bindEagerlyTo(EventBusClientToken)(eventBusClient),
];
const customDependencyToken = createContextToken<CustomDependency>();

refactor(messaging): local strategy transport layer provider

+ customizable event bus timeouts
+ refactore transport layer provider layer
+ event bus no longer resides in global scope