feat(messaging): expose transport layer raw options
MODIFY
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
packages/messaging/src/server/specs/messaging.server.amqp.spec.ts
@@ -249,4 +249,15 @@ describe('messagingServer::AMQP', () => {
microservice = await Util.createAmqpMicroservice(options)({ effects: [test$], output$ });
});
+
+ test('microservice connection exposes raw configuration object', async () => {
+ // given
+ const options = Util.createAmqpOptions();
+
+ // when
+ microservice = await Util.createAmqpMicroservice(options)();
+
+ // then
+ expect(microservice.config.raw).toEqual(options);
+ });
});

MODIFY
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
packages/messaging/src/server/specs/messaging.server.redis.spec.ts
@@ -196,4 +196,15 @@ describe('messagingServer::Redis', () => {
await client.emit({ type: 'TEST' });
});
+
+ test('microservice connection exposes raw configuration object', async () => {
+ // given
+ const options = Util.createRedisOptions();
+
+ // when
+ microservice = await Util.createRedisMicroservice(options)();
+
+ // then
+ expect(microservice.config.raw).toEqual(options);
+ });
});

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
@@ -6,27 +6,34 @@ import { createUuid } from '@marblejs/core/dist/+internal/utils';
import { TransportLayer, TransportMessage, TransportLayerConnection, Transport, DEFAULT_TIMEOUT } from '../transport.interface';
import { AmqpStrategyOptions, AmqpConnectionStatus, AmqpCannotSetExpectAckForNonConsumerConnection } from './amqp.strategy.interface';
-class AmqpStrategyConnection implements TransportLayerConnection {
+class AmqpStrategyConnection implements TransportLayerConnection<Transport.AMQP> {
private msgSubject$ = new Subject<ConsumeMessage>();
private statusSubject$ = new Subject<AmqpConnectionStatus>();
private errorSubject$ = new Subject<Error>();
private closeSubject$ = new Subject();
constructor(
- public type: Transport,
- private isConsumer: boolean,
+ isConsumer: boolean,
+ private opts: AmqpStrategyOptions,
private connectionManager: AmqpConnectionManager,
private channelWrapper: ChannelWrapper,
- private options: AmqpStrategyOptions,
) {
process.nextTick(async () => {
- if (this.isConsumer) await this.consumeMessages();
+ if (isConsumer) await this.consumeMessages();
this.statusSubject$.next(AmqpConnectionStatus.CONNECTED);
});
}
+ get type() {
+ return Transport.AMQP as const;
+ }
+
get config() {
- return { timeout: this.options.timeout ?? DEFAULT_TIMEOUT };
+ return {
+ timeout: this.opts.timeout ?? DEFAULT_TIMEOUT,
+ channel: this.opts.queue,
+ raw: this.opts,
+ };
}
get close$() {
@@ -80,16 +87,16 @@ class AmqpStrategyConnection implements TransportLayerConnection {
private consumeMessages = async () =>
this.channelWrapper.addSetup(async (channel: Channel) => {
await channel.consume(
- this.options.queue,
+ this.opts.queue,
msg => msg && this.msgSubject$.next(msg),
- { noAck: !this.options.expectAck },
+ { noAck: !this.opts.expectAck },
);
});
sendMessage = async (queue: string, msg: TransportMessage<Buffer>) => {
const { data } = msg;
const correlationId = createUuid();
- const timeout = this.options.timeout ?? DEFAULT_TIMEOUT;
+ const timeout = this.opts.timeout ?? DEFAULT_TIMEOUT;
const replyToSubject = new Subject<string>();
const resSubject$ = new Subject<{ msg: ConsumeMessage; tag: string }>();
@@ -142,7 +149,7 @@ class AmqpStrategyConnection implements TransportLayerConnection {
const { correlationId, data, replyTo } = msg;
// don't try to emit the message to the same channel if "ack" mode is enabled
- if (this.options.expectAck && queue === this.getChannel())
+ if (this.opts.expectAck && queue === this.getChannel())
return false;
await this.channelWrapper.sendToQueue(queue, data, {
@@ -154,20 +161,20 @@ class AmqpStrategyConnection implements TransportLayerConnection {
};
ackMessage = (msg: TransportMessage | undefined) => {
- if (this.options.expectAck && msg && !msg.raw.isAcked) {
+ if (this.opts.expectAck && msg && !msg.raw.isAcked) {
this.channelWrapper.ack(msg.raw);
msg.raw.isAcked = true;
}
}
nackMessage = (msg: TransportMessage | undefined, resend = true) => {
- if (this.options.expectAck && msg && !msg.raw.isNacked) {
+ if (this.opts.expectAck && msg && !msg.raw.isNacked) {
this.channelWrapper.nack(msg.raw, false , resend);
msg.raw.isNacked = true;
}
}
- getChannel = () => this.options.queue;
+ getChannel = () => this.opts.queue;
close = async () => {
await this.channelWrapper.close();
@@ -176,11 +183,11 @@ class AmqpStrategyConnection implements TransportLayerConnection {
}
}
-class AmqpStrategy implements TransportLayer {
+class AmqpStrategy implements TransportLayer<Transport.AMQP> {
constructor(private options: AmqpStrategyOptions) {}
get type() {
- return Transport.AMQP;
+ return Transport.AMQP as const;
}
get config() {
@@ -214,14 +221,13 @@ class AmqpStrategy implements TransportLayer {
await (channelWrapper as any).waitForConnect(); // not available in @types/amqp-connection-manager
return new AmqpStrategyConnection(
- Transport.AMQP,
isConsumer,
+ this.options,
connectionManager,
channelWrapper,
- this.options,
);
}
}
-export const createAmqpStrategy = (options: AmqpStrategyOptions): TransportLayer =>
+export const createAmqpStrategy = (options: AmqpStrategyOptions): TransportLayer<Transport.AMQP> =>
new AmqpStrategy(options);

MODIFY
packages/messaging/src/transport/strategies/local.strategy.provider.ts
packages/messaging/src/transport/strategies/local.strategy.provider.ts
import * as O from 'fp-ts/lib/Option';
import { pipe } from 'fp-ts/lib/pipeable';
-import { TransportLayer, TransportLayerConnection } from '../transport.interface';
+import { TransportLayer, TransportLayerConnection, Transport } from '../transport.interface';
import { createLocalStrategy } from './local.strategy';
-class LocalStrategyProvider implements TransportLayer {
+export class LocalStrategyProvider implements TransportLayer<Transport.LOCAL> {
private static instance: LocalStrategyProvider;
- private readonly strategy: TransportLayer;
- private readonly strategyConnection: Promise<TransportLayerConnection>;
+ private readonly strategy: TransportLayer<Transport.LOCAL>;
+ private readonly strategyConnection: Promise<TransportLayerConnection<Transport.LOCAL>>;
private constructor() {
this.strategy = createLocalStrategy({});

MODIFY
packages/messaging/src/transport/strategies/local.strategy.ts
packages/messaging/src/transport/strategies/local.strategy.ts
@@ -5,19 +5,24 @@ import { TransportLayer, TransportLayerConnection, TransportMessage, Transport,
import { throwUnsupportedError } from '../transport.error';
import { LocalStrategyOptions, EVENT_BUS_CHANNEL } from './local.strategy.interface';
-class LocalStrategyConnection implements TransportLayerConnection {
+class LocalStrategyConnection implements TransportLayerConnection<Transport.LOCAL> {
private errorSubject$ = new Subject<Error>();
private rpcSubject$ = new Subject<TransportMessage<Buffer>>();
private msgSubject$ = new Subject<TransportMessage<Buffer>>();
private closeSubject$ = new Subject();
- constructor(
- public type: Transport,
- private opts: TransportLayerConfig,
- ) { }
+ constructor(private opts: TransportLayerConfig) { }
+
+ get type() {
+ return Transport.LOCAL as const;
+ }
get config() {
- return { timeout: this.opts.timeout };
+ return {
+ timeout: this.opts.timeout,
+ channel: this.opts.channel,
+ raw: this.opts,
+ };
}
get status$() {
@@ -79,11 +84,11 @@ class LocalStrategyConnection implements TransportLayerConnection {
getChannel = () => this.opts.channel;
}
-class LocalStrategy implements TransportLayer {
+class LocalStrategy implements TransportLayer<Transport.LOCAL> {
constructor(private options: LocalStrategyOptions) {}
get type() {
- return Transport.LOCAL;
+ return Transport.LOCAL as Transport.LOCAL;
}
get config() {
@@ -95,12 +100,9 @@ class LocalStrategy implements TransportLayer {
}
async connect() {
- return new LocalStrategyConnection(
- Transport.LOCAL,
- this.config,
- );
+ return new LocalStrategyConnection(this.config);
}
}
-export const createLocalStrategy = (options: LocalStrategyOptions): TransportLayer =>
+export const createLocalStrategy = (options: LocalStrategyOptions): TransportLayer<Transport.LOCAL> =>
new LocalStrategy(options);

MODIFY
packages/messaging/src/transport/strategies/redis.strategy.ts
packages/messaging/src/transport/strategies/redis.strategy.ts
@@ -12,20 +12,20 @@ type RedisIncomingMsg = {
channel: string;
};
-class RedisStrategyConnection implements TransportLayerConnection {
+class RedisStrategyConnection implements TransportLayerConnection<Transport.REDIS> {
private statusSubject$ = new Subject<RedisConnectionStatus>();
private producerSubject = new Subject<RedisIncomingMsg>();
private consumerSubject = new Subject<RedisIncomingMsg>();
private closeSubject$ = new Subject();
constructor(
- public type: Transport,
- private opts: { channel: string; isConsumer: boolean; timeout: number },
+ isConsumer: boolean,
+ private opts: RedisStrategyOptions,
private publisher: RedisClient,
private subscriber: RedisClient,
private rpcSubscriber: RedisClient,
) {
- if (opts.isConsumer) {
+ if (isConsumer) {
subscriber.on('message', (channel, content) => this.consumerSubject.next({ content, channel }));
} else {
rpcSubscriber.on('message', (channel, content) => this.producerSubject.next({ content, channel }));
@@ -34,8 +34,16 @@ class RedisStrategyConnection implements TransportLayerConnection {
process.nextTick(() => this.statusSubject$.next(RedisConnectionStatus.CONNECT));
}
+ get type() {
+ return Transport.REDIS as const;
+ }
+
get config() {
- return { timeout: this.opts.timeout };
+ return {
+ timeout: this.opts.timeout ?? DEFAULT_TIMEOUT,
+ channel: this.opts.channel,
+ raw: this.opts,
+ };
}
get status$() {
@@ -118,11 +126,11 @@ class RedisStrategyConnection implements TransportLayerConnection {
getChannel = () => this.opts.channel;
}
-class RedisStrategy implements TransportLayer {
+class RedisStrategy implements TransportLayer<Transport.REDIS> {
constructor(private options: RedisStrategyOptions) {}
get type() {
- return Transport.REDIS;
+ return Transport.REDIS as const;
}
get config() {
@@ -139,21 +147,20 @@ class RedisStrategy implements TransportLayer {
}
async connect(data?: { isConsumer: boolean }) {
- const { channel, timeout } = this.options;
+ const { channel } = this.options;
const isConsumer = !!data?.isConsumer;
const publisher = await RedisHelper.connectClient(this.clientOpts);
const subscriber = await RedisHelper.connectClient(this.clientOpts);
const rpcSubscriber = await RedisHelper.connectClient(this.clientOpts);
- const opts = { isConsumer, channel, timeout: timeout ?? DEFAULT_TIMEOUT };
if (isConsumer) {
await RedisHelper.subscribeChannel(subscriber)(channel);
}
return new RedisStrategyConnection(
- Transport.REDIS,
- opts,
+ isConsumer,
+ this.options,
publisher,
subscriber,
rpcSubscriber,
@@ -161,5 +168,5 @@ class RedisStrategy implements TransportLayer {
}
}
-export const createRedisStrategy = (options: RedisStrategyOptions): TransportLayer =>
+export const createRedisStrategy = (options: RedisStrategyOptions): TransportLayer<Transport.REDIS> =>
new RedisStrategy(options);

MODIFY
packages/messaging/src/transport/strategies/tcp.strategy.ts
packages/messaging/src/transport/strategies/tcp.strategy.ts
-import { TransportLayer } from '../transport.interface';
+import { TransportLayer, Transport } from '../transport.interface';
/* istanbul ignore next */
-export const createTcpStrategy = (): TransportLayer => {
+export const createTcpStrategy = (): TransportLayer<Transport.TCP> => {
// @TODO
return {} as any;
};

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -24,9 +24,9 @@ export type TransportStrategy =
| LocalStrategy
;
-export interface TransportLayer {
- connect: (opts?: { isConsumer: boolean }) => Promise<TransportLayerConnection>;
- type: Transport;
+export interface TransportLayer<T extends Transport = Transport> {
+ connect: (opts?: { isConsumer: boolean }) => Promise<TransportLayerConnection<T>>;
+ type: T;
config: TransportLayerConfig;
}
@@ -36,15 +36,15 @@ export interface TransportLayerConfig {
timeout: number;
}
-export interface TransportLayerConnection {
+export interface TransportLayerConnection<T extends Transport = Transport> {
sendMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<TransportMessage<Buffer>>;
emitMessage: (channel: string, message: TransportMessage<Buffer>) => Promise<boolean>;
ackMessage: (message: TransportMessage | undefined) => void;
nackMessage: (message: TransportMessage | undefined, resend?: boolean) => void;
close: () => Promise<any>;
getChannel(): string;
- type: Transport;
- config: { timeout: number };
+ type: T;
+ config: { timeout: number, channel: string; raw: any; };
message$: Observable<TransportMessage<Buffer>>;
status$: Observable<string>;
error$: Observable<Error>;

MODIFY
packages/messaging/src/transport/transport.provider.ts
packages/messaging/src/transport/transport.provider.ts
-import { Transport, TransportLayer } from './transport.interface';
+import { Transport } from './transport.interface';
import { createAmqpStrategy } from './strategies/amqp.strategy';
import { createRedisStrategy } from './strategies/redis.strategy';
import { provideLocalStrategy } from './strategies/local.strategy.provider';
-export const provideTransportLayer = (transport: Transport, transportOptions: any = {}): TransportLayer => {
+export const provideTransportLayer = <T extends Transport>(transport: T, transportOptions: any = {}) => {
switch (transport) {
case Transport.AMQP:
return createAmqpStrategy(transportOptions);

feat(messaging): expose transport layer raw options