feat(messaging): allow to pass `timeout` option to EventBus config
MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
@@ -11,6 +11,7 @@ import { EventTimerStoreToken, EventTimerStore } from '../eventStore/eventTimerS
export interface EventBusConfig {
listener: ReturnType<typeof messagingListener>;
+ timeout?: number;
}
export const EventBusToken = createContextToken<TransportLayerConnection>('EventBusToken');
@@ -20,7 +21,7 @@ export const eventBus = (config: EventBusConfig) => pipe(
R.ask<Context>(),
R.map(async derivedContext => {
const { listener } = config;
- const transportLayer = provideTransportLayer(Transport.LOCAL);
+ const transportLayer = provideTransportLayer(Transport.LOCAL, { timeout: config.timeout });
const transportLayerConnection = await transportLayer.connect();
const context = await contextFactory(

MODIFY
packages/messaging/src/transport/strategies/local.strategy.provider.ts
packages/messaging/src/transport/strategies/local.strategy.provider.ts
@@ -2,20 +2,21 @@ 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() {
- this.strategy = createLocalStrategy({});
+ private constructor(options: LocalStrategyOptions) {
+ this.strategy = createLocalStrategy(options);
this.strategyConnection = this.strategy.connect();
}
- static getDefault = () => pipe(
+ static getDefault = (options: LocalStrategyOptions) => pipe(
O.fromNullable(LocalStrategyProvider.instance),
- O.getOrElse(() => LocalStrategyProvider.instance = new LocalStrategyProvider()),
+ O.getOrElse(() => LocalStrategyProvider.instance = new LocalStrategyProvider(options)),
);
async connect() {
@@ -31,4 +32,4 @@ export class LocalStrategyProvider implements TransportLayer<Transport.LOCAL> {
}
}
-export const provideLocalStrategy = () => LocalStrategyProvider.getDefault();
+export const provideLocalStrategy = (options: LocalStrategyOptions) => LocalStrategyProvider.getDefault(options);

MODIFY
packages/messaging/src/transport/transport.provider.ts
packages/messaging/src/transport/transport.provider.ts
@@ -10,7 +10,7 @@ export const provideTransportLayer = <T extends Transport>(transport: T, transpo
case Transport.REDIS:
return createRedisStrategy(transportOptions);
case Transport.LOCAL:
- return provideLocalStrategy();
+ return provideLocalStrategy(transportOptions);
default:
throw new Error(`Unsupported transport type: ${transport}`);
}

feat(messaging): allow to pass `timeout` option to EventBus config