fix(messaging): client - return Observable<boolean> instead of EMPTY
MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
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';

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
-import { EMPTY, Subject, fromEvent } from 'rxjs';
+import { Subject, fromEvent, of } from 'rxjs';
import { map, share, filter } from 'rxjs/operators';
import { Channel, Connection, ConsumeMessage, Options } from 'amqplib';
import { TransportLayer, TransportLayerSendOpts, TransportMessage } from '../transport.interface';
@@ -63,15 +63,13 @@ export const createAmqpStrategy = async (options: RmqStrategyOptions): Promise<T
switch (opts.type) {
case 'publish':
channelInstance.assertExchange(queue, 'fanout', { durable: false });
- channelInstance.publish(queue, '', msg.data);
- return EMPTY;
+ 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:
- channelInstance.sendToQueue(queue, msg.data, { replyTo, correlationId });
- return EMPTY;
+ return of(channelInstance.sendToQueue(queue, msg.data, { replyTo, correlationId }));
}
};

fix(messaging): client - return Observable<boolean> instead of EMPTY