refactor(messaging): class TransportLayer implementation
MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
@@ -21,7 +21,7 @@ export const messagingClient = (config: MessagingClientConfig) => {
const emit = (conn: Promise<TransportLayerConnection>) => <T>(msg: T) =>
from(conn).pipe(
mergeMap(c => c.sendMessage(
- c.channel,
+ c.getChannel(),
{ data: msgTransformer.encode(msg as any) },
{ type: 'emit' },
)),
@@ -32,7 +32,7 @@ export const messagingClient = (config: MessagingClientConfig) => {
const publish = (conn: Promise<TransportLayerConnection>) => <T>(msg: T) =>
from(conn).pipe(
mergeMap(c => c.sendMessage(
- c.channel,
+ c.getChannel(),
{ data: msgTransformer.encode(msg as any) },
{ type: 'publish' },
)),
@@ -43,7 +43,7 @@ export const messagingClient = (config: MessagingClientConfig) => {
const send = (conn: Promise<TransportLayerConnection>) => <T, U>(msg: T): Observable<U> =>
from(conn).pipe(
mergeMap(c => c.sendMessage(
- c.channel,
+ c.getChannel(),
{ data: msgTransformer.encode(msg as any), correlationId: createUuid() },
{ type: 'send' }
)),
@@ -60,24 +60,22 @@ export const messagingClient = (config: MessagingClientConfig) => {
return reader.map(ask => {
const transportLayer = provideTransportLayer(transport, options);
- const conn = transportLayer
- .then(server => server.connect())
- .then(server => server.consumeResponse().then(() => server));
+ const connection = transportLayer.connect().then(conn => conn.consumeResponse());
ask(HttpServerEventStreamToken).map(serverEvent$ =>
serverEvent$.pipe(
matchEvent(ServerEvent.close),
take(1),
- mergeMapTo(conn),
+ mergeMapTo(connection),
mergeMap(conn => conn.close()),
),
).getOrElse(EMPTY).subscribe();
return {
- emit: emit(conn),
- send: send(conn),
- publish: publish(conn),
- close: close(conn),
+ 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
@@ -86,18 +86,15 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
effectsSub = subscribeEffects(message$);
};
- const listen = (transportLayer: Promise<TransportLayer>, ask: ContextProvider) => async () => {
- const layer = await transportLayer;
- const conn = await layer.connect();
- await conn.consumeMessage();
- await handleConnection(conn, ask);
- return conn;
+ const listen = (transportLayer: TransportLayer, ask: ContextProvider) => async () => {
+ const connection = await transportLayer.connect();
+ await connection.consumeMessage();
+ await handleConnection(connection, ask);
+ return connection;
};
return reader.map(ask => {
- const transportLayer = ask(TransportLayerToken)
- .map(async layer => await layer)
- .getOrElse(undefined as unknown as Promise<TransportLayer>);
+ const transportLayer = ask(TransportLayerToken).getOrElse(undefined as unknown as TransportLayer);
return {
listen: listen(transportLayer, ask),

MODIFY
packages/messaging/src/server/messaging.server.tokens.ts
packages/messaging/src/server/messaging.server.tokens.ts
import { createContextToken } from '@marblejs/core';
import { TransportLayer } from '../transport/transport.interface';
-export const TransportLayerToken = createContextToken<Promise<TransportLayer>>();
+export const TransportLayerToken = createContextToken<TransportLayer>();

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 } from 'amqplib';
-import { TransportLayer, TransportLayerSendOpts, TransportMessage } from '../transport.interface';
+import { Channel, Connection, ConsumeMessage, Replies } from 'amqplib';
+import { TransportLayer, TransportLayerSendOpts, TransportMessage, TransportLayerConnection } from '../transport.interface';
import { AmqpStrategyOptions } from './amqp.strategy.interface';
-export const createAmqpStrategy = async (options: AmqpStrategyOptions): Promise<TransportLayer> => {
- const msgSubject$ = new Subject<ConsumeMessage>();
- const resSubject$ = new Subject<ConsumeMessage>();
+class AmqpStrategyConnection implements TransportLayerConnection {
+ private msgSubject$ = new Subject<ConsumeMessage>();
+ private resSubject$ = new Subject<ConsumeMessage>();
- const message$ = msgSubject$.asObservable().pipe(
+ 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,
@@ -17,8 +25,10 @@ export const createAmqpStrategy = async (options: AmqpStrategyOptions): Promise<
raw: message,
} as TransportMessage<Buffer>))
);
+ }
- const response$ = resSubject$.asObservable().pipe(
+ get response$() {
+ return this.resSubject$.asObservable().pipe(
share(),
map(message => ({
data: message.content,
@@ -27,75 +37,80 @@ export const createAmqpStrategy = async (options: AmqpStrategyOptions): Promise<
raw: message,
} as TransportMessage<Buffer>))
);
+ }
+
+ get close$() {
+ return fromEvent(this.connection, 'close');
+ }
- const consumeMessage = (channelInstance: Channel) => async () =>
- channelInstance.consume(
- options.queue,
- msg => msg && msgSubject$.next(msg),
+ get error$() {
+ return fromEvent<Error>(this.connection, 'error');
+ }
+
+ consumeMessage = async () => {
+ await this.channel.consume(
+ this.options.queue,
+ msg => msg && this.msgSubject$.next(msg),
);
+ return this;
+ }
- const consumeResponse = (channelInstance: Channel) => (responseQueue: string) => async () =>
- channelInstance.consume(
- responseQueue,
- res => res && resSubject$.next(res),
+ consumeResponse = async () => {
+ await this.channel.consume(
+ this.responseQueue.queue,
+ res => res && this.resSubject$.next(res),
{ noAck: true },
);
+ return this;
+ }
- const sendMessage = (channelInstance: Channel) => (responseQueue: string) => (
- queue: string,
- msg: TransportMessage<Buffer>,
- opts: TransportLayerSendOpts = {},
- ) => {
+ sendMessage = (queue: string, msg: TransportMessage<Buffer>, opts: TransportLayerSendOpts = {}) => {
const { correlationId, replyTo } = msg;
switch (opts.type) {
case 'publish':
- channelInstance.assertExchange(queue, 'fanout', { durable: false });
- return of(channelInstance.publish(queue, '', msg.data));
+ this.channel.assertExchange(queue, 'fanout', { durable: false });
+ return of(this.channel.publish(queue, '', msg.data));
case 'send':
- channelInstance.sendToQueue(queue, msg.data, { correlationId, replyTo: responseQueue });
- return response$.pipe(filter(m => m.correlationId === correlationId));
+ this.channel.sendToQueue(queue, msg.data, { correlationId, replyTo: this.responseQueue.queue });
+ return this.response$.pipe(filter(m => m.correlationId === correlationId));
default:
- return of(channelInstance.sendToQueue(queue, msg.data, { replyTo, correlationId }));
+ return of(this.channel.sendToQueue(queue, msg.data, { replyTo, correlationId }));
}
};
- const ack = (channelInstance: Channel) => (msg: any) =>
- channelInstance.ack(msg);
+ ack = (msg: any) => this.channel.ack(msg);
- const close = (connection: Connection) => (channelInstance: Channel) => async () => {
- await channelInstance.close();
- await connection.close();
+ getChannel = () => this.options.queue;
+
+ close = async () => {
+ await this.channel.close();
+ await this.connection.close();
+ }
}
- const close$ = (connection: Connection) =>
- fromEvent<Error>(connection, 'close');
+class AmqpStrategy implements TransportLayer {
+ constructor(private options: AmqpStrategyOptions) {}
- const error$ = (connection: Connection) =>
- fromEvent<Error>(connection, 'error');
+ async connect() {
+ const { host, queue, queueOptions } = this.options;
- const connect = async () => {
const amqplib = await import('amqplib');
- const connection = await amqplib.connect(options.host);
+ const connection = await amqplib.connect(host);
const channel = await connection.createChannel();
await channel.prefetch(1);
- await channel.assertQueue(options.queue, options.queueOptions);
+ await channel.assertQueue(queue, queueOptions);
const responseQueue = await channel.assertQueue('', { exclusive: true });
- return {
- channel: options.queue,
- ack: ack(channel),
- sendMessage: sendMessage(channel)(responseQueue.queue),
- consumeMessage: consumeMessage(channel),
- consumeResponse: consumeResponse(channel)(responseQueue.queue),
- close: close(connection)(channel),
- error$: error$(connection),
- close$: close$(connection),
- response$,
- message$,
- };
- };
+ return new AmqpStrategyConnection(
+ connection,
+ channel,
+ responseQueue,
+ this.options,
+ );
+ }
+}
- return { connect };
-};
+export const createAmqpStrategy = (options: AmqpStrategyOptions): TransportLayer =>
+ new AmqpStrategy(options);

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -16,11 +16,11 @@ export interface TransportLayer {
export interface TransportLayerConnection {
sendMessage: (channel: string, message: TransportMessage<Buffer>, opts?: TransportLayerSendOpts) => Observable<any>;
- consumeMessage: () => Promise<any>;
- consumeResponse: () => Promise<any>;
+ consumeMessage: () => Promise<TransportLayerConnection>;
+ consumeResponse: () => Promise<TransportLayerConnection>;
ack: (msg: any) => void;
close: () => Promise<any>;
- channel: string;
+ getChannel(): string;
error$: Observable<Error>;
message$: Observable<TransportMessage<Buffer>>;
response$: Observable<TransportMessage<Buffer>>;

MODIFY
packages/messaging/src/transport/transport.provider.ts
packages/messaging/src/transport/transport.provider.ts
@@ -5,7 +5,7 @@ import { createAmqpStrategy } from './strategies/amqp.strategy';
import { createTcpStrategy } from './strategies/tcp.strategy';
import { createNatsStrategy } from './strategies/nats.strategy';
-export const provideTransportLayer = (transport: Transport, transportOptions: any): Promise<TransportLayer> => {
+export const provideTransportLayer = (transport: Transport, transportOptions: any): TransportLayer => {
switch (transport) {
case Transport.AMQP:
return createAmqpStrategy(transportOptions);

refactor(messaging): class TransportLayer implementation