feat(messaging): output$
MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
@@ -8,6 +8,7 @@ import {
MsgMiddlewareEffect,
MsgServerEffect,
ServerEvent,
+ MsgOutputEffect,
} from '@marblejs/messaging';
import { merge } from 'rxjs';
import { map, tap } from 'rxjs/operators';
@@ -44,6 +45,14 @@ const error$: MsgServerEffect = event$ =>
tap(({ error }) => console.error(error)),
);
+const output$: MsgOutputEffect = (event$, _, { initiator }) =>
+ event$.pipe(
+ tap(event => console.log(`processed ::`, event, initiator && {
+ replyTo: initiator.replyTo,
+ correlationId: initiator.correlationId,
+ })),
+ );
+
export const microservice = createMicroservice({
transport: Transport.AMQP,
options: {
@@ -54,6 +63,7 @@ export const microservice = createMicroservice({
messagingListener: messagingListener({
effects: [fibonacci$],
middlewares: [log$],
+ output$,
}),
dependencies: [],
event$: (...args) => merge(

MODIFY
packages/messaging/src/client/messaging.client.interface.ts
packages/messaging/src/client/messaging.client.interface.ts
@@ -5,7 +5,7 @@ import { AmqpStrategy } from '../transport/strategies/amqp.strategy.interface';
import { TransportMessageTransformer } from '../transport/transport.interface';
export interface MessagingClient {
- emit: <T>(data: T) => Observable<any>;
+ emit: <T>(data: T) => Observable<boolean>;
send: <T, U>(data: T) => Observable<U>;
close: () => Observable<any>;
}

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 } from '../transport/transport.interface';
+import { TransportLayerConnection, TransportMessage } from '../transport/transport.interface';
type MsgClient = TransportLayerConnection;
@@ -17,7 +17,12 @@ export interface MsgEffect<
O = Event,
Client = MsgClient,
Err extends Error = Error,
-> extends Effect<I, O, Client, Err> {}
+> extends Effect<I, O, Client, Err, TransportMessage<any>> {}
+
+export interface MsgOutputEffect<
+ I = Event,
+ O = Event,
+> extends MsgEffect<I, O> {}
export interface MsgServerEffect<T extends Event = Event>
extends MsgEffect<T, any, undefined> {}

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -14,7 +14,7 @@ import {
TransportLayer,
} from '../transport/transport.interface';
import { jsonTransformer } from '../transport/transport.transformer';
-import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
+import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect, MsgOutputEffect } from '../effects/messaging.effects.interface';
import { TransportLayerToken, ServerEventsToken } from './messaging.server.tokens';
import { AllServerEvents, ServerEvent } from './messaging.server.events';
@@ -22,6 +22,7 @@ export interface MessagingListenerConfig {
effects?: MsgEffect<any, any>[];
middlewares?: MsgMiddlewareEffect<any, any>[];
error$?: MsgErrorEffect;
+ output$?: MsgOutputEffect;
msgTransformer?: TransportMessageTransformer<any>;
}
@@ -29,6 +30,7 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const {
effects = [],
middlewares = [],
+ output$ = (msg$ => msg$) as MsgOutputEffect,
error$ = (msg$ => msg$) as MsgErrorEffect,
msgTransformer = jsonTransformer,
} = config;
@@ -56,6 +58,10 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
withLatestFrom(msg$),
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
+ publish(msg$ => output$(msg$.pipe(map(m => m.data)), conn, createEffectMetadata({ ask, initiator: msg })).pipe(
+ withLatestFrom(msg$),
+ )),
+ map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
catchError(error => error$(of(msg.data), conn, createEffectMetadata({ ask, error })).pipe(
map(data => ({ ...msg, data } as TransportMessage<any>)),
)),

feat(messaging): output$