feat(messaging): amqp-connection-manager for handling connection issues (version 2)
MODIFY
packages/messaging/package.json
packages/messaging/package.json
"@marblejs/core": "^2.3.1",
"@marblejs/middleware-io": "^2.3.1",
"amqplib": "^0.5.3",
- "@types/amqplib": "^0.5.11"
+ "amqp-connection-manager": "^3.0.0",
+ "@types/amqplib": "^0.5.11",
+ "@types/amqp-connection-manager": "^2.0.4"
},
"publishConfig": {
"access": "public"

MODIFY
packages/messaging/src/client/messaging.client.spec.ts
packages/messaging/src/client/messaging.client.spec.ts
@@ -17,7 +17,7 @@ describe('#messagingClient', () => {
};
const runClient = () => messagingClient(clientOptions).run(createContext());
- const runServer = () => createAmqpStrategy(clientOptions.options).connect();
+ const runServer = () => createAmqpStrategy(clientOptions.options).connect({ isConsumer: true });
const createMessage = (data: any, correlationId?: string): TransportMessage<Buffer> => ({
data: Buffer.from(JSON.stringify(data)),
correlationId: correlationId,
@@ -27,7 +27,7 @@ describe('#messagingClient', () => {
const client = runClient();
const server = await runServer();
- server.consumeMessage().subscribe(async msg => {
+ server.message$.subscribe(async msg => {
expect(msg.data.toString()).toEqual(JSON.stringify({ test: 'test' }));
await server.close();
done();
@@ -42,7 +42,7 @@ describe('#messagingClient', () => {
const client = runClient();
const server = await runServer();
- server.consumeMessage().subscribe(async msg => {
+ server.message$.subscribe(async msg => {
const count = Number(msg.data.toString()) + 1;
const response = createMessage(count, msg.correlationId);

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
@@ -51,7 +51,7 @@ export const messagingClient = (config: MessagingClientConfig) => {
return reader.map(ask => {
const transportLayer = provideTransportLayer(transport, options);
- const connection = transportLayer.connect();
+ const connection = transportLayer.connect({ isConsumer: false });
ask(serverEvent$)
.map(teardownOnClose$(connection))

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -46,7 +46,7 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
const combinedEffects = combineEffects(...effects);
const combinedMiddlewares = combineMiddlewares(...middlewares);
- const message$ = conn.consumeMessage().pipe(
+ const message$ = conn.message$.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, createEffectMetadata({ ask, initiator: msg })).pipe(
@@ -103,7 +103,7 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
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();
+ const connection = await transportLayer.connect({ isConsumer: true });
handleConnection(connection, serverEventsSubject, ask);
serverEventsSubject.next(ServerEvent.listening(host, channel));

MODIFY
packages/messaging/src/server/messaging.server.spec.ts
packages/messaging/src/server/messaging.server.spec.ts
@@ -25,7 +25,7 @@ describe('messagingServer', () => {
}).run();
const runClient = () =>
- createAmqpStrategy(options).connect();
+ createAmqpStrategy(options).connect({ isConsumer: false });
const createMessage = (data: any): TransportMessage<Buffer> => ({
data: Buffer.from(JSON.stringify(data)),
@@ -55,6 +55,11 @@ describe('messagingServer', () => {
test('emits event to consumer', async done => {
const eventSubject = new Subject();
+ eventSubject.subscribe(event => {
+ expect(event).toEqual({ type: 'EVENT_TEST_RESPONSE', payload: 2 });
+ setTimeout(() => server.close().then(done), 1000);
+ });
+
const event$: MsgEffect = event$ =>
event$.pipe(
matchEvent('EVENT_TEST'),
@@ -68,12 +73,9 @@ describe('messagingServer', () => {
const client = await runClient();
const message = createMessage({ type: 'EVENT_TEST', payload: 1 });
- await client.emitMessage(options.queue, message);
+ const emitResult = await client.emitMessage(options.queue, message);
- eventSubject.subscribe(event => {
- expect(event).toEqual({ type: 'EVENT_TEST_RESPONSE', payload: 2 });
- setTimeout(() => server.close().then(done), 1000);
- });
+ expect(emitResult).toEqual(true);
});
test('reacts to thrown error', async done => {
@@ -81,6 +83,12 @@ describe('messagingServer', () => {
const expectedError = new Error('test_error');
const errorSubject = new Subject<[Event, Error | undefined]>();
+ errorSubject.subscribe(data => {
+ expect(data[0]).toEqual(expectedMessage);
+ expect(data[1]).toEqual(expectedError);
+ setTimeout(() => server.close().then(done), 1000);
+ });
+
const event$: MsgEffect = event$ =>
event$.pipe(
matchEvent('EVENT_TEST'),
@@ -95,13 +103,9 @@ describe('messagingServer', () => {
const server = await runServer(event$, error$);
const client = await runClient();
- await client.emitMessage(options.queue, createMessage(expectedMessage));
+ const emitResult = await client.emitMessage(options.queue, createMessage(expectedMessage));
- errorSubject.subscribe(data => {
- expect(data[0]).toEqual(expectedMessage);
- expect(data[1]).toEqual(expectedError);
- setTimeout(() => server.close().then(done), 1000);
- });
+ expect(emitResult).toEqual(true);
});
});

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
import { Subject, fromEvent, from } from 'rxjs';
-import { map, filter, take, mergeMap, mapTo } from 'rxjs/operators';
-import { Channel, Connection, ConsumeMessage } from 'amqplib';
+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';
class AmqpStrategyConnection implements TransportLayerConnection {
+ private closeSubject$ = new Subject();
+
constructor(
- private connection: Connection,
- private channel: Channel,
+ private msgSubject$: Subject<ConsumeMessage>,
+ private connectionManager: AmqpConnectionManager,
+ private channelWrapper: ChannelWrapper,
private options: AmqpStrategyOptions,
) {}
get close$() {
- return fromEvent(this.connection, 'close');
+ return this.closeSubject$.asObservable();
}
get error$() {
- return fromEvent<Error>(this.connection, 'error');
+ return fromEvent<Error>(this.channelWrapper, 'error');
}
- consumeMessage = () => {
- const msgSubject$ = new Subject<ConsumeMessage>();
-
- this.channel.consume(
- this.options.queue,
- msg => msg && msgSubject$.next(msg),
- { noAck: this.options.expectAck !== undefined
- ? !this.options.expectAck
- : true },
- );
-
- return msgSubject$.asObservable().pipe(
+ get message$() {
+ return this.msgSubject$.asObservable().pipe(
map(message => ({
data: message.content,
replyTo: message.properties.replyTo,
correlationId: message.properties.correlationId,
raw: message,
- } as TransportMessage<Buffer>))
+ } as TransportMessage<Buffer>)),
);
}
sendMessage = async (queue: string, msg: TransportMessage<Buffer>) => {
+ const replyToSubject = new Subject<string>();
+
const { correlationId, data } = msg;
- const resSubject$ = new Subject<ConsumeMessage>();
- const replyQueue = await this.channel.assertQueue('', {
+ const resSubject$ = new Subject<{ msg: ConsumeMessage; tag: string }>();
+
+ replyToSubject
+ .pipe(first())
+ .subscribe(async replyTo => {
+ await this.channelWrapper.sendToQueue(queue, data, {
+ correlationId,
+ replyTo,
+ });
+ });
+
+ await this.channelWrapper.addSetup(async (channel: Channel) => {
+ const replyQueue = await channel.assertQueue('', {
exclusive: true,
autoDelete: true,
});
- const consumer = await this.channel.consume(
+ const consumer = await channel.consume(
replyQueue.queue,
- res => res && resSubject$.next(res),
+ msg => msg && resSubject$.next({ msg, tag: consumer.consumerTag }),
{ noAck: true },
);
- this.channel.sendToQueue(queue, data, {
- correlationId,
- replyTo: replyQueue.queue,
+ replyToSubject.next(replyQueue.queue);
});
return resSubject$.asObservable().pipe(
- filter(raw => raw.properties.correlationId === correlationId),
+ filter(raw => raw.msg.properties.correlationId === correlationId),
take(1),
- mergeMap(raw => from(this.channel.cancel(consumer.consumerTag)).pipe(
+ mergeMap(raw => from(this.channelWrapper.addSetup((channel: Channel) => channel.cancel(raw.tag))).pipe(
mapTo(({
- data: raw.content,
- replyTo: raw.properties.replyTo,
- correlationId: raw.properties.correlationId,
+ data: raw.msg.content,
+ replyTo: raw.msg.properties.replyTo,
+ correlationId: raw.msg.properties.correlationId,
raw,
} as TransportMessage<Buffer>)),
)),
@@ -75,29 +80,30 @@ class AmqpStrategyConnection implements TransportLayerConnection {
emitMessage = async (queue: string, msg: TransportMessage<Buffer>) => {
const { correlationId, data, replyTo } = msg;
- return Promise.resolve(this.channel.sendToQueue(queue, data, {
+ return this.channelWrapper.sendToQueue(queue, data, {
replyTo,
correlationId,
- }));
+ });
};
ackMessage = (message: TransportMessage<any> | undefined) => {
if (message) {
- this.channel.ack(message.raw);
+ this.channelWrapper.ack(message.raw);
}
}
nackMessage = (message: TransportMessage<any> | undefined, resend = true) => {
if (message) {
- this.channel.nack(message.raw, false , resend);
+ this.channelWrapper.nack(message.raw, false , resend);
}
}
getChannel = () => this.options.queue;
close = async () => {
- await this.channel.close();
- await this.connection.close();
+ await this.channelWrapper.close();
+ await this.connectionManager.close();
+ this.closeSubject$.next();
}
}
@@ -111,19 +117,57 @@ class AmqpStrategy implements TransportLayer {
});
}
- async connect() {
+ async connect(opts: { isConsumer: boolean }) {
const { host, queue, queueOptions, prefetchCount } = this.options;
+ const msgSubject$ = new Subject<ConsumeMessage>();
+
+ await import('amqplib');
- const amqplib = await import('amqplib');
- const connection = await amqplib.connect(host);
- const channel = await connection.createChannel();
+ const amqplib = await import('amqp-connection-manager');
+ const connectionManager = await amqplib.connect([host]);
+ const channelWrapper = connectionManager.createChannel({
+ json: false,
+ 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 },
+ )
+ }
+ },
+ });
+
+ 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(
- connection,
- channel,
+ msgSubject$,
+ connectionManager,
+ channelWrapper,
this.options,
);
}

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -11,7 +11,7 @@ export enum Transport {
}
export interface TransportLayer {
- connect: () => Promise<TransportLayerConnection>;
+ connect: (opts: { isConsumer: boolean }) => Promise<TransportLayerConnection>;
config: TransportLayerConfig;
}
@@ -21,13 +21,13 @@ export interface TransportLayerConfig {
}
export interface TransportLayerConnection {
- sendMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<any>;
+ sendMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<TransportMessage<Buffer>>;
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;
+ message$: Observable<TransportMessage<Buffer>>;
error$: Observable<Error>;
close$: Observable<any>;
}

MODIFY
scripts/wait.js
scripts/wait.js
@@ -8,7 +8,8 @@ const log = msg => console.info(chalk.yellow(msg));
const waitForRabbitMq = async () => {
try {
- await amqplib.connect('amqp://localhost:5672');
+ const conn = await amqplib.connect('amqp://localhost:5672');
+ const channel = await conn.createChannel();
process.exit();
} catch(err) {
log(' -- Waiting for RabbitMQ to be ready...');

MODIFY
yarn.lock
yarn.lock
dependencies:
any-observable "^0.3.0"
-"@types/amqplib@^0.5.11":
+"@types/amqp-connection-manager@^2.0.4":
+ version "2.0.4"
+ resolved "https://registry.yarnpkg.com/@types/amqp-connection-manager/-/amqp-connection-manager-2.0.4.tgz#ba898c236e67748982dc11bb77998df963dd240b"
+ integrity sha512-rfbH2oFMbd5vuhrzNqvNnJ07WVh91RKDJnKZkAOCV4XTM72x5lOBXgl93XEiEB/hI9PZeOmEYj16xnqvsE3zyw==
+ dependencies:
+ "@types/amqplib" "*"
+
+"@types/amqplib@*", "@types/amqplib@^0.5.11":
version "0.5.13"
resolved "https://registry.yarnpkg.com/@types/amqplib/-/amqplib-0.5.13.tgz#a0dc9be08f38b91e5f7532acf0c89b0e515cdc6c"
integrity sha512-0r8cJfUajAlcHxlJsZyqRrSAYzv+pJPsaAQjDC6e5rACi3AOIuR0AfevJG7NLvw3Qu4EGhf3KS/ECwd5cUAY2A==
@@ -1099,6 +1106,13 @@ ajv@^6.9.1:
json-schema-traverse "^0.4.1"
uri-js "^4.2.2"
+amqp-connection-manager@^3.0.0:
+ version "3.0.0"
+ resolved "https://registry.yarnpkg.com/amqp-connection-manager/-/amqp-connection-manager-3.0.0.tgz#9e273c616122033ca802f962b4d0e48895a5ff9a"
+ integrity sha512-a5MsUDsG+CqMjwk/WNFSTE0H4pAaWJXw7L24QFa3MeaB+KA05PXoBsppYlIzaIqc1XLWZwjO9J42AFHNrDsVFQ==
+ dependencies:
+ promise-breaker "^5.0.0"
+
amqplib@^0.5.3:
version "0.5.3"
resolved "https://registry.yarnpkg.com/amqplib/-/amqplib-0.5.3.tgz#7ccfc85d12ee7cd3c6dc861bb07f0648ec3d7193"
@@ -5630,6 +5644,11 @@ progress@^2.0.0:
resolved "https://registry.yarnpkg.com/progress/-/progress-2.0.3.tgz#7e8cf8d8f5b8f239c1bc68beb4eb78567d572ef8"
integrity sha512-7PiHtLll5LdnKIMw100I+8xJXR5gW2QwWYkT6iJva0bXitZKa/XMrSbdmg3r2Xnaidz9Qumd0VPaMrZlF9V9sA==
+promise-breaker@^5.0.0:
+ version "5.0.0"
+ resolved "https://registry.yarnpkg.com/promise-breaker/-/promise-breaker-5.0.0.tgz#58e8541f1619554057da95a211794d7834d30c1d"
+ integrity sha512-mgsWQuG4kJ1dtO6e/QlNDLFtMkMzzecsC69aI5hlLEjGHFNpHrvGhFi4LiK5jg2SMQj74/diH+wZliL9LpGsyA==
+
promise-inflight@^1.0.1:
version "1.0.1"
resolved "https://registry.yarnpkg.com/promise-inflight/-/promise-inflight-1.0.1.tgz#98472870bf228132fcbdd868129bad12c3c029e3"

feat(messaging): amqp-connection-manager for handling connection issues (version 2)