feat(messaging): client
MODIFY
packages/messaging/package.json
packages/messaging/package.json
{
"name": "@marblejs/messaging",
- "version": "2.1.0",
+ "version": "2.1.1",
"description": "Messaging module for Marble.js",
"main": "./dist/index.js",
"types": "./dist/index.d.ts",
"watch": "tsc -w",
"build": "tsc",
"clean": "rimraf dist",
- "lint": "tslint --config ../../tslint.json --project ./",
"test": "jest --config ../../jest.config.js"
},
"files": [

DELETE
packages/messaging/src/client/.gitkeep
packages/messaging/src/client/.gitkeep
ADD
packages/messaging/src/client/messaging.client.interface.ts
+import { Observable } from 'rxjs';
+
+export interface MessagingClient {
+ emit: <T>(data: T) => Observable<any>;
+ send: <T, U>(data: T) => Observable<U>;
+ publish: <T>(data: T) => Observable<any>;
+}

ADD
packages/messaging/src/client/messaging.client.ts
+import { reader } from '@marblejs/core';
+import { from, Observable } from 'rxjs';
+import { mergeMap, take, map, tap } from 'rxjs/operators';
+import {
+ Transport,
+ TransportMessage,
+ TransportMessageTransformer,
+ TransportLayerConnection,
+} from '../transport/transport.interface';
+import { provideTransportLayer } from '../transport/transport.provider';
+import { jsonTransformer } from '../transport/transport.transformer';
+import { MessagingClient } from './messaging.client.interface';
+import { createUuid } from '@marblejs/core/dist/+internal/utils';
+
+export interface MessagingClientConfig {
+ transport?: Transport;
+ msgTransformer?: TransportMessageTransformer<any>;
+ options?: any; // @TODO
+}
+
+export const messagingClient = (config: MessagingClientConfig = {}) => {
+ const {
+ transport = Transport.TCP,
+ msgTransformer = jsonTransformer,
+ options = {},
+ } = config;
+
+ const emit = (conn$: Observable<TransportLayerConnection>) => <T>(msg: T) =>
+ conn$.pipe(
+ mergeMap(conn => conn.sendMessage(
+ options.queue,
+ { data: msgTransformer.encode(msg as any) },
+ { type: 'emit' },
+ )),
+ take(1),
+ );
+
+ const publish = (conn$: Observable<TransportLayerConnection>) => <T>(msg: T) =>
+ conn$.pipe(
+ mergeMap(conn => conn.sendMessage(
+ options.queue,
+ { data: msgTransformer.encode(msg as any) },
+ { type: 'publish' },
+ )),
+ take(1),
+ );
+
+ const send = (conn$: Observable<TransportLayerConnection>) => <T, U>(msg: T): Observable<U> =>
+ conn$.pipe(
+ mergeMap(conn => conn.sendMessage(
+ options.queue,
+ { data: msgTransformer.encode(msg as any), correlationId: createUuid() },
+ { type: 'send' }
+ )),
+ map(m => m as TransportMessage<Buffer>),
+ map(m => msgTransformer.decode(m.data) as U),
+ take(1),
+ );
+
+ return reader.map(() => {
+ const transportLayer = provideTransportLayer(transport, options);
+ const conn$ = from(transportLayer).pipe(
+ mergeMap(server => server.connect()),
+ tap(conn => conn.consumeResponse()),
+ );
+
+ return {
+ emit: emit(conn$),
+ send: send(conn$),
+ publish: publish(conn$),
+ } as MessagingClient;
+ });
+};

MODIFY
packages/messaging/src/listener/messaging.listener.ts
packages/messaging/src/listener/messaging.listener.ts
@@ -6,9 +6,8 @@ import {
combineEffects,
createEffectMetadata,
} from '@marblejs/core';
-import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { from, Observable, Subscription } from 'rxjs';
-import { map, mergeMap, publish, withLatestFrom } from 'rxjs/operators';
+import { map, mergeMap, publish, withLatestFrom, tap } from 'rxjs/operators';
import {
Transport,
TransportMessage,
@@ -43,7 +42,7 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const combinedMiddlewares = combineMiddlewares(...middlewares);
const defaultMetadata = createEffectMetadata({ ask });
- const message$ = conn.handleMessage().pipe(
+ const message$ = conn.message$.pipe(
map(msg => ({ ...msg, data: msgTransformer.decode(msg.data) } as TransportMessage<any>)),
publish(msg$ => combinedMiddlewares(msg$.pipe(map(m => m.data)), conn, defaultMetadata).pipe(
withLatestFrom(msg$),
@@ -59,7 +58,6 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
input$.subscribe(
msg => {
if (msg.replyTo) {
- console.log(msg);
conn.sendMessage(msg.replyTo, {
data: msgTransformer.encode(msg.data),
correlationId: msg.correlationId,
@@ -81,60 +79,10 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const transportLayer = provideTransportLayer(transport, options);
from(transportLayer)
- .pipe(mergeMap(server => server.connect()))
+ .pipe(
+ mergeMap(server => server.connect()),
+ tap(conn => conn.consumeMessage()),
+ )
.subscribe(conn => handleConnection(conn, ask));
-
- // @TODO: simulate msgs posting
- transportLayer
- .then(server => server.connect())
- .then(conn => {
- const msg1 = msgTransformer.encode({ type: 'ADD', payload: 1 });
- const msg2 = msgTransformer.encode({ type: 'ADD', payload: 2 });
- const msg3 = msgTransformer.encode({ type: 'SUM' });
-
- const msg4 = msgTransformer.encode({ type: 'ADD', payload: 3 });
- const msg5 = msgTransformer.encode({ type: 'ADD', payload: 6 });
- const msg6 = msgTransformer.encode({ type: 'SUM' });
-
- setTimeout(() => {
- conn.sendMessage(options.queue, {
- data: msg1,
- });
- }, 1);
-
- setTimeout(() => {
- conn.sendMessage(options.queue, {
- data: msg2,
- });
- }, 2);
-
- setTimeout(() => {
- conn.sendMessage(options.queue, {
- data: msg3,
- replyTo: options.queue + '_1',
- correlationId: createUuid(),
- });
- }, 3);
-
- setTimeout(() => {
- conn.sendMessage(options.queue, {
- data: msg4,
- });
- }, 10);
-
- setTimeout(() => {
- conn.sendMessage(options.queue, {
- data: msg5,
- });
- }, 11);
-
- setTimeout(() => {
- conn.sendMessage(options.queue, {
- data: msg6,
- replyTo: options.queue + '_2',
- correlationId: createUuid(),
- });
- }, 12);
- });
});
};

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
import { EMPTY, Subject, from, fromEvent } from 'rxjs';
-import { map, share } from 'rxjs/operators';
+import { map, share, filter } from 'rxjs/operators';
import { Channel, Connection, ConsumeMessage, Options } from 'amqplib';
-import { TransportLayer, TransportMessage } from '../transport.interface';
+import { TransportLayer, TransportLayerSendOpts, TransportMessage } from '../transport.interface';
interface RmqStrategyOptions {
host: string;
@@ -13,25 +13,21 @@ interface RmqStrategyOptions {
}
export const createAmqpStrategy = async (options: RmqStrategyOptions): Promise<TransportLayer> => {
+ const msgSubject$ = new Subject<ConsumeMessage>();
+ const resSubject$ = new Subject<ConsumeMessage>();
+ const responseQueue = options.queue + '__response';
- const sendMessage = (channelInstance: Channel) => (queue: string, msg: TransportMessage<Buffer>) => {
- const { correlationId, replyTo } = msg;
-
- channelInstance.sendToQueue(queue, msg.data, { correlationId, replyTo });
-
- return EMPTY;
- };
-
- const handleMessage = (channelInstance: Channel) => () => {
- const msg$ = new Subject<ConsumeMessage>();
-
- channelInstance.consume(
- options.queue,
- msg => msg && msg$.next(msg),
- { noAck: true },
+ const message$ = msgSubject$.asObservable().pipe(
+ share(),
+ map(message => ({
+ data: message.content,
+ replyTo: message.properties.replyTo,
+ correlationId: message.properties.correlationId,
+ raw: message,
+ } as TransportMessage<Buffer>))
);
- return msg$.asObservable().pipe(
+ const response$ = resSubject$.asObservable().pipe(
share(),
map(message => ({
data: message.content,
@@ -40,6 +36,43 @@ export const createAmqpStrategy = async (options: RmqStrategyOptions): Promise<T
raw: message,
} as TransportMessage<Buffer>))
);
+
+ const consumeMessage = (channelInstance: Channel) => () => {
+ channelInstance.consume(
+ options.queue,
+ msg => msg && msgSubject$.next(msg),
+ { noAck: true },
+ );
+ };
+
+ const consumeResponse = (channelInstance: Channel) => () => {
+ channelInstance.consume(
+ responseQueue,
+ res => res && resSubject$.next(res),
+ { noAck: true },
+ );
+ };
+
+ const sendMessage = (channelInstance: Channel) => (
+ queue: string,
+ msg: TransportMessage<Buffer>,
+ opts: TransportLayerSendOpts = {},
+ ) => {
+ const { correlationId, replyTo } = msg;
+
+ switch (opts.type) {
+ case 'publish':
+ channelInstance.assertExchange(queue, 'fanout', { durable: false });
+ channelInstance.publish(queue, '', msg.data);
+ return EMPTY;
+ case 'send':
+ channelInstance.prefetch(1);
+ channelInstance.sendToQueue(queue, msg.data, { correlationId, replyTo: responseQueue });
+ return response$.pipe(filter(m => m.correlationId === correlationId));
+ default:
+ channelInstance.sendToQueue(queue, msg.data, { replyTo, correlationId });
+ return EMPTY;
+ }
};
const close = (connection: Connection) => () =>
@@ -54,12 +87,16 @@ export const createAmqpStrategy = async (options: RmqStrategyOptions): Promise<T
const channel = await connection.createChannel();
channel.assertQueue(options.queue, options.queueOptions);
+ channel.assertQueue(responseQueue);
return {
sendMessage: sendMessage(channel),
- handleMessage: handleMessage(channel),
+ consumeMessage: consumeMessage(channel),
+ consumeResponse: consumeResponse(channel),
close: close(connection),
error$: error$(connection),
+ response$,
+ message$,
};
};

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -15,10 +15,17 @@ export interface TransportLayer {
}
export interface TransportLayerConnection {
- sendMessage: (channel: string, message: TransportMessage<Buffer>) => Observable<any>;
- handleMessage: () => Observable<TransportMessage<Buffer>>;
+ sendMessage: (channel: string, message: TransportMessage<Buffer>, opts?: TransportLayerSendOpts) => Observable<any>;
+ consumeMessage: () => void;
+ consumeResponse: () => void;
close: () => Observable<any>;
error$: Observable<Error>;
+ message$: Observable<TransportMessage<Buffer>>;
+ response$: Observable<TransportMessage<Buffer>>;
+}
+
+export interface TransportLayerSendOpts {
+ type?: 'publish' | 'send' | 'emit';
}
export interface TransportMessageTransformer<T> {

feat(messaging): client