feat(messaging): improved reactivity for connection status
MODIFY
packages/messaging/src/server/messaging.server.events.ts
packages/messaging/src/server/messaging.server.events.ts
import { createEvent, EventsUnion, Event } from '@marblejs/core';
export enum ServerEventType {
- LISTENING = 'listening',
+ STATUS = 'status',
CLOSE = 'close',
ERROR = 'error',
}
export const ServerEvent = {
- listening: createEvent(
- ServerEventType.LISTENING,
- (host: string, channel: string) => ({ host, channel }),
+ status: createEvent(
+ ServerEventType.STATUS,
+ (host: string, channel: string, type: string) => ({ host, channel, type }),
),
close: createEvent(
ServerEventType.CLOSE,
@@ -22,8 +22,8 @@ export const ServerEvent = {
export type AllServerEvents = EventsUnion<typeof ServerEvent>;
-export function isListeningEvent(event: Event): event is ReturnType<typeof ServerEvent.listening> {
- return event.type === ServerEventType.LISTENING;
+export function isStatusEvent(event: Event): event is ReturnType<typeof ServerEvent.status> {
+ return event.type === ServerEventType.STATUS;
}
export function isCloseEvent(event: Event): event is ReturnType<typeof ServerEvent.close> {

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -106,7 +106,10 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const connection = await transportLayer.connect({ isConsumer: true });
handleConnection(connection, serverEventsSubject, ask);
- serverEventsSubject.next(ServerEvent.listening(host, channel));
+
+ connection.status$
+ .pipe(takeUntil(connection.close$))
+ .subscribe(type => serverEventsSubject.next(ServerEvent.status(host,channel, type)));
return connection;
};

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
-import { Subject, fromEvent, from } from 'rxjs';
+import { Subject, fromEvent, from, merge } from 'rxjs';
import { map, filter, take, mergeMap, mapTo, first } from 'rxjs/operators';
import { Channel, ConsumeMessage } from 'amqplib';
import { AmqpConnectionManager, ChannelWrapper } from 'amqp-connection-manager';
import { TransportLayer, TransportMessage, TransportLayerConnection } from '../transport.interface';
import { AmqpStrategyOptions } from './amqp.strategy.interface';
+export enum AmqpConnectionStatus {
+ CONNECTED = 'CONNECTED',
+ CHANNEL_CONNECTED = 'CHANNEL_CONNECTED',
+ CONNECTION_LOST = 'CONNECTION_LOST',
+ CHANNEL_CONNECTION_LOST = 'CHANNEL_CONNECTION_LOST',
+};
+
class AmqpStrategyConnection implements TransportLayerConnection {
private closeSubject$ = new Subject();
@@ -23,6 +30,22 @@ class AmqpStrategyConnection implements TransportLayerConnection {
return fromEvent<Error>(this.channelWrapper, 'error');
}
+ get status$() {
+ const connect$ = fromEvent(this.connectionManager, 'connect')
+ .pipe(mapTo(AmqpConnectionStatus.CONNECTED));
+
+ const connectChannel$ = fromEvent(this.channelWrapper, 'connect')
+ .pipe(mapTo(AmqpConnectionStatus.CHANNEL_CONNECTED));
+
+ const diconnect$ = fromEvent(this.connectionManager, 'disconnect')
+ .pipe(mapTo(AmqpConnectionStatus.CONNECTION_LOST));
+
+ const diconnectChannel$ = fromEvent(this.channelWrapper, 'close')
+ .pipe(mapTo(AmqpConnectionStatus.CHANNEL_CONNECTION_LOST));
+
+ return merge(connect$, connectChannel$, diconnect$, diconnectChannel$);
+ }
+
get message$() {
return this.msgSubject$.asObservable().pipe(
map(message => ({
@@ -144,26 +167,6 @@ class AmqpStrategy implements TransportLayer {
},
});
- connectionManager.on('connect', () => {
- console.log('connected');
- });
-
- connectionManager.on('disconnect', () => {
- console.log('disconnected');
- });
-
- channelWrapper.on('connect', () => {
- console.log('channel', 'connect');
- });
-
- channelWrapper.on('close', () => {
- console.log('channel', 'close');
- });
-
- channelWrapper.on('error', () => {
- console.log('channel', 'error');
- });
-
return new AmqpStrategyConnection(
msgSubject$,
connectionManager,

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -28,6 +28,7 @@ export interface TransportLayerConnection {
close: () => Promise<any>;
getChannel(): string;
message$: Observable<TransportMessage<Buffer>>;
+ status$: Observable<string>;
error$: Observable<Error>;
close$: Observable<any>;
}

feat(messaging): improved reactivity for connection status