fix(messaging): connection closing
MODIFY
packages/messaging/src/client/messaging.client.interface.ts
packages/messaging/src/client/messaging.client.interface.ts
@@ -4,4 +4,5 @@ export interface MessagingClient {
emit: <T>(data: T) => Observable<any>;
send: <T, U>(data: T) => Observable<U>;
publish: <T>(data: T) => Observable<any>;
+ close: () => Observable<any>;
}

MODIFY
packages/messaging/src/client/messaging.client.ts
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 { mergeMap, take, map, tap, mapTo } from 'rxjs/operators';
import {
Transport,
TransportMessage,
@@ -32,6 +32,7 @@ export const messagingClient = (config: MessagingClientConfig = {}) => {
{ data: msgTransformer.encode(msg as any) },
{ type: 'emit' },
)),
+ mapTo(true),
take(1),
);
@@ -42,6 +43,7 @@ export const messagingClient = (config: MessagingClientConfig = {}) => {
{ data: msgTransformer.encode(msg as any) },
{ type: 'publish' },
)),
+ mapTo(true),
take(1),
);
@@ -57,6 +59,12 @@ export const messagingClient = (config: MessagingClientConfig = {}) => {
take(1),
);
+ const close = (conn$: Observable<TransportLayerConnection>) => () =>
+ conn$.pipe(
+ mergeMap(conn => conn.close()),
+ take(1),
+ );
+
return reader.map(() => {
const transportLayer = provideTransportLayer(transport, options);
const conn$ = from(transportLayer).pipe(
@@ -68,6 +76,7 @@ export const messagingClient = (config: MessagingClientConfig = {}) => {
emit: emit(conn$),
send: send(conn$),
publish: publish(conn$),
+ close: close(conn$),
} as MessagingClient;
});
};

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
export { messagingListener } from './listener/messaging.listener';
+export * from './client/messaging.client';
export * from './transport/transport.interface';
export * from './effects/messaging-effects.interface';

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 { EMPTY, Subject, fromEvent } from 'rxjs';
import { map, share, filter } from 'rxjs/operators';
import { Channel, Connection, ConsumeMessage, Options } from 'amqplib';
import { TransportLayer, TransportLayerSendOpts, TransportMessage } from '../transport.interface';
@@ -37,21 +37,21 @@ export const createAmqpStrategy = async (options: RmqStrategyOptions): Promise<T
} as TransportMessage<Buffer>))
);
- const consumeMessage = (channelInstance: Channel) => () => {
+ const consumeMessage = (channelInstance: Channel) => () => Promise.resolve(
channelInstance.consume(
options.queue,
msg => msg && msgSubject$.next(msg),
{ noAck: true },
+ ),
);
- };
- const consumeResponse = (channelInstance: Channel) => () => {
+ const consumeResponse = (channelInstance: Channel) => () => Promise.resolve(
channelInstance.consume(
responseQueue,
res => res && resSubject$.next(res),
{ noAck: true },
+ ),
);
- };
const sendMessage = (channelInstance: Channel) => (
queue: string,
@@ -75,8 +75,8 @@ export const createAmqpStrategy = async (options: RmqStrategyOptions): Promise<T
}
};
- const close = (connection: Connection) => () =>
- from(connection.close());
+ const close = (connection: Connection) => (channelInstance: Channel) => () =>
+ Promise.resolve(channelInstance.close().then(() => connection.close()))
const error$ = (connection: Connection) =>
fromEvent<Error>(connection, 'error');
@@ -86,14 +86,14 @@ export const createAmqpStrategy = async (options: RmqStrategyOptions): Promise<T
const connection = await amqplib.connect(options.host);
const channel = await connection.createChannel();
- channel.assertQueue(options.queue, options.queueOptions);
- channel.assertQueue(responseQueue);
+ await channel.assertQueue(options.queue, options.queueOptions);
+ await channel.assertQueue(responseQueue);
return {
sendMessage: sendMessage(channel),
consumeMessage: consumeMessage(channel),
consumeResponse: consumeResponse(channel),
- close: close(connection),
+ close: close(connection)(channel),
error$: error$(connection),
response$,
message$,

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

fix(messaging): connection closing