feat(messaging): support for transactions
MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -45,16 +45,15 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const errorSubject = new Subject<Error>();
const combinedEffects = combineEffects(...effects);
const combinedMiddlewares = combineMiddlewares(...middlewares);
- const defaultMetadata = createEffectMetadata({ ask });
const message$ = conn.consumeMessage().pipe(
map(msg => ({ ...msg, data: msgTransformer.decode(msg.data) } as TransportMessage<any>)),
mergeMap(msg => of(msg).pipe(
- publish(msg$ => combinedMiddlewares(msg$.pipe(map(m => m.data)), conn, defaultMetadata).pipe(
+ publish(msg$ => combinedMiddlewares(msg$.pipe(map(m => m.data)), conn, createEffectMetadata({ ask, initiator: msg })).pipe(
withLatestFrom(msg$),
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
- publish(msg$ => combinedEffects(msg$.pipe(map(m => m.data)), conn, defaultMetadata).pipe(
+ publish(msg$ => combinedEffects(msg$.pipe(map(m => m.data)), conn, createEffectMetadata({ ask, initiator: msg })).pipe(
withLatestFrom(msg$),
)),
map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.interface.ts
packages/messaging/src/transport/strategies/amqp.strategy.interface.ts
@@ -10,7 +10,6 @@ export interface AmqpStrategyOptions {
host: string;
queue: string;
queueOptions?: Options.AssertQueue;
- socketOptions?: any;
prefetchCount?: number;
- isGlobalPrefetchCount?: boolean;
+ expectAck?: boolean;
}

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
@@ -25,7 +25,9 @@ class AmqpStrategyConnection implements TransportLayerConnection {
this.channel.consume(
this.options.queue,
msg => msg && msgSubject$.next(msg),
- { noAck: true },
+ { noAck: this.options.expectAck !== undefined
+ ? !this.options.expectAck
+ : true },
);
return msgSubject$.asObservable().pipe(
@@ -79,6 +81,18 @@ class AmqpStrategyConnection implements TransportLayerConnection {
}));
};
+ ackMessage = (message: TransportMessage<any> | undefined) => {
+ if (message) {
+ this.channel.ack(message.raw);
+ }
+ }
+
+ nackMessage = (message: TransportMessage<any> | undefined, resend = true) => {
+ if (message) {
+ this.channel.nack(message.raw, false , resend);
+ }
+ }
+
getChannel = () => this.options.queue;
close = async () => {
@@ -98,13 +112,13 @@ class AmqpStrategy implements TransportLayer {
}
async connect() {
- const { host, queue, queueOptions } = this.options;
+ const { host, queue, queueOptions, prefetchCount } = this.options;
const amqplib = await import('amqplib');
const connection = await amqplib.connect(host);
const channel = await connection.createChannel();
- await channel.prefetch(1);
+ await channel.prefetch(prefetchCount || 1);
await channel.assertQueue(queue, queueOptions);
return new AmqpStrategyConnection(

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -24,6 +24,8 @@ export interface TransportLayerConnection {
sendMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<any>;
emitMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<any>;
consumeMessage: () => Observable<TransportMessage<Buffer>>;
+ ackMessage: (message: TransportMessage<any> | undefined) => void;
+ nackMessage: (message: TransportMessage<any> | undefined, resend?: boolean) => void;
close: () => Promise<any>;
getChannel(): string;
error$: Observable<Error>;

feat(messaging): support for transactions