imp(messaging): error handling + proper job acknowledgement
RENAME
packages/messaging/src/effects/messaging-effects.interface.ts
packages/messaging/src/effects/messaging.effects.interface.ts
@@ -10,9 +10,7 @@ export interface MsgMiddlewareEffect<
export interface MsgErrorEffect<
Err extends Error = Error,
- I = Event,
- O = Event
-> extends MsgEffect<I, O, MsgClient, Err> {}
+> extends MsgEffect<Err, any, MsgClient, Err> {}
export interface MsgEffect<
I = Event,

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
@@ -2,7 +2,7 @@ export { messagingListener } from './listener/messaging.listener';
export * from './client/messaging.client';
export * from './client/messaging.client.interface';
export * from './transport/transport.interface';
-export * from './effects/messaging-effects.interface';
+export * from './effects/messaging.effects.interface';
export * from './server/messaging.server';
export * from './server/messaging.server.interface';
export * from './server/messaging.server.tokens';

MODIFY
packages/messaging/src/listener/messaging.listener.ts
packages/messaging/src/listener/messaging.listener.ts
@@ -5,8 +5,8 @@ import {
combineEffects,
createEffectMetadata,
} from '@marblejs/core';
-import { Observable, Subscription } from 'rxjs';
-import { map, publish, withLatestFrom } from 'rxjs/operators';
+import { Observable, Subscription, Subject, merge } from 'rxjs';
+import { map, publish, withLatestFrom, takeUntil } from 'rxjs/operators';
import {
TransportMessage,
TransportMessageTransformer,
@@ -14,12 +14,13 @@ import {
TransportLayer,
} from '../transport/transport.interface';
import { jsonTransformer } from '../transport/transport.transformer';
-import { MsgEffect, MsgMiddlewareEffect } from '../effects/messaging-effects.interface';
+import { MsgEffect, MsgMiddlewareEffect, MsgErrorEffect } from '../effects/messaging.effects.interface';
import { TransportLayerToken } from '../server/messaging.server.tokens';
export interface MessagingListenerConfig {
effects?: MsgEffect<any, any>[];
middlewares?: MsgMiddlewareEffect<any, any>[];
+ error$?: MsgErrorEffect;
msgTransformer?: TransportMessageTransformer<any>;
}
@@ -27,12 +28,14 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const {
effects = [],
middlewares = [],
+ error$ = (msg$ => msg$) as MsgErrorEffect,
msgTransformer = jsonTransformer,
} = config;
const handleConnection = (conn: TransportLayerConnection, ask: ContextProvider) => {
let effectsSub: Subscription;
+ const errorSubject = new Subject<Error>();
const combinedEffects = combineEffects(...effects);
const combinedMiddlewares = combineMiddlewares(...middlewares);
const defaultMetadata = createEffectMetadata({ ask });
@@ -49,8 +52,16 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
);
- const subscribeEffects = (input$: Observable<TransportMessage<any>>) =>
- input$.subscribe(
+ error$(merge(
+ errorSubject.asObservable(),
+ conn.error$,
+ ), conn, defaultMetadata)
+ .pipe(takeUntil(conn.close$))
+ .subscribe();
+
+ const subscribeEffects = (input$: Observable<TransportMessage<any>>) => input$
+ .pipe(takeUntil(conn.close$))
+ .subscribe(
msg => {
if (msg.replyTo) {
conn.sendMessage(msg.replyTo, {
@@ -59,10 +70,10 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
raw: msg.raw,
});
}
+ conn.ack(msg.raw);
},
error => {
- // @TODO: handle errors
- console.error(error);
+ errorSubject.next(error);
if (effectsSub.closed) { effectsSub = subscribeEffects(message$); }
},
);

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
@@ -7,7 +7,6 @@ import { AmqpStrategyOptions } from './amqp.strategy.interface';
export const createAmqpStrategy = async (options: AmqpStrategyOptions): Promise<TransportLayer> => {
const msgSubject$ = new Subject<ConsumeMessage>();
const resSubject$ = new Subject<ConsumeMessage>();
- const responseQueue = options.queue + '__response';
const message$ = msgSubject$.asObservable().pipe(
share(),
@@ -29,23 +28,20 @@ export const createAmqpStrategy = async (options: AmqpStrategyOptions): Promise<
} as TransportMessage<Buffer>))
);
- const consumeMessage = (channelInstance: Channel) => () => Promise.resolve(
+ const consumeMessage = (channelInstance: Channel) => async () =>
channelInstance.consume(
options.queue,
msg => msg && msgSubject$.next(msg),
- { noAck: true },
- ),
);
- const consumeResponse = (channelInstance: Channel) => () => Promise.resolve(
+ const consumeResponse = (channelInstance: Channel) => (responseQueue: string) => async () =>
channelInstance.consume(
responseQueue,
res => res && resSubject$.next(res),
{ noAck: true },
- ),
);
- const sendMessage = (channelInstance: Channel) => (
+ const sendMessage = (channelInstance: Channel) => (responseQueue: string) => (
queue: string,
msg: TransportMessage<Buffer>,
opts: TransportLayerSendOpts = {},
@@ -57,7 +53,6 @@ export const createAmqpStrategy = async (options: AmqpStrategyOptions): Promise<
channelInstance.assertExchange(queue, 'fanout', { durable: false });
return of(channelInstance.publish(queue, '', msg.data));
case 'send':
- channelInstance.prefetch(1);
channelInstance.sendToQueue(queue, msg.data, { correlationId, replyTo: responseQueue });
return response$.pipe(filter(m => m.correlationId === correlationId));
default:
@@ -65,8 +60,16 @@ export const createAmqpStrategy = async (options: AmqpStrategyOptions): Promise<
}
};
- const close = (connection: Connection) => (channelInstance: Channel) => () =>
- Promise.resolve(channelInstance.close().then(() => connection.close()))
+ const ack = (channelInstance: Channel) => (msg: any) =>
+ channelInstance.ack(msg);
+
+ const close = (connection: Connection) => (channelInstance: Channel) => async () => {
+ await channelInstance.close();
+ await connection.close();
+ }
+
+ const close$ = (connection: Connection) =>
+ fromEvent<Error>(connection, 'close');
const error$ = (connection: Connection) =>
fromEvent<Error>(connection, 'error');
@@ -76,16 +79,19 @@ export const createAmqpStrategy = async (options: AmqpStrategyOptions): Promise<
const connection = await amqplib.connect(options.host);
const channel = await connection.createChannel();
+ await channel.prefetch(1);
await channel.assertQueue(options.queue, options.queueOptions);
- await channel.assertQueue(responseQueue);
+ const responseQueue = await channel.assertQueue('', { exclusive: true });
return {
channel: options.queue,
- sendMessage: sendMessage(channel),
+ ack: ack(channel),
+ sendMessage: sendMessage(channel)(responseQueue.queue),
consumeMessage: consumeMessage(channel),
- consumeResponse: consumeResponse(channel),
+ consumeResponse: consumeResponse(channel)(responseQueue.queue),
close: close(connection)(channel),
error$: error$(connection),
+ close$: close$(connection),
response$,
message$,
};

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

imp(messaging): error handling + proper job acknowledgement