fix(messaging): don't crash internal messages stream when unhandled error occurs
MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
@@ -91,6 +91,10 @@ export const microservice = createMicroservice({
),
});
+const bootstrap = async () => {
+ await microservice();
+};
+
if (process.env.NODE_ENV !== 'test') {
- microservice();
+ bootstrap();
}

MODIFY
packages/messaging/src/server/messaging.server.listener.ts
packages/messaging/src/server/messaging.server.listener.ts
@@ -98,11 +98,16 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
if (effectsSub.closed) { effectsSub = subscribeEffects(message$); }
}
+ const onSubscribeEffectsClose = () => {
+ effectsSub = subscribeEffects(message$);
+ }
+
const subscribeEffects = (input$: Observable<TransportMessage<any>>) => input$
.pipe(takeUntil(conn.close$))
.subscribe(
onSubscribeEffectsOutput(conn),
onSubscribeEffectsError(errorSubject),
+ onSubscribeEffectsClose,
);
conn.close$

MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
import { Event, matchEvent, use, EventError, createEvent } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
-import { map, tap, mergeMapTo, delay } from 'rxjs/operators';
+import { map, tap, mergeMapTo, delay, bufferCount } from 'rxjs/operators';
import { Transport } from '../../transport/transport.interface';
import { MsgEffect, MsgErrorEffect } from '../../effects/messaging.effects.interface';
import { Subject, throwError } from 'rxjs';
@@ -97,14 +97,18 @@ describe('messagingServer::AMQP', () => {
await client.emitMessage(options.queue, message);
});
- test('reacts to thrown Error', async done => {
+ test('reacts to thrown Error and doesn\'t crash internal messages stream', async done => {
const event = { type: 'EVENT_TEST' };
const error = new Error('test_error');
const errorSubject = new Subject<[Event | undefined, Error]>();
- errorSubject.subscribe(data => {
- expect(data[0]).toBeUndefined();
- expect(data[1]).toEqual(error);
+ errorSubject
+ .pipe(bufferCount(2))
+ .subscribe(data => {
+ expect(data[0][0]).toBeUndefined();
+ expect(data[0][1]).toEqual(error);
+ expect(data[1][0]).toBeUndefined();
+ expect(data[1][1]).toEqual(error);
setTimeout(() => server.close().then(done), 1000);
});
@@ -124,9 +128,8 @@ describe('messagingServer::AMQP', () => {
const client = await runClient(Transport.AMQP, options);
const server = await runServer(Transport.AMQP, options)(event$, error$);
- const emitResult = await client.emitMessage(options.queue, createMessage(event));
-
- expect(emitResult).toEqual(true);
+ await client.emitMessage(options.queue, createMessage(event));
+ await client.emitMessage(options.queue, createMessage(event));
});
test('reacts to thrown EventError', async done => {

fix(messaging): don't crash internal messages stream when unhandled error occurs