refactor(messaging): sendMessage + emitMessage
ADD
packages/messaging/src/client/messaging.client.spec.ts
+import { createContext } from '@marblejs/core';
+import { messagingClient } from './messaging.client';
+import { MessagingClientConfig } from './messaging.client.interface';
+import { Transport, TransportMessage } from '../transport/transport.interface';
+import { createAmqpStrategy } from '../transport/strategies/amqp.strategy';
+
+describe('#messagingClient', () => {
+
+ describe('AMQP', () => {
+ const clientOptions: MessagingClientConfig = {
+ transport: Transport.AMQP,
+ options: {
+ host: 'amqp://localhost:5672',
+ queue: 'test_queue_client',
+ queueOptions: { durable: false },
+ },
+ };
+
+ const runClient = () => messagingClient(clientOptions).run(createContext());
+ const runServer = () => createAmqpStrategy(clientOptions.options).connect();
+
+ test('emits event to consumer', async done => {
+ const client = runClient();
+ const server = await runServer();
+
+ server.consumeMessage().subscribe(async msg => {
+ expect(msg.data.toString()).toEqual(JSON.stringify({ test: 'test' }));
+ server.ack(msg);
+ await server.close();
+ done();
+ });
+
+ const result = await client.emit({ test: 'test' }).toPromise();
+ expect(result).toEqual(true);
+ await client.close().toPromise();
+ });
+
+ test('sends command to consumer', async () => {
+ const client = runClient();
+ const server = await runServer();
+
+ server.consumeMessage().subscribe(async msg => {
+ const count = String(Number(msg.data.toString()) + 1);
+ const response: TransportMessage<Buffer> = {
+ data: Buffer.from(count),
+ correlationId: msg.correlationId,
+ };
+
+ await server.emitMessage(msg.replyTo || '', response);
+ server.ack(msg);
+ await server.close();
+ });
+
+ const result = await client.send(1).toPromise();
+ expect(result).toEqual(2);
+ await client.close().toPromise();
+ });
+ });
+
+});

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
@@ -20,21 +20,9 @@ export const messagingClient = (config: MessagingClientConfig) => {
const emit = (conn: Promise<TransportLayerConnection>) => <T>(msg: T) =>
from(conn).pipe(
- mergeMap(c => c.sendMessage(
- c.getChannel(),
- { data: msgTransformer.encode(msg as any) },
- { type: 'emit' },
- )),
- mapTo(true),
- take(1),
- );
-
- const publish = (conn: Promise<TransportLayerConnection>) => <T>(msg: T) =>
- from(conn).pipe(
- mergeMap(c => c.sendMessage(
+ mergeMap(c => c.emitMessage(
c.getChannel(),
{ data: msgTransformer.encode(msg as any) },
- { type: 'publish' },
)),
mapTo(true),
take(1),
@@ -45,7 +33,6 @@ export const messagingClient = (config: MessagingClientConfig) => {
mergeMap(c => c.sendMessage(
c.getChannel(),
{ data: msgTransformer.encode(msg as any), correlationId: createUuid() },
- { type: 'send' }
)),
map(m => m as TransportMessage<Buffer>),
map(m => msgTransformer.decode(m.data) as U),
@@ -60,7 +47,7 @@ export const messagingClient = (config: MessagingClientConfig) => {
return reader.map(ask => {
const transportLayer = provideTransportLayer(transport, options);
- const connection = transportLayer.connect().then(conn => conn.consumeResponse());
+ const connection = transportLayer.connect();
ask(HttpServerEventStreamToken).map(serverEvent$ =>
serverEvent$.pipe(
@@ -74,7 +61,6 @@ export const messagingClient = (config: MessagingClientConfig) => {
return {
emit: emit(connection),
send: send(connection),
- publish: publish(connection),
close: close(connection),
} as MessagingClient;
});

MODIFY
packages/messaging/src/listener/messaging.listener.ts
packages/messaging/src/listener/messaging.listener.ts
@@ -40,7 +40,7 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const combinedMiddlewares = combineMiddlewares(...middlewares);
const defaultMetadata = createEffectMetadata({ ask });
- const message$ = conn.message$.pipe(
+ const message$ = conn.consumeMessage().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$),
@@ -62,13 +62,13 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const onSubscribeEffectsOutput = (conn: TransportLayerConnection) => (msg: TransportMessage<any>) => {
if (msg.replyTo) {
- conn.sendMessage(msg.replyTo, {
+ conn.emitMessage(msg.replyTo, {
data: msgTransformer.encode(msg.data),
correlationId: msg.correlationId,
raw: msg.raw,
});
}
- conn.ack(msg.raw);
+ conn.ack(msg);
}
const onSubscribeEffectsError = (errorSubject: Subject<Error>) => (error: Error) => {
@@ -88,8 +88,9 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const listen = (transportLayer: TransportLayer, ask: ContextProvider) => async () => {
const connection = await transportLayer.connect();
- await connection.consumeMessage();
- await handleConnection(connection, ask);
+
+ handleConnection(connection, ask);
+
return connection;
};

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
-import { Subject, fromEvent, of } from 'rxjs';
-import { map, share, filter } from 'rxjs/operators';
-import { Channel, Connection, ConsumeMessage, Replies } from 'amqplib';
-import { TransportLayer, TransportLayerSendOpts, TransportMessage, TransportLayerConnection } from '../transport.interface';
+import { Subject, fromEvent, from } from 'rxjs';
+import { map, filter, take, mergeMap, mapTo } from 'rxjs/operators';
+import { Channel, Connection, ConsumeMessage } from 'amqplib';
+import { TransportLayer, TransportMessage, TransportLayerConnection } from '../transport.interface';
import { AmqpStrategyOptions } from './amqp.strategy.interface';
class AmqpStrategyConnection implements TransportLayerConnection {
- private msgSubject$ = new Subject<ConsumeMessage>();
- private resSubject$ = new Subject<ConsumeMessage>();
-
constructor(
private connection: Connection,
private channel: Channel,
- private responseQueue: Replies.AssertQueue,
private options: AmqpStrategyOptions,
) {}
- get message$() {
- return this.msgSubject$.asObservable().pipe(
- share(),
- map(message => ({
- data: message.content,
- replyTo: message.properties.replyTo,
- correlationId: message.properties.correlationId,
- raw: message,
- } as TransportMessage<Buffer>))
- );
- }
-
- get response$() {
- return this.resSubject$.asObservable().pipe(
- share(),
- map(message => ({
- data: message.content,
- replyTo: message.properties.replyTo,
- correlationId: message.properties.correlationId,
- raw: message,
- } as TransportMessage<Buffer>))
- );
- }
-
get close$() {
return fromEvent(this.connection, 'close');
}
@@ -47,39 +19,60 @@ class AmqpStrategyConnection implements TransportLayerConnection {
return fromEvent<Error>(this.connection, 'error');
}
- consumeMessage = async () => {
- await this.channel.consume(
+ consumeMessage = () => {
+ const msgSubject$ = new Subject<ConsumeMessage>();
+
+ this.channel.consume(
this.options.queue,
- msg => msg && this.msgSubject$.next(msg),
+ msg => msg && msgSubject$.next(msg),
+ );
+
+ return msgSubject$.asObservable().pipe(
+ map(message => ({
+ data: message.content,
+ replyTo: message.properties.replyTo,
+ correlationId: message.properties.correlationId,
+ raw: message,
+ } as TransportMessage<Buffer>))
);
- return this;
}
- consumeResponse = async () => {
- await this.channel.consume(
- this.responseQueue.queue,
- res => res && this.resSubject$.next(res),
+ sendMessage = async (queue: string, msg: TransportMessage<Buffer>) => {
+ const { correlationId, data } = msg;
+ const resSubject$ = new Subject<ConsumeMessage>();
+ const replyQueue = await this.channel.assertQueue('', { exclusive: true });
+
+ const consumer = await this.channel.consume(
+ replyQueue.queue,
+ res => res && resSubject$.next(res),
{ noAck: true },
);
- return this;
- }
- sendMessage = (queue: string, msg: TransportMessage<Buffer>, opts: TransportLayerSendOpts = {}) => {
- const { correlationId, replyTo } = msg;
-
- switch (opts.type) {
- case 'publish':
- this.channel.assertExchange(queue, 'fanout', { durable: false });
- return of(this.channel.publish(queue, '', msg.data));
- case 'send':
- this.channel.sendToQueue(queue, msg.data, { correlationId, replyTo: this.responseQueue.queue });
- return this.response$.pipe(filter(m => m.correlationId === correlationId));
- default:
- return of(this.channel.sendToQueue(queue, msg.data, { replyTo, correlationId }));
- }
+ this.channel.sendToQueue(queue, data, { correlationId, replyTo: replyQueue.queue });
+
+ return resSubject$.asObservable().pipe(
+ filter(raw => raw.properties.correlationId === correlationId),
+ take(1),
+ mergeMap(raw => from(this.channel.cancel(consumer.consumerTag)).pipe(
+ mapTo(({
+ data: raw.content,
+ replyTo: raw.properties.replyTo,
+ correlationId: raw.properties.correlationId,
+ raw,
+ } as TransportMessage<Buffer>)),
+ )),
+ ).toPromise();
+ };
+
+ emitMessage = async (queue: string, msg: TransportMessage<Buffer>) => {
+ const { correlationId, data, replyTo } = msg;
+ return Promise.resolve(this.channel.sendToQueue(queue, data, {
+ replyTo,
+ correlationId,
+ }));
};
- ack = (msg: any) => this.channel.ack(msg);
+ ack = (msg: TransportMessage) => this.channel.ack(msg.raw);
getChannel = () => this.options.queue;
@@ -101,12 +94,10 @@ class AmqpStrategy implements TransportLayer {
await channel.prefetch(1);
await channel.assertQueue(queue, queueOptions);
- const responseQueue = await channel.assertQueue('', { exclusive: true });
return new AmqpStrategyConnection(
connection,
channel,
- responseQueue,
this.options,
);
}

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -15,22 +15,16 @@ export interface TransportLayer {
}
export interface TransportLayerConnection {
- sendMessage: (channel: string, message: TransportMessage<Buffer>, opts?: TransportLayerSendOpts) => Observable<any>;
- consumeMessage: () => Promise<TransportLayerConnection>;
- consumeResponse: () => Promise<TransportLayerConnection>;
- ack: (msg: any) => void;
+ sendMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<any>;
+ emitMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<any>;
+ consumeMessage: () => Observable<TransportMessage<Buffer>>;
+ ack: (msg: TransportMessage<any>) => void;
close: () => Promise<any>;
getChannel(): string;
error$: Observable<Error>;
- message$: Observable<TransportMessage<Buffer>>;
- response$: Observable<TransportMessage<Buffer>>;
close$: Observable<any>;
}
-export interface TransportLayerSendOpts {
- type?: 'publish' | 'send' | 'emit';
-}
-
export interface TransportMessageTransformer<T> {
decode: (incomingEvent: Buffer) => T;
encode: (outgoingEvent: T) => Buffer;

refactor(messaging): sendMessage + emitMessage