imp(messaging): continuous msg stream + error handling imp + nack/ack imp
MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
@@ -14,6 +14,9 @@ import {
import { merge } from 'rxjs';
import { map, tap, filter, distinctUntilChanged } from 'rxjs/operators';
+const log = (tag: string) => (data: any) =>
+ console.log(tag, data);
+
const fibonacci = (n: number): number =>
n === 0 || n === 1
? n
@@ -21,7 +24,7 @@ const fibonacci = (n: number): number =>
const log$: MsgMiddlewareEffect = event$ =>
event$.pipe(
- tap(event => console.log('server ::', event)),
+ tap(({ type, payload }) => log('INPUT')({ type, payload })),
);
const fibonacci$: MsgEffect = event$ =>
@@ -59,7 +62,9 @@ const error$: MsgServerEffect = event$ =>
const output$: MsgOutputEffect = event$ =>
event$.pipe(
- tap(({ event, initiator }) => console.log(`processed ::`, event, {
+ tap(({ event, initiator }) => log('OUTPUT')({
+ event: event.type,
+ payload: event.payload,
replyTo: initiator.replyTo,
correlationId: initiator.correlationId,
})),

MODIFY
packages/@integration/test/messaging.integration.spec.ts
packages/@integration/test/messaging.integration.spec.ts
@@ -5,6 +5,8 @@ import { microservice as messagingServer } from '../src/messaging/server';
const microservice = messagingServer();
beforeAll(() => microservice);
+beforeAll(() => jest.spyOn(console, 'log').mockImplementation());
+beforeAll(() => jest.spyOn(console, 'info').mockImplementation());
describe('messaging integration', () => {
test('GET /fib returns 10, 11, 12, 13, 14 th fibonacci number', async () =>

MODIFY
packages/core/src/event/event.interface.ts
packages/core/src/event/event.interface.ts
export type EventType = string;
-export interface Event<T = unknown, U = any, V extends EventType = EventType> {
- type: V;
- payload?: T;
- error?: U;
+export interface Event<P = unknown, E = any, T extends EventType = EventType> {
+ type: T;
+ payload?: P;
+ error?: E;
+ raw?: any;
}
-export interface ValidatedEvent<T = unknown, V extends EventType = EventType> {
- type: V;
- payload: T;
+export interface ValidatedEvent<P = unknown, T extends EventType = EventType> {
+ type: T;
+ payload: P;
+ raw?: any;
}
export type EventsUnion<A extends {

MODIFY
packages/messaging/src/effects/messaging.effects.interface.ts
packages/messaging/src/effects/messaging.effects.interface.ts
@@ -11,7 +11,7 @@ export interface MsgMiddlewareEffect<
export interface MsgErrorEffect<
I = Event,
Err extends Error = Error,
-> extends MsgEffect<{ event: I; error: Err }, Event, MsgClient> {}
+> extends MsgEffect<{ event?: I; error: Err }, any, MsgClient> {}
export interface MsgEffect<
I = Event,

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
import {
reader,
+ Event,
ContextProvider,
combineMiddlewares,
combineEffects,
createEffectContext,
useContext,
+ EventError,
} from '@marblejs/core';
import * as O from 'fp-ts/lib/Option';
import * as R from 'fp-ts/lib/Reader';
import { pipe } from 'fp-ts/lib/pipeable';
import { Observable, Subscription, Subject, of } from 'rxjs';
-import { map, publish, withLatestFrom, takeUntil, catchError, mergeMap, take } from 'rxjs/operators';
+import { map, publish, withLatestFrom, takeUntil, catchError, take } from 'rxjs/operators';
import {
TransportMessage,
TransportMessageTransformer,
@@ -31,7 +33,7 @@ export interface MessagingListenerConfig {
}
const defaultOutput$: MsgOutputEffect = msg$ => msg$.pipe(map(m => m.event));
-const defaultError$: MsgErrorEffect = msg$ => msg$.pipe(map(m => m.event));
+const defaultError$: MsgErrorEffect = msg$ => msg$;
export const messagingListener = (config: MessagingListenerConfig = {}) => {
const {
@@ -54,9 +56,13 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const combinedMiddlewares = combineMiddlewares(...middlewares);
const ctx = createEffectContext({ ask, client: conn });
+ const decode = (msg: TransportMessage<Buffer>): TransportMessage<Event> => ({
+ ...msg,
+ data: { ...msgTransformer.decode(msg.data), raw: msg.raw },
+ });
+
const message$ = conn.message$.pipe(
- map(msg => ({ ...msg, data: msgTransformer.decode(msg.data) } as TransportMessage<any>)),
- mergeMap(msg => of(msg).pipe(
+ map(decode),
publish(msg$ => combinedMiddlewares(msg$.pipe(map(m => m.data)), ctx).pipe(
withLatestFrom(msg$),
)),
@@ -69,10 +75,9 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
withLatestFrom(msg$),
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- catchError(error => error$(of({ event: msg.data, error }), ctx).pipe(
- map(data => ({ ...msg, data } as TransportMessage<any>)),
+ catchError((error: EventError) => error$(of({ event: error.event, error }), ctx).pipe(
+ map(data => ({ data } as TransportMessage<any>)),
)),
- ))
);
const onSubscribeEffectsOutput = (conn: TransportLayerConnection) => (msg: TransportMessage<any>) => {

MODIFY
packages/messaging/src/server/messaging.server.spec.ts
packages/messaging/src/server/messaging.server.spec.ts
-import { Event, matchEvent, use } from '@marblejs/core';
+import { Event, matchEvent, use, EventError, createEvent } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { map, tap, mergeMapTo } from 'rxjs/operators';
import { createMicroservice } from './messaging.server';
@@ -7,31 +7,33 @@ import { Transport, TransportMessage } from '../transport/transport.interface';
import { createAmqpStrategy } from '../transport/strategies/amqp.strategy';
import { MsgEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
import { Subject, throwError } from 'rxjs';
+import { AmqpStrategyOptions } from '../transport/strategies/amqp.strategy.interface';
describe('messagingServer', () => {
describe('AMQP', () => {
- const options = {
+ const createOptions = (config: { expectAck?: boolean; queue?: string } = {}): AmqpStrategyOptions => ({
host: 'amqp://localhost:5672',
- queue: 'test_queue_server',
+ queue: config.queue || 'test_queue_server',
+ expectAck: config.expectAck,
queueOptions: { durable: false },
- };
+ });
- const runServer = (effect$?: MsgEffect, error$?: MsgErrorEffect) =>
+ const runServer = (options: AmqpStrategyOptions) => (effect$?: MsgEffect, error$?: MsgErrorEffect) =>
createMicroservice({
options,
transport: Transport.AMQP,
messagingListener: messagingListener(effect$ ? { effects: [effect$], error$ } : undefined),
})();
- const runClient = () =>
+ const runClient = (options: AmqpStrategyOptions) =>
createAmqpStrategy(options).connect({ isConsumer: false });
const createMessage = (data: any): TransportMessage<Buffer> => ({
data: Buffer.from(JSON.stringify(data)),
});
- test('receives RPC response from consumer', async () => {
+ test('handles RPC event', async () => {
const rpc$: MsgEffect = event$ =>
event$.pipe(
matchEvent('RPC_TEST'),
@@ -40,8 +42,9 @@ describe('messagingServer', () => {
map(payload => ({ type: 'RPC_TEST_RESULT', payload: payload + 1 })),
);
- const client = await runClient();
- const server = await runServer(rpc$);
+ const options = createOptions({ expectAck: false });
+ const client = await runClient(options);
+ const server = await runServer(options)(rpc$);
const message = createMessage({ type: 'RPC_TEST', payload: 1 });
const result = await client.sendMessage(options.queue, message);
@@ -52,7 +55,7 @@ describe('messagingServer', () => {
await server.close();
});
- test('emits event to consumer', async done => {
+ test('handles published event', async done => {
const eventSubject = new Subject();
eventSubject.subscribe(event => {
@@ -69,8 +72,9 @@ describe('messagingServer', () => {
tap(event => eventSubject.next(event)),
);
- const server = await runServer(event$);
- const client = await runClient();
+ const options = createOptions({ expectAck: false });
+ const client = await runClient(options);
+ const server = await runServer(options)(event$);
const message = createMessage({ type: 'EVENT_TEST', payload: 1 });
const emitResult = await client.emitMessage(options.queue, message);
@@ -78,33 +82,93 @@ describe('messagingServer', () => {
expect(emitResult).toEqual(true);
});
- test('reacts to thrown error', async done => {
- const expectedMessage = { type: 'EVENT_TEST' };
- const expectedError = new Error('test_error');
- const errorSubject = new Subject<[Event, Error | undefined]>();
+ test('acks processed event', async done => {
+ const eventSubject = new Subject();
+
+ eventSubject.subscribe(event => {
+ expect(event).toEqual({ type: 'EVENT_TEST_RESPONSE', payload: 2 });
+ setTimeout(() => server.close().then(done), 1000);
+ });
+
+ const event$: MsgEffect = (event$, { client }) =>
+ event$.pipe(
+ matchEvent('EVENT_TEST'),
+ use(eventValidator$(t.number)),
+ map(event => {
+ client.ackMessage(event.raw);
+ return { type: 'EVENT_TEST_RESPONSE', payload: event.payload + 1 };
+ }),
+ tap(event => eventSubject.next(event)),
+ );
+
+ const options = createOptions({ expectAck: true, queue: 'test_ack_queue_server' });
+ const client = await runClient(options);
+ const server = await runServer(options)(event$);
+ const message = createMessage({ type: 'EVENT_TEST', payload: 1 });
+
+ await client.emitMessage(options.queue, message);
+ });
+
+ test('reacts to thrown Error', async done => {
+ const event = { type: 'EVENT_TEST' };
+ const error = new Error('test_error');
+ const errorSubject = new Subject<[Event | undefined, Error]>();
+
+ errorSubject.subscribe(data => {
+ expect(data[0]).toBeUndefined();
+ expect(data[1]).toEqual(error);
+ setTimeout(() => server.close().then(done), 1000);
+ });
+
+ const event$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('EVENT_TEST'),
+ mergeMapTo(throwError(error)),
+ );
+
+ const error$: MsgErrorEffect = event$ =>
+ event$.pipe(
+ tap(({ event, error }) => errorSubject.next([event, error])),
+ map(() => createEvent('ERROR')),
+ );
+
+ const options = createOptions({ expectAck: false });
+ const client = await runClient(options);
+ const server = await runServer(options)(event$, error$);
+
+ const emitResult = await client.emitMessage(options.queue, createMessage(event));
+
+ expect(emitResult).toEqual(true);
+ });
+
+ test('reacts to thrown EventError', async done => {
+ const event = { type: 'EVENT_TEST' };
+ const error = new EventError(event, 'test_error');
+ const errorSubject = new Subject<[Event | undefined, Error]>();
errorSubject.subscribe(data => {
- expect(data[0]).toEqual(expectedMessage);
- expect(data[1]).toEqual(expectedError);
+ expect(data[0]).toEqual(event);
+ expect(data[1]).toEqual(error);
setTimeout(() => server.close().then(done), 1000);
});
const event$: MsgEffect = event$ =>
event$.pipe(
matchEvent('EVENT_TEST'),
- mergeMapTo(throwError(expectedError)),
+ mergeMapTo(throwError(error)),
);
const error$: MsgErrorEffect = event$ =>
event$.pipe(
tap(({ event, error }) => errorSubject.next([event, error])),
- map(({ event }) => event),
+ map(() => createEvent('ERROR')),
);
- const server = await runServer(event$, error$);
- const client = await runClient();
+ const options = createOptions({ expectAck: false });
+ const client = await runClient(options);
+ const server = await runServer(options)(event$, error$);
- const emitResult = await client.emitMessage(options.queue, createMessage(expectedMessage));
+ const emitResult = await client.emitMessage(options.queue, createMessage(event));
expect(emitResult).toEqual(true);
});

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
@@ -41,11 +41,11 @@ class AmqpStrategyConnection implements TransportLayerConnection {
get message$() {
return this.msgSubject$.asObservable().pipe(
- map(message => ({
- data: message.content,
- replyTo: message.properties.replyTo,
- correlationId: message.properties.correlationId,
- raw: message,
+ map(raw => ({
+ data: raw.content,
+ replyTo: raw.properties.replyTo,
+ correlationId: raw.properties.correlationId,
+ raw,
} as TransportMessage<Buffer>)),
);
}
@@ -102,15 +102,15 @@ class AmqpStrategyConnection implements TransportLayerConnection {
});
};
- ackMessage = (message: TransportMessage<any> | undefined) => {
- if (message) {
- this.channelWrapper.ack(message.raw);
+ ackMessage = (msg: ConsumeMessage | undefined) => {
+ if (msg) {
+ this.channelWrapper.ack(msg);
}
}
- nackMessage = (message: TransportMessage<any> | undefined, resend = true) => {
- if (message) {
- this.channelWrapper.nack(message.raw, false , resend);
+ nackMessage = (msg: ConsumeMessage | undefined, resend = true) => {
+ if (msg) {
+ this.channelWrapper.nack(msg, false , resend);
}
}

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -23,8 +23,8 @@ export interface TransportLayerConfig {
export interface TransportLayerConnection {
sendMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<TransportMessage<Buffer>>;
emitMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<any>;
- ackMessage: (message: TransportMessage<any> | undefined) => void;
- nackMessage: (message: TransportMessage<any> | undefined, resend?: boolean) => void;
+ ackMessage: (message: any | undefined) => void;
+ nackMessage: (message: any | undefined, resend?: boolean) => void;
close: () => Promise<any>;
getChannel(): string;
message$: Observable<TransportMessage<Buffer>>;

MODIFY
packages/websockets/src/effects/ws-effects.interface.ts
packages/websockets/src/effects/ws-effects.interface.ts
@@ -11,7 +11,7 @@ export interface WsErrorEffect<
T extends Error = Error,
U = Event,
V = Event
-> extends WsEffect<{ event: U; error: T }, V, MarbleWebSocketClient> {}
+> extends WsEffect<{ event?: U; error: T }, V, MarbleWebSocketClient> {}
export interface WsConnectionEffect<
T extends http.IncomingMessage = http.IncomingMessage

imp(messaging): continuous msg stream + error handling imp + nack/ack imp