fix(messaging): eventBus - register internal client
ADD
packages/@integration/src/cqrs/__mock__/__mock__dependencies.ts
+import * as assert from 'assert';
+import { useContext, createContextToken, createReader } from '@marblejs/core';
+import { EventBusClientToken } from '@marblejs/messaging';
+
+export const SomeDependencyToken = createContextToken('SomeDependency');
+
+/**
+ * @description Some random dependency used for checking if EventBusClient can be properly resolved
+ */
+export const SomeDependency = createReader(ask => {
+ const eventBusClient = useContext(EventBusClientToken)(ask);
+
+ assert.strictEqual(typeof eventBusClient.emit, 'function');
+
+ return 'nothing special';
+});

MODIFY
packages/@integration/src/cqrs/effects/eventbus.effects.ts
packages/@integration/src/cqrs/effects/eventbus.effects.ts
+/* eslint-disable @typescript-eslint/no-unused-vars */
+
import { of } from 'rxjs';
import { map, tap, delay } from 'rxjs/operators';
import { pipe } from 'fp-ts/lib/pipeable';
import { act, matchEvent, useContext, LoggerToken, LoggerTag } from '@marblejs/core';
-import { MsgEffect } from '@marblejs/messaging';
+import { MsgEffect, EventBusClientToken } from '@marblejs/messaging';
import { GenerateOfferDocumentCommand } from '../domain/Offer.command';
import { OfferDocumentSavedEvent, OfferDocumentCreatedEvent } from '../domain/Offer.event';
+import { SomeDependencyToken } from '../__mock__/__mock__dependencies';
const tag = LoggerTag.EVENT_BUS;
export const generateOfferDocument$: MsgEffect = (event$, ctx) => {
const logger = useContext(LoggerToken)(ctx.ask);
+ const _1 = useContext(SomeDependencyToken)(ctx.ask);
+ const _2 = useContext(EventBusClientToken)(ctx.ask);
return event$.pipe(
matchEvent(GenerateOfferDocumentCommand),

MODIFY
packages/@integration/src/cqrs/index.ts
packages/@integration/src/cqrs/index.ts
import * as T from 'fp-ts/lib/Task';
import { pipe } from 'fp-ts/lib/pipeable';
import { isTestEnv, getPortEnv } from '@marblejs/core/dist/+internal/utils';
-import { httpListener, createServer, bindEagerlyTo } from '@marblejs/core';
+import { httpListener, createServer, bindEagerlyTo, bindTo } from '@marblejs/core';
import { messagingListener, EventBusClientToken, eventBusClient, EventBusToken, eventBus } from '@marblejs/messaging';
import { bodyParser$ } from '@marblejs/middleware-body';
import { logger$ } from '@marblejs/middleware-logger';
import { postDocumentsGenerate$ } from './effects/http.effects';
import { generateOfferDocument$, offerDocumentCreated$ } from './effects/eventbus.effects';
+import { SomeDependencyToken, SomeDependency } from './__mock__/__mock__dependencies';
const eventBusListener = messagingListener({
effects: [
@@ -31,6 +32,7 @@ export const server = () => createServer({
dependencies: [
bindEagerlyTo(EventBusToken)(eventBus({ listener: eventBusListener })),
bindEagerlyTo(EventBusClientToken)(eventBusClient),
+ bindTo(SomeDependencyToken)(SomeDependency),
],
});

MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
@@ -242,12 +242,12 @@ describe('#eventBus', () => {
await eventBusClientInstance.close();
});
- test('fails if event bus client is registered before main EventBus reader', async () => {
+ test('doesn\'t fail 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'));
+ await expect(context).resolves.toBeDefined();
});
});

MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
-import { pipe } from 'fp-ts/lib/pipeable';
import * as R from 'fp-ts/lib/Reader';
-import { Context, bindTo, createContextToken, contextFactory, bindEagerlyTo, DerivedContextToken, logContext, LoggerTag } from '@marblejs/core';
+import * as M from 'fp-ts/lib/Map';
+import * as O from 'fp-ts/lib/Option';
+import { pipe, constant } from 'fp-ts/lib/function';
+import { Context, bindTo, contextFactory, bindEagerlyTo, DerivedContextToken, logContext, LoggerTag, ordContextToken, createContextToken } from '@marblejs/core';
import { TransportLayerToken } from '../server/messaging.server.tokens';
import { Transport, TransportLayerConnection } from '../transport/transport.interface';
import { messagingListener } from '../server/messaging.server.listener';
import { EventTimerStoreToken, EventTimerStore } from '../eventStore/eventTimerStore';
import { createLocalStrategy } from '../transport/strategies/local.strategy';
+import { messagingClient } from '../client/messaging.client';
+import { EventBusClientToken } from './messaging.eventBusClient.reader';
export interface EventBusConfig {
listener: ReturnType<typeof messagingListener>;
@@ -25,10 +29,25 @@ export const eventBus = (config: EventBusConfig) => pipe(
const transportLayer = createLocalStrategy(config);
const transportLayerConnection = await transportLayer.connect();
- const context = await contextFactory(
+ const setEventBusClientInDerivedContext = (derivedContext: Context): Context =>
+ pipe(
+ M.lookup(ordContextToken)(EventBusClientToken)(derivedContext),
+ O.fold(constant(derivedContext), () => derivedContext.set(EventBusClientToken, internalMessagingClient)),
+ );
+
+ const internalMessagingClient = await pipe(
+ await contextFactory(bindEagerlyTo(EventBusToken)(() => transportLayerConnection)),
+ messagingClient({ transport: Transport.LOCAL, options: {} }),
+ );
+
+ const context = await pipe(
+ setEventBusClientInDerivedContext(derivedContext),
+ derivedContext => contextFactory(
bindEagerlyTo(DerivedContextToken)(() => derivedContext),
+ bindEagerlyTo(EventBusClientToken)(() => internalMessagingClient),
bindTo(TransportLayerToken)(() => transportLayer),
bindTo(EventTimerStoreToken)(EventTimerStore),
+ ),
);
logContext(LoggerTag.EVENT_BUS)(context);

MODIFY
packages/messaging/src/eventbus/messaging.eventBusClient.reader.ts
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 { createContextToken, createReader, useContext, LoggerToken, LoggerTag, LoggerLevel } from '@marblejs/core';
import { MessagingClient } from '../client/messaging.client.interface';
-import { messagingClient } from '../client/messaging.client';
-export const EventBusClientToken = createContextToken<MessagingClient>('EventBusClient');
+export interface EventBusClient extends MessagingClient {}
-export const eventBusClient = pipe(
- R.ask<Context>(),
- R.map(messagingClient({
- transport: Transport.LOCAL,
- options: {},
- })),
-);
+export const EventBusClientToken = createContextToken<EventBusClient>('EventBusClient');
+
+export const eventBusClient = createReader(ask => {
+ const logger = useContext(LoggerToken)(ask);
+
+ logger({
+ tag: LoggerTag.EVENT_BUS,
+ level: LoggerLevel.WARN,
+ type: 'eventBusClient',
+ message: '"EventBusClient" requires to be registered eagerly before main "EventBus" reader.',
+ })();
+});

MODIFY
packages/testing/src/testBed/testBedContainer.spec.ts
packages/testing/src/testBed/testBedContainer.spec.ts
@@ -41,7 +41,7 @@ describe('#createTestBedSetup cleanup container', () => {
await cleanup();
- expect(eventBusSpy).toHaveBeenCalledTimes(2);
+ expect(eventBusSpy).toHaveBeenCalledTimes(1);
expect(eventBusClientSpy).toHaveBeenCalledTimes(1);
});

fix(messaging): eventBus - register internal client