imp(messaging): continuous stream
MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
-import { matchEvent, use } from '@marblejs/core';
+import { of } from 'rxjs';
+import { map, tap, mergeMap, bufferCount, mapTo, catchError } from 'rxjs/operators';
+import { matchEvent, use, Event } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
import { createMicroservice, messagingListener, Transport, MsgEffect } from '@marblejs/messaging';
-import { map, tap } from 'rxjs/operators';
const fib = (n: number): number =>
(n === 0 || n === 1) ? n : fib(n - 1) + fib(n - 2);
@@ -10,9 +11,19 @@ const fibonacci$: MsgEffect = event$ =>
event$.pipe(
matchEvent('FIB'),
use(eventValidator$(t.number)),
+ mergeMap(event => of(event).pipe(
tap(event => { if (event.payload >= 45) throw new Error('Too high number!') }),
map(event => fib(event.payload)),
- map(payload => ({ type: 'FIB_RESULT', payload })),
+ map(payload => ({ ...event, type: 'FIB_RESULT', payload } as Event)),
+ catchError(error => of({ ...event, error: { name: error.name, message: error.message } } as Event)),
+ )),
+ );
+
+const test$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('TEST'),
+ bufferCount(2),
+ mapTo({ type: 'TEST_RESULT' }),
);
export const microservice = createMicroservice({
@@ -23,7 +34,7 @@ export const microservice = createMicroservice({
queueOptions: { durable: false },
},
messagingListener: messagingListener({
- effects: [fibonacci$],
+ effects: [fibonacci$, test$],
}),
});

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

MODIFY
packages/messaging/src/effects/messaging.effects.interface.ts
packages/messaging/src/effects/messaging.effects.interface.ts
import { Effect, Event } from '@marblejs/core';
-import { TransportLayerConnection, TransportMessage } from '../transport/transport.interface';
+import { TransportLayerConnection } from '../transport/transport.interface';
type MsgClient = TransportLayerConnection;
@@ -9,9 +9,8 @@ export interface MsgMiddlewareEffect<
> extends MsgEffect<I, O> {}
export interface MsgErrorEffect<
- I = Event,
Err extends Error = Error,
-> extends MsgEffect<{ event: I; error: Err }, Event, MsgClient> {}
+> extends MsgEffect<Err, Event, MsgClient> {}
export interface MsgEffect<
I = Event,
@@ -22,7 +21,7 @@ export interface MsgEffect<
export interface MsgOutputEffect<
I = Event,
O = Event,
-> extends MsgEffect<{ event: I; initiator: TransportMessage<any> }, O> {}
+> extends MsgEffect<I, O> {}
export interface MsgServerEffect<T extends Event = Event>
extends MsgEffect<T, any, undefined> {}

MODIFY
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
packages/messaging/src/middlewares/messaging.eventLogger.middleware.ts
-import { tap, map } from 'rxjs/operators';
+import { tap } from 'rxjs/operators';
import { useContext } from '@marblejs/core';
import { provideLogger, LoggerLevel } from '../server/messaging.server.logger';
-import { MsgMiddlewareEffect, MsgOutputEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
+import { MsgMiddlewareEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
import { TransportLayerToken } from '../server/messaging.server.tokens';
export const inputLogger$: MsgMiddlewareEffect = (event$, ctx) => {
@@ -11,7 +11,7 @@ export const inputLogger$: MsgMiddlewareEffect = (event$, ctx) => {
return event$.pipe(
tap(event => {
const channel = transportLayer.config.channel;
- const message = `${event.type}, id: ${event.raw.correlationId}`;
+ const message = `${event.type}, id: ${event.metadata?.correlationId}`;
logger({
level: LoggerLevel.INFO,
@@ -28,13 +28,14 @@ export const outputLogger$: MsgOutputEffect = (event$, ctx) => {
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
return event$.pipe(
- tap(({ event, initiator }) => {
+ tap(event => {
+ const { error, metadata, type } = event;
const channel = transportLayer.config.channel;
- const message = event.error
- ? `${event.error.name}, ${event.error.message} for event ${event.type}`
- : initiator.replyTo
- ? `${event.type}, id: ${initiator.correlationId} and sent to ${initiator.replyTo}`
- : `${event.type}, id: ${initiator.correlationId}`;
+ const message = error
+ ? `"${error.name}", "${error.message}" for event ${type}`
+ : metadata?.replyTo
+ ? `${event.type}, id: ${metadata?.correlationId ?? '-'} and sent to ${metadata?.replyTo ?? '-'}`
+ : `${event.type}, id: ${metadata?.correlationId ?? '-'}`;
logger({
level: event.error ? LoggerLevel.ERROR : LoggerLevel.INFO,
@@ -43,23 +44,19 @@ export const outputLogger$: MsgOutputEffect = (event$, ctx) => {
channel,
});
}),
- map(({ event }) => event),
);
};
-export const errorLogger$: MsgErrorEffect = (event$, ctx) => {
+export const errorLogger$: MsgMiddlewareEffect = (event$, ctx) => {
const logger = provideLogger(ctx.ask);
const transportLayer = useContext(TransportLayerToken)(ctx.ask);
return event$.pipe(
- tap(({ error, event }) => logger({
+ tap(event => logger({
tag: 'ERROR',
- message: event
- ? `${error.name}, ${error.message} for event ${event.type}`
- : `${error.name}, ${error.message}`,
+ message: `"${event.error?.name ?? '-'}: ${event.error?.message ?? '-' }" for event ${event.type}`,
level: LoggerLevel.ERROR,
channel: transportLayer.config.channel,
})),
- map(({ event }) => event),
);
};

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -5,8 +5,8 @@ import {
createEffectContext,
createListener,
} from '@marblejs/core';
-import { Subject, of} from 'rxjs';
-import { map, publish, takeUntil, catchError, mergeMapTo, mergeMap, tap } from 'rxjs/operators';
+import { of, Observable} from 'rxjs';
+import { map, publish, catchError, takeUntil } from 'rxjs/operators';
import {
TransportMessage,
TransportMessageTransformer,
@@ -32,19 +32,15 @@ const defaultEffect$: MsgEffect = msg$ =>
msg$;
const defaultOutput$: MsgOutputEffect = msg$ =>
- msg$.pipe(map(m => m.event));
+ msg$;
const defaultError$: MsgErrorEffect = msg$ =>
- msg$.pipe(map(({ event, error }) => ({
- type: event.type,
- payload: event.payload,
- error: { name: error.name, message: error.message },
- } as Event)));
+ msg$.pipe(map(error => ({ type: 'UNHANDLED_ERROR', error: { name: error.name, message: error.message } } as Event)));
export const messagingListener = createListener<MessagingListenerConfig, MessagingListener>(config => ask => {
const {
- effects = [defaultEffect$],
middlewares = [],
+ effects = [defaultEffect$],
output$ = defaultOutput$,
error$ = defaultError$,
msgTransformer = jsonTransformer,
@@ -55,47 +51,54 @@ export const messagingListener = createListener<MessagingListenerConfig, Messagi
const combinedMiddlewares = combineMiddlewares(inputLogger$, ...middlewares);
const ctx = createEffectContext({ ask, client: connection });
- const decode = (msg: TransportMessage<Buffer>): TransportMessage<Event> => ({
- ...msg,
- data: { ...msgTransformer.decode(msg.data), raw: msg },
+ const decode = (msg: TransportMessage<Buffer>): Event => ({
+ ...msgTransformer.decode(msg.data),
+ metadata: {
+ replyTo: msg.replyTo,
+ correlationId: msg.correlationId,
+ raw: msg,
+ }
});
- const subject = new Subject<TransportMessage<Buffer>>();
-
- subject
- .pipe(
+ const stream = connection.message$.pipe(
takeUntil(connection.close$),
map(decode),
- mergeMap(initiator => of(initiator.data).pipe(
publish(e$ => combinedMiddlewares(e$, ctx)),
publish(e$ => combinedEffects(e$, ctx)),
- publish(e$ => outputLogger$(e$.pipe(map(event => ({ initiator, event }))), ctx)),
- publish(e$ => output$(e$.pipe(map(event => ({ initiator, event }))), ctx)),
+ publish(e$ => output$(e$, ctx)),
+ publish(e$ => outputLogger$(e$, ctx)),
catchError(error => {
- const e$ = of({ event: initiator.data, error });
- return errorLogger$(e$, ctx).pipe(
- mergeMapTo(error$(e$, ctx)),
+ const e$ = of(error);
+ return error$(e$, ctx).pipe(
+ publish(e$ => errorLogger$(e$, ctx)),
);
}),
- tap(async event => {
- if (initiator.replyTo) {
- const { replyTo, correlationId, raw } = initiator;
- const { type, payload, error } = event;
- return connection.emitMessage(replyTo, {
+ );
+
+ const subscribe = (event$: Observable<Event<unknown, any, string>>) =>
+ event$.subscribe(
+ event => {
+ const { metadata, type, payload, error } = event;
+
+ if (metadata && metadata.replyTo) {
+ const { replyTo, correlationId, raw } = metadata;
+
+ connection.emitMessage(replyTo, {
data: msgTransformer.encode({ type, payload, error }),
correlationId,
raw,
});
}
+ },
+ () => {
+ console.error('Unexpected stream error!'); // @TODO: handle unexpected error
+ subscribe(event$);
+ },
+ () => {
+ subscribe(event$);
+ },
+ );
- return true;
- }),
- )),
- )
- .subscribe();
-
- connection.message$
- .pipe(takeUntil(connection.close$))
- .subscribe(msg => subject.next(msg));
+ subscribe(stream);
};
});

MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
@@ -14,15 +14,14 @@ const createOptions = (config: { expectAck?: boolean; queue?: string } = {}): Am
queueOptions: { durable: false },
});
-describe.only('messagingServer::AMQP', () => {
+describe('messagingServer::AMQP', () => {
test('handles RPC event', async () => {
const rpc$: MsgEffect = event$ =>
event$.pipe(
matchEvent('RPC_TEST'),
- delay(250),
+ delay(50),
use(eventValidator$(t.number)),
- map(event => event.payload),
- map(payload => ({ type: 'RPC_TEST_RESULT', payload: payload + 1 })),
+ map(event => ({ ...event, type: 'RPC_TEST_RESULT', payload: event.payload + 1 })),
);
const options = createOptions({ expectAck: false });
@@ -65,9 +64,7 @@ describe.only('messagingServer::AMQP', () => {
const server = await runServer(Transport.AMQP, options)(event$);
const message = createMessage({ type: 'EVENT_TEST', payload: 1 });
- const emitResult = await client.emitMessage(options.queue, message);
-
- expect(emitResult).toEqual(true);
+ await client.emitMessage(options.queue, message);
});
test('acks processed event', async done => {
@@ -83,7 +80,7 @@ describe.only('messagingServer::AMQP', () => {
matchEvent('EVENT_TEST'),
use(eventValidator$(t.number)),
map(event => {
- client.ackMessage(event.raw);
+ client.ackMessage(event.metadata?.raw);
return { type: 'EVENT_TEST_RESPONSE', payload: event.payload + 1 };
}),
tap(event => eventSubject.next(event)),
@@ -97,18 +94,18 @@ describe.only('messagingServer::AMQP', () => {
await client.emitMessage(options.queue, message);
});
- test.only('reacts to thrown Error and doesn\'t crash internal messages stream', async done => {
+ test('reacts to thrown Error and doesn\'t crash internal messages stream', async done => {
const event1: Event = { type: 'EVENT_TEST_1' };
const event2: Event = { type: 'EVENT_TEST_2' };
const error = new Error('test_error');
- const outputSubject = new Subject<[Event, Error | undefined]>();
+ const outputSubject = new Subject<[Event | undefined, Error | undefined]>();
outputSubject
.pipe(bufferCount(2))
.subscribe(data => {
- expect(data[0][0].type).toEqual(event1.type);
+ expect(data[0][0]).toBeUndefined();
expect(data[0][1]).toEqual(error);
- expect(data[1][0].type).toEqual(event2.type);
+ expect(data[1][0]).toEqual(event2);
expect(data[1][1]).toBeUndefined();
setTimeout(() => server.close().then(done), 1000);
});
@@ -122,18 +119,18 @@ describe.only('messagingServer::AMQP', () => {
const event2$: MsgEffect = event$ =>
event$.pipe(
matchEvent(event2.type),
+ map(event => ({ type: event.type })),
);
const error$: MsgErrorEffect = event$ =>
event$.pipe(
- tap(({ event, error }) => outputSubject.next([event, error])),
- map(({ event }) => event),
+ tap(error => outputSubject.next([undefined, error])),
+ map(() => ({ type: 'UNHANDLED' })),
);
const output$: MsgOutputEffect = event$ =>
event$.pipe(
- tap(({ event }) => outputSubject.next([event, undefined])),
- map(({ event }) => event),
+ tap((event => outputSubject.next([event, undefined]))),
);
const options = createOptions({ expectAck: false });

MODIFY
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
@@ -30,8 +30,7 @@ describe('messagingServer::Redis', () => {
event$.pipe(
matchEvent('RPC_TEST'),
use(eventValidator$(t.number)),
- map(event => event.payload),
- map(payload => ({ type: 'RPC_TEST_RESULT', payload: payload + 1 })),
+ map(event => ({ ...event, type: 'RPC_TEST_RESULT', payload: event.payload + 1 })),
);
const options = createOptions();
@@ -68,9 +67,8 @@ describe('messagingServer::Redis', () => {
const client = await runClient(Transport.REDIS, options);
const server = await runServer(Transport.REDIS, options)(event$);
const message = createMessage({ type: 'EVENT_TEST', payload: 1 });
- const emitResult = await client.emitMessage(options.channel, message);
- expect(emitResult).toEqual(true);
+ await client.emitMessage(options.channel, message);
});
test('throws an UnsupportedError for unsupported "ackMessage/nackMessage"', async done => {
@@ -82,21 +80,20 @@ describe('messagingServer::Redis', () => {
const event$: MsgEffect = (event$, ctx) =>
event$.pipe(
matchEvent('EVENT_TEST'),
- tap(event => ctx.client.ackMessage(event.raw)),
+ tap(event => ctx.client.ackMessage(event.metadata?.raw)),
);
const error$: MsgErrorEffect = event$ =>
event$.pipe(
- map(({ event, error}) => ({ event, error: { name: error.name, message: error.message }})),
- tap(async ({ event, error }) => {
- expect(event.type).toEqual('EVENT_TEST');
- expect(error).toEqual(expectedEventError);
+ map(error => ({ type: 'UNHANDLED_ERROR', error: { name: error.name, message: error.message }})),
+ tap(async event => {
+ expect(event.type).toEqual('UNHANDLED_ERROR');
+ expect(event.error).toEqual(expectedEventError);
await wait();
await server.close();
await client.close();
done();
}),
- map(({ event }) => event),
);
const options = createOptions();

imp(messaging): continuous stream