feat(messaging): default register `rejectUnhandled$` middleware
MODIFY
packages/core/src/index.spec.ts
packages/core/src/index.spec.ts
@@ -3,11 +3,30 @@ import * as API from './index';
describe('@marblejs/core public API', () => {
test('apis are defined', () => {
expect(API.httpListener).toBeDefined();
- expect(API.HttpError).toBeDefined();
- expect(API.CoreError).toBeDefined();
expect(API.defaultError$).toBeDefined();
expect(API.EffectFactory).toBeDefined();
expect(API.combineRoutes).toBeDefined();
+ expect(API.combineEffects).toBeDefined();
+ expect(API.combineMiddlewares).toBeDefined();
+ expect(API.createEffectContext).toBeDefined();
expect(API.event).toBeDefined();
+ expect(API.coreErrorFactory).toBeDefined();
+ expect(API.r).toBeDefined();
+
+ // errors
+ expect(API.HttpError).toBeDefined();
+ expect(API.HttpRequestError).toBeDefined();
+ expect(API.CoreError).toBeDefined();
+ expect(API.EventError).toBeDefined();
+ expect(API.isEventError).toBeDefined();
+ expect(API.isCoreError).toBeDefined();
+ expect(API.isHttpError).toBeDefined();
+ expect(API.isHttpRequestError).toBeDefined();
+
+ // operators
+ expect(API.use).toBeDefined();
+ expect(API.act).toBeDefined();
+ expect(API.switchToProtocol).toBeDefined();
+ expect(API.matchEvent).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 { flow } from 'fp-ts/lib/function';
import { deleteAt } from 'fp-ts/lib/Map';
import * as R from 'fp-ts/lib/Reader';
-import { Context, resolve, bindTo, register, setoidContextToken, createContextToken } from '@marblejs/core';
+import { Context, resolve, bindTo, setoidContextToken, createContextToken, registerAll } 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';
export interface EventBusConfig {
listener: ReturnType<typeof messagingListener>;
@@ -24,8 +24,11 @@ export const eventBus = (config: EventBusConfig) => pipe(
const ctx = deleteAt(setoidContextToken)(EventBusToken)(rootContext);
const transportLayer = provideTransportLayer(Transport.LOCAL);
const transportLayerConnection = await transportLayer.connect();
- const boundTransportLayer = bindTo(TransportLayerToken)(() => transportLayer);
- const context = await flow(register(boundTransportLayer), resolve)(ctx);
+ const registerDependencies = registerAll([
+ bindTo(TransportLayerToken)(() => transportLayer),
+ bindTo(EventTimerStoreToken)(EventTimerStore),
+ ]);
+ const context = await pipe(ctx, registerDependencies, resolve);
listener(context)(transportLayerConnection);

MODIFY
packages/messaging/src/index.spec.ts
packages/messaging/src/index.spec.ts
@@ -9,5 +9,11 @@ describe('@marblejs/messaging', () => {
expect(API.messagingClient).toBeDefined();
expect(API.messagingListener).toBeDefined();
expect(API.reply).toBeDefined();
+ expect(API.ackEvent).toBeDefined();
+ expect(API.nackEvent).toBeDefined();
+ expect(API.nackAndResendEvent).toBeDefined();
+ expect(API.EVENT_BUS_CHANNEL).toBeDefined();
+ expect(API.AmqpConnectionStatus).toBeDefined();
+ expect(API.RedisConnectionStatus).toBeDefined();
});
});

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
@@ -10,7 +10,6 @@ export { LocalStrategyOptions, EVENT_BUS_CHANNEL } from './transport/strategies/
// effects, middlewares
export * from './effects/messaging.effects.interface';
-export * from './middlewares/messaging.ack.middleware';
// handy functions
export { reply } from './reply/reply';

MODIFY
packages/messaging/src/middlewares/messaging.ack.middleware.ts
packages/messaging/src/middlewares/messaging.ack.middleware.ts
@@ -5,10 +5,9 @@ import { flow } from 'fp-ts/lib/function';
import { pipe } from 'fp-ts/lib/pipeable';
import { MsgEffect } from '../effects/messaging.effects.interface';
import { EventTimerStoreToken } from '../eventStore/eventTimerStore';
-import { Transport } from '../transport/transport.interface';
+import { Transport, TransportLayerConnection } from '../transport/transport.interface';
import { nackEvent } from '../ack/ack';
-
-const ACK_TRANSPORTS = [Transport.AMQP];
+import { AmqpStrategyOptions } from '../transport/strategies/amqp.strategy.interface';
/**
* Automatically tries to reject all unhandled events when `timeout` defined by the transport layer options occurs.
@@ -21,8 +20,14 @@ export const rejectUnhandled$: MsgEffect = (event$, ctx) => {
const logger = useContext(LoggerToken)(ctx.ask);
const timeout = ctx.client.config.timeout;
- const isAckTransport = (type: Transport) =>
- ACK_TRANSPORTS.includes(type);
+ const isAckModeSupportedAndEnabled = (conn: TransportLayerConnection): boolean => {
+ switch (conn.type) {
+ case Transport.AMQP:
+ return !!(conn.config.raw as AmqpStrategyOptions).expectAck;
+ default:
+ return false;
+ }
+ }
const logRejection = pipe(
logger({
@@ -40,7 +45,7 @@ export const rejectUnhandled$: MsgEffect = (event$, ctx) => {
return event$.pipe(
tap(event => {
- if (isAckTransport(ctx.client.type)) {
+ if (isAckModeSupportedAndEnabled(ctx.client)) {
const handler = handleEventRejection(event);
eventTimerStore.register(timeout)(handler)(event)();
}

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -18,6 +18,7 @@ import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect, MsgOutputEffect } from
import { inputLogger$, outputLogger$, errorLogger$ } from '../middlewares/messaging.eventLogger.middleware';
import { outputRouter$, outputErrorEncoder$ } from '../middlewares/messaging.eventOutput.middleware';
import { idApplier$ } from '../middlewares/messaging.eventInput.middleware';
+import { rejectUnhandled$ } from '../middlewares/messaging.ack.middleware';
export interface MessagingListenerConfig {
effects?: MsgEffect[];
@@ -53,7 +54,7 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
const logger = useContext(LoggerToken)(ask);
const combinedEffects = combineEffects(...effects);
- const combinedMiddlewares = combineMiddlewares(idApplier$, inputLogger$, ...middlewares);
+ const combinedMiddlewares = combineMiddlewares(idApplier$, rejectUnhandled$, inputLogger$, ...middlewares);
return connection => {
const errorSubject = new Subject<Error>();

MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
@@ -5,7 +5,6 @@ import { forkJoin, TimeoutError } from 'rxjs';
import { map, tap, delay, mapTo } from 'rxjs/operators';
import { TransportLayerConnection } from '../../transport/transport.interface';
import { MsgEffect } from '../../effects/messaging.effects.interface';
-import { rejectUnhandled$ } from '../../middlewares/messaging.ack.middleware';
import { MessagingClient } from '../../client/messaging.client.interface';
import { reply } from '../../reply/reply';
import * as Util from '../../util/messaging.test.util';
@@ -138,7 +137,7 @@ describe('messagingServer::AMQP', () => {
)(done);
// when
- microservice = await Util.createAmqpMicroservice(optionsMicroservice)({ middlewares: [rejectUnhandled$], effects: [ack$], output$ });
+ microservice = await Util.createAmqpMicroservice(optionsMicroservice)({ effects: [ack$], output$ });
client = await Util.createAmqpClient(optionsClient);
await client.emit({ type: 'ACK', payload: 1 });
@@ -164,7 +163,7 @@ describe('messagingServer::AMQP', () => {
)(done);
// when
- microservice = await Util.createAmqpMicroservice(optionsMicroservice)({ middlewares: [rejectUnhandled$], effects: [ack$], output$ });
+ microservice = await Util.createAmqpMicroservice(optionsMicroservice)({ effects: [ack$], output$ });
client = await Util.createAmqpClient(optionsClient);
await client.emit({ type: 'TEST' });

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, createContext, Event, register, LoggerToken, mockLogger } from '@marblejs/core';
+import { bindTo, Event, LoggerToken, mockLogger, contextFactory, Context } from '@marblejs/core';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { pipe } from 'fp-ts/lib/pipeable';
-import { flow } from 'fp-ts/lib/function';
import { merge } from 'rxjs';
import { tap, catchError, take, toArray, filter, ignoreElements, map } from 'rxjs/operators';
import { createMicroservice } from '../server/messaging.server';
-import { TransportMessage, Transport } from '../transport/transport.interface';
+import { TransportMessage, Transport, TransportLayerConnection } 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 { 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) =>
+export const runEventBus = (config: MessagingListenerConfig): Promise<TransportLayerConnection> =>
pipe(
createTestContext(),
- eventBus({ listener: messagingListener(config) }),
+ async ctx => eventBus({ listener: messagingListener(config) })(await ctx),
);
-export const runEventBusClient = () =>
+export const runEventBusClient = (): Promise<MessagingClient> =>
pipe(
createTestContext(),
- eventBusClient,
+ async ctx => eventBusClient(await ctx),
);
// microservice
@@ -33,10 +33,10 @@ export const createTestMicroservice = (transport: any, options: any) => async (c
return (await createMicroservice({ options, transport, listener }))();
};
-export const createTestClient = (transport: any, options: any) =>
+export const createTestClient = (transport: any, options: any): Promise<MessagingClient> =>
pipe(
createTestContext(),
- messagingClient({ transport, options: { ...options, expectAck: false } }),
+ async ctx => messagingClient({ transport, options: { ...options, expectAck: false } })(await ctx),
);
export const createAmqpClient = (options: AmqpStrategyOptions) =>
@@ -71,10 +71,9 @@ export const createMessage = (data: any, correlationId?: string): TransportMessa
correlationId: correlationId ?? uuid(),
});
-export const createTestContext = () => {
- const boundLogger = bindTo(LoggerToken)(mockLogger);
- return flow(register(boundLogger))(createContext());
-};
+export const createTestContext = (): Promise<Context> => contextFactory(
+ bindTo(LoggerToken)(mockLogger),
+);
export const assertOutputEvent = (...outEvent: Event[]) => (done: jest.DoneCallback): MsgOutputEffect => event$ => {
const isAssertionEvent = (event: Event) =>

feat(messaging): default register `rejectUnhandled$` middleware