feat(messaging): handle server events
MODIFY
packages/messaging/src/effects/messaging.effects.interface.ts
packages/messaging/src/effects/messaging.effects.interface.ts
@@ -18,3 +18,6 @@ export interface MsgEffect<
Client = MsgClient,
Err extends Error = Error,
> extends Effect<I, O, Client, Err> {}
+
+export interface MsgServerEffect<T extends Event = Event>
+ extends MsgEffect<T, any, undefined> {}

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
@@ -5,4 +5,5 @@ export * from './effects/messaging.effects.interface';
export * from './server/messaging.server';
export * from './server/messaging.server.interface';
export * from './server/messaging.server.tokens';
+export * from './server/messaging.server.events';
export { messagingListener } from './server/messaging.server.listener';

ADD
packages/messaging/src/server/messaging.server.events.ts
+import { createEvent, EventsUnion, Event } from '@marblejs/core';
+
+export enum ServerEventType {
+ LISTENING = 'listening',
+ CLOSE = 'close',
+}
+
+export const ServerEvent = {
+ listening: createEvent(
+ ServerEventType.LISTENING,
+ (host: string, channel: string) => ({ host, channel }),
+ ),
+ close: createEvent(
+ ServerEventType.CLOSE,
+ ),
+};
+
+export type AllServerEvents = EventsUnion<typeof ServerEvent>;
+
+export function isListeningEvent(event: Event): event is ReturnType<typeof ServerEvent.listening> {
+ return event.type === ServerEventType.LISTENING;
+}
+
+export function isCloseEvent(event: Event): event is ReturnType<typeof ServerEvent.close> {
+ return event.type === ServerEventType.CLOSE;
+}

MODIFY
packages/messaging/src/server/messaging.server.interface.ts
packages/messaging/src/server/messaging.server.interface.ts
@@ -3,10 +3,12 @@ import { NatsStrategy } from '../transport/strategies/nats.strategy.interface';
import { TcpStrategy } from '../transport/strategies/tcp.strategy.interface';
import { AmqpStrategy } from '../transport/strategies/amqp.strategy.interface';
import { messagingListener } from './messaging.server.listener';
+import { MsgServerEffect } from '../effects/messaging.effects.interface';
type ConfigurationBase = {
messagingListener: ReturnType<typeof messagingListener>;
dependencies?: BoundDependency<any>[];
+ event$?: MsgServerEffect;
}
export type CreateMicroserviceConfig =

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -6,7 +6,7 @@ import {
createEffectMetadata,
} from '@marblejs/core';
import { Observable, Subscription, Subject, of } from 'rxjs';
-import { map, publish, withLatestFrom, takeUntil, catchError, mergeMap } from 'rxjs/operators';
+import { map, publish, withLatestFrom, takeUntil, catchError, mergeMap, take } from 'rxjs/operators';
import {
TransportMessage,
TransportMessageTransformer,
@@ -15,7 +15,8 @@ import {
} from '../transport/transport.interface';
import { jsonTransformer } from '../transport/transport.transformer';
import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
-import { TransportLayerToken } from './messaging.server.tokens';
+import { TransportLayerToken, ServerEventsToken } from './messaging.server.tokens';
+import { AllServerEvents, ServerEvent } from './messaging.server.events';
export interface MessagingListenerConfig {
effects?: MsgEffect<any, any>[];
@@ -32,7 +33,11 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
msgTransformer = jsonTransformer,
} = config;
- const handleConnection = (conn: TransportLayerConnection, ask: ContextProvider) => {
+ const handleConnection = (
+ conn: TransportLayerConnection,
+ serverEventsSubject: Subject<AllServerEvents>,
+ ask: ContextProvider,
+ ) => {
let effectsSub: Subscription;
const errorSubject = new Subject<Error>();
@@ -79,13 +84,20 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
onSubscribeEffectsError(errorSubject),
);
+ conn.close$
+ .pipe(take(1))
+ .subscribe(() => serverEventsSubject.next(ServerEvent.close()));
+
effectsSub = subscribeEffects(message$);
};
const listen = (transportLayer: TransportLayer, ask: ContextProvider) => async () => {
+ const { host, channel } = transportLayer.config;
+ const serverEventsSubject = ask(ServerEventsToken).getOrElse(undefined as unknown as Subject<AllServerEvents>);
const connection = await transportLayer.connect();
- handleConnection(connection, ask);
+ handleConnection(connection, serverEventsSubject, ask);
+ serverEventsSubject.next(ServerEvent.listening(host, channel));
return connection;
};

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

MODIFY
packages/messaging/src/server/messaging.server.ts
packages/messaging/src/server/messaging.server.ts
-import { createContext, registerAll, bindTo } from '@marblejs/core';
+import { createContext, registerAll, bindTo, createEffectMetadata, lookup } from '@marblejs/core';
+import { Subject } from 'rxjs';
+import { takeWhile } from 'rxjs/operators';
import { CreateMicroserviceConfig } from './messaging.server.interface';
import { provideTransportLayer } from '../transport/transport.provider';
-import { TransportLayerToken } from './messaging.server.tokens';
+import { TransportLayerToken, ServerEventsToken } from './messaging.server.tokens';
+import { AllServerEvents, isCloseEvent } from './messaging.server.events';
export const createMicroservice = (config: CreateMicroserviceConfig) => {
const {
+ event$,
options,
transport,
dependencies = [],
messagingListener,
} = config;
+ const serverEventSubject = new Subject<AllServerEvents>();
const transportLayer = provideTransportLayer(transport, options);
const boundTransportLayer = bindTo(TransportLayerToken)(() => transportLayer);
- const context = registerAll([ boundTransportLayer, ...dependencies ])(createContext());
+ const boundServerEvents = bindTo(ServerEventsToken)(() => serverEventSubject);
+ const context = registerAll([ boundTransportLayer, boundServerEvents, ...dependencies ])(createContext());
const listenerWithContext = messagingListener.run(context);
+ if (event$) {
+ const serverEvent$ = serverEventSubject.asObservable().pipe(takeWhile(e => !isCloseEvent(e)));
+ const metadata = createEffectMetadata({ ask: lookup(context) });
+ event$(serverEvent$, undefined, metadata).subscribe();
+ }
+
return {
run: listenerWithContext.listen,
}

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
@@ -90,6 +90,13 @@ class AmqpStrategyConnection implements TransportLayerConnection {
class AmqpStrategy implements TransportLayer {
constructor(private options: AmqpStrategyOptions) {}
+ get config() {
+ return ({
+ host: this.options.host,
+ channel: this.options.queue,
+ });
+ }
+
async connect() {
const { host, queue, queueOptions } = this.options;

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -12,6 +12,12 @@ export enum Transport {
export interface TransportLayer {
connect: () => Promise<TransportLayerConnection>;
+ config: TransportLayerConfig;
+}
+
+export interface TransportLayerConfig {
+ host: string;
+ channel: string;
}
export interface TransportLayerConnection {

feat(messaging): handle server events