feat(messaging): createMicroservice + improved transport options type inference
MODIFY
packages/@integration/src/messaging/server.ts
packages/@integration/src/messaging/server.ts
-import { matchEvent, use, createContext } from '@marblejs/core';
+import { matchEvent, use } from '@marblejs/core';
import { eventValidator$, t } from '@marblejs/middleware-io';
-import { messagingListener, Transport, MsgEffect, MsgMiddlewareEffect } from '@marblejs/messaging';
+import { createMicroservice, messagingListener, Transport, MsgEffect, MsgMiddlewareEffect } from '@marblejs/messaging';
import { map, tap } from 'rxjs/operators';
const fibonacci = (n: number): number =>
@@ -21,23 +21,20 @@ const fibonacci$: MsgEffect = event$ =>
map(payload => ({ type: 'FIB_RESULT', payload })),
);
-const listener = messagingListener({
- effects: [
- fibonacci$,
- ],
- middlewares: [
- log$,
- ],
+export const microservice = createMicroservice({
transport: Transport.AMQP,
options: {
host: 'amqp://localhost:5672',
queue: 'test_queue',
queueOptions: { durable: false },
},
+ messagingListener: messagingListener({
+ effects: [fibonacci$],
+ middlewares: [log$],
+ }),
+ dependencies: [],
});
-export const server = listener();
-
if (process.env.NODE_ENV !== 'test') {
- server.run(createContext());
+ microservice.run();
}

MODIFY
packages/@integration/test/messaging.integration.spec.ts
packages/@integration/test/messaging.integration.spec.ts
import * as request from 'supertest';
import { server as httpServer } from '../src/messaging/client';
-import { server as messagingServer } from '../src/messaging/server';
-import { createContext } from '@marblejs/core';
+import { microservice as messagingServer } from '../src/messaging/server';
-const microservice = messagingServer.run(createContext());
+const microservice = messagingServer.run();
beforeAll(() => microservice);
@@ -17,4 +16,4 @@ describe('messaging integration', () => {
]));
});
-afterAll(() => microservice.then(s => s.close()));
+afterAll(() => microservice.then(con => con.close()));

