refactor(messaging): deprecated event bus and messaging client aliases
DELETE
packages/messaging/src/client/messaging.client.interface.ts
-import { Event } from '@marblejs/core';
-import { Observable } from 'rxjs';
-import { TransportMessageTransformer, TransportStrategy } from '../transport/transport.interface';
-
-export interface MessagingClient {
- send: <T extends Event = Event, U extends Event = Event>(data: U) => Observable<T>;
- emit: <T extends Event = Event>(data: T) => Promise<void>;
- close: () => Promise<void>;
-}
-
-type ConfigurationBase = {
- msgTransformer?: TransportMessageTransformer;
-}
-
-export type MessagingClientConfig =
- & TransportStrategy
- & ConfigurationBase
- ;

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
@@ -19,10 +19,24 @@ import {
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 { TransportMessage, DEFAULT_TIMEOUT } from '../transport/transport.interface';
+import { TransportMessage, DEFAULT_TIMEOUT, TransportMessageTransformer, TransportStrategy } from '../transport/transport.interface';
import { provideTransportLayer } from '../transport/transport.provider';
import { jsonTransformer } from '../transport/transport.transformer';
-import { MessagingClient, MessagingClientConfig } from './messaging.client.interface';
+
+export interface MessagingClient {
+ send: <T extends Event = Event, U extends Event = Event>(data: U) => Observable<T>;
+ emit: <T extends Event = Event>(data: T) => Promise<void>;
+ close: () => Promise<void>;
+}
+
+type ConfigurationBase = {
+ msgTransformer?: TransportMessageTransformer;
+}
+
+export type MessagingClientConfig =
+ & TransportStrategy
+ & ConfigurationBase
+ ;
const getHttpModule: T.Task<O.Some<typeof import('@marblejs/http')> | O.None> =
pipe(
@@ -30,7 +44,12 @@ const getHttpModule: T.Task<O.Some<typeof import('@marblejs/http')> | O.None> =
TE.map(O.some),
TE.getOrElseW(constant(T.of(O.none))));
-export const messagingClient = (config: MessagingClientConfig) => {
+/**
+ * @param config `MessagingClientConfig`
+ * @returns asynchronous reader of `MessagingClient`
+ * @since v3.0
+ */
+export const MessagingClient = (config: MessagingClientConfig) => {
const {
transport,
options,
@@ -114,3 +133,11 @@ export const messagingClient = (config: MessagingClientConfig) => {
};
}));
};
+
+/**
+ * An alias for `MessagingClient`
+ *
+ * @deprecated since version `v4.0`. Use `MessagingClient` instead.
+ * Will be removed in version `v5.0`
+ */
+export const messagingClient = MessagingClient;

MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.spec.ts
@@ -8,8 +8,8 @@ import { MsgEffect } from '../effects/messaging.effects.interface';
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';
+import { EventBusToken, EventBus } from './messaging.eventBus.reader';
+import { EventBusClientToken, EventBusClient } from './messaging.eventBusClient.reader';
describe('#eventBus', () => {
test('handles RPC event', async () => {
@@ -227,8 +227,8 @@ describe('#eventBus', () => {
const ask = pipe(
await contextFactory(
- bindEagerlyTo(EventBusToken)(eventBus({ listener, timeout })),
- bindEagerlyTo(EventBusClientToken)(eventBusClient),
+ bindEagerlyTo(EventBusToken)(EventBus({ listener, timeout })),
+ bindEagerlyTo(EventBusClientToken)(EventBusClient),
),
lookup,
);
@@ -250,8 +250,8 @@ describe('#eventBus', () => {
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() })),
+ bindEagerlyTo(EventBusClientToken)(EventBusClient),
+ bindEagerlyTo(EventBusToken)(EventBus({ listener: messagingListener() })),
);
await expect(context).resolves.toBeDefined();

MODIFY
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
packages/messaging/src/eventbus/messaging.eventBus.reader.ts
@@ -8,7 +8,7 @@ import { Transport, TransportLayerConnection } from '../transport/transport.inte
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 { MessagingClient } from '../client/messaging.client';
import { EventBusClientToken } from './messaging.eventBusClient.reader';
export interface EventBusConfig {
@@ -22,7 +22,15 @@ export interface EventBus extends TransportLayerConnection<Transport.LOCAL> {
export const EventBusToken = createContextToken<EventBus>('EventBus');
-export const eventBus = (config: EventBusConfig) => pipe(
+/**
+ * `EventBus` has to be registered eagerly before `EventBusClient` reader
+ *
+ * @param config `EventBusConfig`
+ * @returns asynchronous reader of `EventBus`
+ * @since v3.0
+ */
+export const EventBus = (config: EventBusConfig) =>
+ pipe(
R.ask<Context>(),
R.map<Context, Promise<EventBus>>(async derivedContext => {
const { listener } = config;
@@ -37,7 +45,7 @@ export const eventBus = (config: EventBusConfig) => pipe(
const internalMessagingClient = await pipe(
await contextFactory(bindEagerlyTo(EventBusToken)(() => transportLayerConnection)),
- messagingClient({ transport: Transport.LOCAL, options: {} }),
+ MessagingClient({ transport: Transport.LOCAL, options: {} }),
);
const context = await pipe(
@@ -61,3 +69,11 @@ export const eventBus = (config: EventBusConfig) => pipe(
return eventBus;
}),
);
+
+/**
+ * An alias for `EventBus`
+ *
+ * @deprecated since version `v4.0`. Use `EventBusClient` instead.
+ * Will be removed in version `v5.0`
+ */
+ export const eventBus = EventBus;

MODIFY
packages/messaging/src/eventbus/messaging.eventBusClient.reader.ts
packages/messaging/src/eventbus/messaging.eventBusClient.reader.ts
import { createContextToken, createReader, useContext, LoggerToken, LoggerTag, LoggerLevel } from '@marblejs/core';
-import { MessagingClient } from '../client/messaging.client.interface';
+import { MessagingClient } from '../client/messaging.client';
export interface EventBusClient extends MessagingClient {}
export const EventBusClientToken = createContextToken<EventBusClient>('EventBusClient');
-export const eventBusClient = createReader(ask => {
+/**
+ * `EventBusClient` has to be registered eagerly after main `EventBus`
+ * @returns asynchronous reader of `EventBus`
+ * @since v3.0
+ */
+export const EventBusClient = createReader(ask => {
const logger = useContext(LoggerToken)(ask);
const logWarning = logger({
tag: LoggerTag.EVENT_BUS,
level: LoggerLevel.WARN,
type: 'eventBusClient',
- message: '"EventBusClient" requires to be registered eagerly before main "EventBus" reader.',
+ message: '"EventBus" requires to be registered eagerly before "EventBusClient" reader.',
});
logWarning();
});
+
+/**
+ * An alias for `EventBusClient`
+ *
+ * @deprecated since version `v4.0`. Use `EventBusClient` instead.
+ * Will be removed in version `v5.0`
+ */
+export const eventBusClient = EventBusClient;

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

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
// client
export * from './client/messaging.client';
-export * from './client/messaging.client.interface';
// transport
export * from './transport/transport.interface';

MODIFY
packages/messaging/src/middlewares/messaging.eventLogger.middleware.spec.ts
packages/messaging/src/middlewares/messaging.eventLogger.middleware.spec.ts
@@ -10,11 +10,10 @@ import {
EventError,
} from '@marblejs/core';
import { Marbles } from '@marblejs/core/dist/+internal/testing';
-import { constUndefined } from 'fp-ts/lib/function';
-import { pipe } from 'fp-ts/lib/pipeable';
import { io } from 'fp-ts/lib/IO';
+import { constUndefined, pipe } from 'fp-ts/lib/function';
import { messagingListener } from '../server/messaging.server.listener';
-import { eventBus, EventBus } from '../eventbus/messaging.eventBus.reader';
+import { EventBus } from '../eventbus/messaging.eventBus.reader';
import { inputLogger$, outputLogger$, exceptionLogger$ } from './messaging.eventLogger.middleware';
describe('inputLogger$', () => {
@@ -26,7 +25,7 @@ describe('inputLogger$', () => {
const provideEventBus = async (logger: any) => pipe(
await contextFactory(bindEagerlyTo(LoggerToken)(createReader(() => logger))),
- eventBus({ listener: messagingListener() }));
+ EventBus({ listener: messagingListener() }));
const provideEffectContextForLogger = async (logger: any) => pipe(
await provideEventBus(logger),

MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
@@ -6,10 +6,10 @@ import { map, tap, delay, mapTo } from 'rxjs/operators';
import { flow } from 'fp-ts/lib/function';
import { TransportLayerConnection } from '../../transport/transport.interface';
import { MsgEffect } from '../../effects/messaging.effects.interface';
-import { MessagingClient } from '../../client/messaging.client.interface';
import { reply } from '../../reply/reply';
import * as Util from '../../util/messaging.test.util';
import { ackEvent } from '../../ack/ack';
+import { MessagingClient } from '../../client/messaging.client';
describe('messagingServer::AMQP', () => {
let client: MessagingClient;

MODIFY
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
@@ -6,10 +6,10 @@ import { map, tap, delay, mapTo } from 'rxjs/operators';
import { flow } from 'fp-ts/lib/function';
import { TransportLayerConnection } from '../../transport/transport.interface';
import { MsgEffect } from '../../effects/messaging.effects.interface';
-import { MessagingClient } from '../../client/messaging.client.interface';
import { reply } from '../../reply/reply';
import * as Util from '../../util/messaging.test.util';
import { ackEvent } from '../../ack/ack';
+import { MessagingClient } from '../../client/messaging.client';
describe('messagingServer::Redis', () => {
let client: MessagingClient;

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 { pipe } from 'fp-ts/lib/function';
import { createContext, contextFactory, bindEagerlyTo, bindTo, register } from '@marblejs/core';
-import { EventBusToken, eventBus } from '../eventbus/messaging.eventBus.reader';
+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';
@@ -25,7 +25,7 @@ describe('#provideTransportLayer', () => {
test('provides LOCAL (EventBus) transport layer', async () => {
pipe(
- await contextFactory(bindEagerlyTo(EventBusToken)(eventBus({ listener: messagingListener() }))),
+ await contextFactory(bindEagerlyTo(EventBusToken)(EventBus({ listener: messagingListener() }))),
provideTransportLayer(Transport.LOCAL),
E.fold(fail, transportLayer => expect(transportLayer.type).toEqual(Transport.LOCAL)),
);
@@ -42,7 +42,7 @@ describe('#provideTransportLayer', () => {
test('returns error if EventBus is not evaluated in the context', () => {
pipe(
createContext(),
- register(bindTo(EventBusToken)(eventBus({ listener: messagingListener() }))),
+ register(bindTo(EventBusToken)(EventBus({ listener: messagingListener() }))),
provideTransportLayer(Transport.LOCAL),
E.fold(error => expect(error.message).toEqual('Cannot provide non-evaluated EventBus transport layer'), fail),
);

MODIFY
packages/messaging/src/util/messaging.test.util.ts
packages/messaging/src/util/messaging.test.util.ts
@@ -8,22 +8,24 @@ import { createMicroservice } from '../server/messaging.server';
import { TransportMessage, Transport } from '../transport/transport.interface';
import { MsgOutputEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
import { messagingListener, MessagingListenerConfig } from '../server/messaging.server.listener';
-import { EventBusToken, eventBus } from '../eventbus/messaging.eventBus.reader';
-import { EventBusClientToken, eventBusClient } from '../eventbus/messaging.eventBusClient.reader';
-import { messagingClient } from '../client/messaging.client';
+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 createEventBusTestBed = async (config: MessagingListenerConfig) =>
pipe(
await contextFactory(
- bindEagerlyTo(EventBusToken)(eventBus({ listener: messagingListener(config) })),
- bindEagerlyTo(EventBusClientToken)(eventBusClient),
+ bindEagerlyTo(EventBusToken)(EventBus({ listener: messagingListener(config) })),
+ bindEagerlyTo(EventBusClientToken)(EventBusClient),
),
lookup,
- ask => ([useContext(EventBusToken)(ask), useContext(EventBusClientToken)(ask)] as const),
+ ask => ([
+ useContext(EventBusToken)(ask),
+ useContext(EventBusClientToken)(ask),
+ ] as const),
);
// microservice
@@ -35,7 +37,7 @@ export const createTestMicroservice = (transport: any, options: any) => async (c
export const createTestClient = (transport: any, options: any): Promise<MessagingClient> =>
pipe(
createTestContext(),
- async ctx => messagingClient({ transport, options: { ...options, expectAck: false } })(await ctx),
+ async ctx => MessagingClient({ transport, options: { ...options, expectAck: false } })(await ctx),
);
export const createAmqpClient = (options: AmqpStrategyOptions) =>

MODIFY
packages/testing/src/testBed/testBedContainer.spec.ts
packages/testing/src/testBed/testBedContainer.spec.ts
import { bindEagerlyTo, useContext, bindTo, createContextToken } from '@marblejs/core';
import { httpListener } from '@marblejs/http';
-import { EventBusClientToken, eventBusClient, EventBusToken, eventBus, messagingListener } from '@marblejs/messaging';
+import { EventBusClientToken, EventBusClient, EventBusToken, EventBus, messagingListener } from '@marblejs/messaging';
import { createHttpTestBed } from './http/http.testBed';
import { createTestBedSetup } from './testBedSetup';
import { DependencyCleanup } from './testBedContainer.interface';
@@ -13,8 +13,8 @@ describe('#createTestBedSetup cleanup container', () => {
});
const dependencies = [
- bindEagerlyTo(EventBusToken)(eventBus({ listener: messagingListener() })),
- bindEagerlyTo(EventBusClientToken)(eventBusClient),
+ bindEagerlyTo(EventBusToken)(EventBus({ listener: messagingListener() })),
+ bindEagerlyTo(EventBusClientToken)(EventBusClient),
];
const customDependencyToken = createContextToken<CustomDependency>();

refactor(messaging): deprecated event bus and messaging client aliases