fix(messaging): messages are consumed before full startup
MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
@@ -197,4 +197,28 @@ describe('messagingServer::AMQP', () => {
await client.emit({ type: 'TEST' });
});
+
+ test('consumes pending messages after startup', async done => {
+ // given
+ const options = Util.createAmqpOptions();
+ const replyTo = createUuid();
+
+ const test$: MsgEffect = event$ =>
+ event$.pipe(
+ matchEvent('TEST'),
+ mapTo(reply(replyTo)({ type: 'TEST_RESULT' })),
+ );
+
+ // then
+ const output$ = Util.assertOutputEvent({
+ type: 'TEST_RESULT',
+ })(done);
+
+ // when (order matters)
+ client = await Util.createAmqpClient(options);
+
+ await client.emit({ type: 'TEST' });
+
+ microservice = await Util.createAmqpMicroservice(options)({ effects: [test$], output$ });
+ });
});

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
@@ -7,16 +7,20 @@ import { TransportLayer, TransportMessage, TransportLayerConnection, Transport,
import { AmqpStrategyOptions, AmqpConnectionStatus, AmqpCannotSetExpectAckForNonConsumerConnection } from './amqp.strategy.interface';
class AmqpStrategyConnection implements TransportLayerConnection {
+ private msgSubject$ = new Subject<ConsumeMessage>();
private statusSubject$ = new Subject<AmqpConnectionStatus>();
private closeSubject$ = new Subject();
constructor(
- private msgSubject$: Subject<ConsumeMessage>,
+ isConsumer: boolean,
private connectionManager: AmqpConnectionManager,
private channelWrapper: ChannelWrapper,
private options: AmqpStrategyOptions,
) {
- process.nextTick(() => this.statusSubject$.next(AmqpConnectionStatus.CONNECTED));
+ process.nextTick(async () => {
+ if (isConsumer) await this.consumeMessages();
+ this.statusSubject$.next(AmqpConnectionStatus.CONNECTED);
+ });
}
get close$() {
@@ -55,6 +59,15 @@ class AmqpStrategyConnection implements TransportLayerConnection {
);
}
+ private consumeMessages = async () =>
+ this.channelWrapper.addSetup(async (channel: Channel) => {
+ await channel.consume(
+ this.options.queue,
+ msg => msg && this.msgSubject$.next(msg),
+ { noAck: !this.options.expectAck },
+ );
+ });
+
sendMessage = async (queue: string, msg: TransportMessage<Buffer>) => {
const { data } = msg;
const correlationId = createUuid();
@@ -160,9 +173,9 @@ class AmqpStrategy implements TransportLayer {
async connect(opts?: { isConsumer: boolean }) {
const { host, queue, queueOptions, prefetchCount, expectAck } = this.options;
- const msgSubject$ = new Subject<ConsumeMessage>();
+ const isConsumer = !!opts?.isConsumer;
- if (!opts?.isConsumer && expectAck)
+ if (!isConsumer && expectAck)
throw new AmqpCannotSetExpectAckForNonConsumerConnection();
await import('amqplib');
@@ -175,23 +188,13 @@ class AmqpStrategy implements TransportLayer {
setup: async (channel: Channel) => {
await channel.prefetch(prefetchCount || 1);
await channel.assertQueue(queue, queueOptions);
-
- if (opts?.isConsumer) {
- await channel.consume(
- this.options.queue,
- msg => msg && msgSubject$.next(msg),
- { noAck: this.options.expectAck !== undefined
- ? !this.options.expectAck
- : true },
- )
- }
},
});
await (channelWrapper as any).waitForConnect(); // not available in @types/amqp-connection-manager
return new AmqpStrategyConnection(
- msgSubject$,
+ isConsumer,
connectionManager,
channelWrapper,
this.options,

fix(messaging): messages are consumed before full startup