MODIFY
packages/messaging/src/client/messaging.client.interface.ts
packages/messaging/src/client/messaging.client.interface.ts
import { Observable } from 'rxjs';
+import { NatsStrategy } from '../transport/strategies/nats.strategy.interface';
+import { TcpStrategy } from '../transport/strategies/tcp.strategy.interface';
+import { AmqpStrategy } from '../transport/strategies/amqp.strategy.interface';
+import { TransportMessageTransformer } from '../transport/transport.interface';
export interface MessagingClient {
emit: <T>(data: T) => Observable<any>;
@@ -6,3 +10,13 @@ export interface MessagingClient {
publish: <T>(data: T) => Observable<any>;
close: () => Observable<any>;
}
+
+type ConfigurationBase = {
+ msgTransformer?: TransportMessageTransformer<any>;
+}
+
+export type MessagingClientConfig =
+ | AmqpStrategy & ConfigurationBase
+ | NatsStrategy & ConfigurationBase
+ | TcpStrategy & ConfigurationBase
+ ;

MODIFY
packages/messaging/src/client/messaging.client.ts
packages/messaging/src/client/messaging.client.ts
@@ -4,21 +4,14 @@ import { mergeMap, take, map, mapTo, mergeMapTo } from 'rxjs/operators';
import {
Transport,
TransportMessage,
- TransportMessageTransformer,
TransportLayerConnection,
} from '../transport/transport.interface';
import { provideTransportLayer } from '../transport/transport.provider';
import { jsonTransformer } from '../transport/transport.transformer';
-import { MessagingClient } from './messaging.client.interface';
+import { MessagingClient, MessagingClientConfig } from './messaging.client.interface';
import { createUuid } from '@marblejs/core/dist/+internal/utils';
-export interface MessagingClientConfig {
- transport?: Transport;
- msgTransformer?: TransportMessageTransformer<any>;
- options?: any; // @TODO
-}
-
-export const messagingClient = (config: MessagingClientConfig = {}) => {
+export const messagingClient = (config: MessagingClientConfig) => {
const {
transport = Transport.TCP,
msgTransformer = jsonTransformer,
@@ -28,7 +21,7 @@ export const messagingClient = (config: MessagingClientConfig = {}) => {
const emit = (conn: Promise<TransportLayerConnection>) => <T>(msg: T) =>
from(conn).pipe(
mergeMap(c => c.sendMessage(
- options.queue,
+ c.channel,
{ data: msgTransformer.encode(msg as any) },
{ type: 'emit' },
)),
@@ -39,7 +32,7 @@ export const messagingClient = (config: MessagingClientConfig = {}) => {
const publish = (conn: Promise<TransportLayerConnection>) => <T>(msg: T) =>
from(conn).pipe(
mergeMap(c => c.sendMessage(
- options.queue,
+ c.channel,
{ data: msgTransformer.encode(msg as any) },
{ type: 'publish' },
)),
@@ -50,7 +43,7 @@ export const messagingClient = (config: MessagingClientConfig = {}) => {
const send = (conn: Promise<TransportLayerConnection>) => <T, U>(msg: T): Observable<U> =>
from(conn).pipe(
mergeMap(c => c.sendMessage(
- options.queue,
+ c.channel,
{ data: msgTransformer.encode(msg as any), correlationId: createUuid() },
{ type: 'send' }
)),

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
@@ -3,3 +3,6 @@ export * from './client/messaging.client';
export * from './client/messaging.client.interface';
export * from './transport/transport.interface';
export * from './effects/messaging-effects.interface';
+export * from './server/messaging.server';
+export * from './server/messaging.server.interface';
+export * from './server/messaging.server.tokens';

MODIFY
packages/messaging/src/listener/messaging.listener.ts
packages/messaging/src/listener/messaging.listener.ts
@@ -8,30 +8,26 @@ import {
import { Observable, Subscription } from 'rxjs';
import { map, publish, withLatestFrom } from 'rxjs/operators';
import {
- Transport,
TransportMessage,
TransportMessageTransformer,
TransportLayerConnection,
+ TransportLayer,
} from '../transport/transport.interface';
-import { provideTransportLayer } from '../transport/transport.provider';
import { jsonTransformer } from '../transport/transport.transformer';
import { MsgEffect, MsgMiddlewareEffect } from '../effects/messaging-effects.interface';
+import { TransportLayerToken } from '../server/messaging.server.tokens';
export interface MessagingListenerConfig {
effects?: MsgEffect<any, any>[];
middlewares?: MsgMiddlewareEffect<any, any>[];
- transport?: Transport;
msgTransformer?: TransportMessageTransformer<any>;
- options?: any; // @TODO
}
export const messagingListener = (config: MessagingListenerConfig = {}) => {
const {
effects = [],
middlewares = [],
- transport = Transport.TCP,
msgTransformer = jsonTransformer,
- options = {},
} = config;
const handleConnection = (conn: TransportLayerConnection, ask: ContextProvider) => {
@@ -74,15 +70,21 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
effectsSub = subscribeEffects(message$);
};
- return () => reader.map(ask => {
- const transportLayer = provideTransportLayer(transport, options);
-
- return transportLayer
- .then(server => server.connect())
- .then(conn => conn.consumeMessage().then(() => conn))
- .then(conn => {
- handleConnection(conn, ask);
+ const listen = (transportLayer: Promise<TransportLayer>, ask: ContextProvider) => async () => {
+ const layer = await transportLayer;
+ const conn = await layer.connect();
+ await conn.consumeMessage();
+ await handleConnection(conn, ask);
return conn;
- });
+ };
+
+ return reader.map(ask => {
+ const transportLayer = ask(TransportLayerToken)
+ .map(async layer => await layer)
+ .getOrElse(undefined as unknown as Promise<TransportLayer>);
+
+ return {
+ listen: listen(transportLayer, ask),
+ }
});
};

ADD
packages/messaging/src/server/messaging.server.interface.ts
+import { BoundDependency } from '@marblejs/core';
+import { messagingListener } from '../listener/messaging.listener';
+import { NatsStrategy } from '../transport/strategies/nats.strategy.interface';
+import { TcpStrategy } from '../transport/strategies/tcp.strategy.interface';
+import { AmqpStrategy } from '../transport/strategies/amqp.strategy.interface';
+
+type ConfigurationBase = {
+ messagingListener: ReturnType<typeof messagingListener>;
+ dependencies?: BoundDependency<any>[];
+}
+
+export type CreateMicroserviceConfig =
+ | AmqpStrategy & ConfigurationBase
+ | NatsStrategy & ConfigurationBase
+ | TcpStrategy & ConfigurationBase
+ ;

ADD
packages/messaging/src/server/messaging.server.tokens.ts
+import { createContextToken } from '@marblejs/core';
+import { TransportLayer } from '../transport/transport.interface';
+
+export const TransportLayerToken = createContextToken<Promise<TransportLayer>>();

ADD
packages/messaging/src/server/messaging.server.ts
+import { createContext, registerAll, bindTo } from '@marblejs/core';
+import { CreateMicroserviceConfig } from './messaging.server.interface';
+import { provideTransportLayer } from '../transport/transport.provider';
+import { Transport } from '../transport/transport.interface';
+import { TransportLayerToken } from './messaging.server.tokens';
+
+export const createMicroservice = (config: CreateMicroserviceConfig) => {
+ const {
+ options = {},
+ dependencies = [],
+ transport = Transport.AMQP,
+ messagingListener,
+ } = config;
+
+ const transportLayer = provideTransportLayer(transport, options);
+ const boundTransportLayer = bindTo(TransportLayerToken)(() => transportLayer);
+ const context = registerAll([ boundTransportLayer, ...dependencies ])(createContext());
+ const listenerWithContext = messagingListener.run(context);
+
+ return {
+ run: listenerWithContext.listen,
+ }
+};

ADD
packages/messaging/src/transport/strategies/amqp.strategy.interface.ts
+import { Options } from 'amqplib';
+import { Transport } from '../transport.interface';
+
+export interface AmqpStrategy {
+ transport: Transport.AMQP;
+ options: AmqpStrategyOptions;
+}
+
+export interface AmqpStrategyOptions {
+ host: string;
+ queue: string;
+ queueOptions?: Options.AssertQueue;
+ socketOptions?: any;
+ prefetchCount?: number;
+ isGlobalPrefetchCount?: boolean;
+}

MODIFY
packages/messaging/src/transport/strategies/amqp.strategy.ts
packages/messaging/src/transport/strategies/amqp.strategy.ts
import { Subject, fromEvent, of } from 'rxjs';
import { map, share, filter } from 'rxjs/operators';
-import { Channel, Connection, ConsumeMessage, Options } from 'amqplib';
+import { Channel, Connection, ConsumeMessage } from 'amqplib';
import { TransportLayer, TransportLayerSendOpts, TransportMessage } from '../transport.interface';
+import { AmqpStrategyOptions } from './amqp.strategy.interface';
-interface RmqStrategyOptions {
- host: string;
- queue: string;
- queueOptions?: Options.AssertQueue;
- socketOptions?: any;
- prefetchCount?: number;
- isGlobalPrefetchCount?: boolean;
-}
-
-export const createAmqpStrategy = async (options: RmqStrategyOptions): Promise<TransportLayer> => {
+export const createAmqpStrategy = async (options: AmqpStrategyOptions): Promise<TransportLayer> => {
const msgSubject$ = new Subject<ConsumeMessage>();
const resSubject$ = new Subject<ConsumeMessage>();
const responseQueue = options.queue + '__response';
@@ -88,6 +80,7 @@ export const createAmqpStrategy = async (options: RmqStrategyOptions): Promise<T
await channel.assertQueue(responseQueue);
return {
+ channel: options.queue,
sendMessage: sendMessage(channel),
consumeMessage: consumeMessage(channel),
consumeResponse: consumeResponse(channel),

ADD
packages/messaging/src/transport/strategies/nats.strategy.interface.ts
+import { Transport } from '../transport.interface';
+
+export interface NatsStrategy {
+ transport: Transport.NATS;
+ options: NatsStrategyOptions;
+}
+
+export interface NatsStrategyOptions {
+ // @TODO
+}

ADD
packages/messaging/src/transport/strategies/tcp.strategy.interface.ts
+import { Transport } from '../transport.interface';
+
+export interface TcpStrategy {
+ transport: Transport.TCP;
+ options: TcpStrategyOptions;
+}
+
+export interface TcpStrategyOptions {
+ // @TODO
+}

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
@@ -19,6 +19,7 @@ export interface TransportLayerConnection {
consumeMessage: () => Promise<any>;
consumeResponse: () => Promise<any>;
close: () => Promise<any>;
+ channel: string;
error$: Observable<Error>;
message$: Observable<TransportMessage<Buffer>>;
response$: Observable<TransportMessage<Buffer>>;

feat(messaging): createMicroservice + improved transport options type inference