feat(messaging): continuous message stream
MODIFY
packages/core/src/+internal/utils/string.util.ts
packages/core/src/+internal/utils/string.util.ts
+import * as uuid from 'uuid';
import { fromNullable } from 'fp-ts/lib/Option';
export const isString = (value: any): value is string =>
@@ -18,3 +19,5 @@ export const stringify = (value: any): string =>
typeof value === 'function'
? (value.displayName || value.name)
: JSON.stringify(value);
+
+export const createUuid = () => uuid.v4();

DELETE
packages/messaging/src/effects/.gitkeep
packages/messaging/src/effects/.gitkeep
ADD
packages/messaging/src/effects/messaging-effects.interface.ts
+import { Effect, Event } from '@marblejs/core';
+
+type MsgClient = any; // @TODO
+
+export interface MsgMiddlewareEffect<
+ I = Event,
+ O = Event,
+> extends MsgEffect<I, O> {}
+
+export interface MsgErrorEffect<
+ Err extends Error = Error,
+ I = Event,
+ O = Event
+> extends MsgEffect<I, O, MsgClient, Err> {}
+
+export interface MsgEffect<
+ I = Event,
+ O = Event,
+ Client = MsgClient,
+ Err extends Error = Error,
+> extends Effect<I, O, Client, Err> {}

MODIFY
packages/messaging/src/index.ts
packages/messaging/src/index.ts
export { messagingListener } from './listener/messaging.listener';
export * from './transport/transport.interface';
+export * from './effects/messaging-effects.interface';

MODIFY
packages/messaging/src/listener/messaging.listener.ts
packages/messaging/src/listener/messaging.listener.ts
-import { reader, ContextReader } from '@marblejs/core';
-import { map } from 'rxjs/operators';
-import { Transport, TransportMessageTransformer } from '../transport/transport.interface';
-import { provideTransportServer } from '../transport/transport.provider';
+import {
+ reader,
+ ContextReader,
+ ContextProvider,
+ combineMiddlewares,
+ combineEffects,
+ createEffectMetadata,
+} from '@marblejs/core';
+import { createUuid } from '@marblejs/core/dist/+internal/utils';
+import { from, Observable, Subscription } from 'rxjs';
+import { map, mergeMap, publish, withLatestFrom } 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 { MsgEffect, MsgMiddlewareEffect } from '../effects/messaging-effects.interface';
export interface MessagingListenerConfig {
- effects?: any[]; // @TODO
- middlewares?: any[]; // @TODO
+ effects?: MsgEffect<any, any>[];
+ middlewares?: MsgMiddlewareEffect<any, any>[];
transport?: Transport;
- messageTransformer?: TransportMessageTransformer<any>;
+ msgTransformer?: TransportMessageTransformer<any>;
options?: any; // @TODO
}
@@ -17,23 +32,109 @@ export const messagingListener = (config: MessagingListenerConfig = {}) => {
effects = [],
middlewares = [],
transport = Transport.TCP,
- messageTransformer = jsonTransformer,
+ msgTransformer = jsonTransformer,
options = {},
} = config;
- const transportServer = provideTransportServer(transport, options);
+ const handleConnection = (conn: TransportLayerConnection, ask: ContextProvider) => {
+ let effectsSub: Subscription;
- return (): ContextReader => reader.map(() => {
- // @TODO
- transportServer.then(server => {
- server.connect().then(con => {
- con.handleMessage().pipe(
- map(messageTransformer.decode),
- ).subscribe(console.log);
+ const combinedEffects = combineEffects(...effects);
+ const combinedMiddlewares = combineMiddlewares(...middlewares);
+ const defaultMetadata = createEffectMetadata({ ask });
- const encodedMessage = messageTransformer.encode({ type: 'TEST', payload: 'test_message' });
- con.sendMessage(encodedMessage);
+ const message$ = conn.handleMessage().pipe(
+ map(msg => ({ ...msg, data: msgTransformer.decode(msg.data) } as TransportMessage<any>)),
+ publish(msg$ => combinedMiddlewares(msg$.pipe(map(m => m.data)), conn, defaultMetadata).pipe(
+ withLatestFrom(msg$),
+ )),
+ map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
+ publish(msg$ => combinedEffects(msg$.pipe(map(m => m.data)), conn, defaultMetadata).pipe(
+ withLatestFrom(msg$),
+ )),
+ map(([data, msg]) => ({ ...msg, data } as TransportMessage<any>)),
+ );
+
+ const subscribeEffects = (input$: Observable<TransportMessage<any>>) =>
+ input$.subscribe(
+ msg => {
+ if (msg.replyTo) {
+ console.log(msg);
+ conn.sendMessage(msg.replyTo, {
+ data: msgTransformer.encode(msg.data),
+ correlationId: msg.correlationId,
+ raw: msg.raw,
+ });
+ }
+ },
+ error => {
+ // @TODO: handle errors
+ console.error(error);
+ if (effectsSub.closed) { effectsSub = subscribeEffects(message$); }
+ },
+ );
+
+ effectsSub = subscribeEffects(message$);
+ };
+
+ return (): ContextReader => reader.map(ask => {
+ const transportLayer = provideTransportLayer(transport, options);
+
+ from(transportLayer)
+ .pipe(mergeMap(server => server.connect()))
+ .subscribe(conn => handleConnection(conn, ask));
+
+ // @TODO: simulate msgs posting
+ transportLayer
+ .then(server => server.connect())
+ .then(conn => {
+ const msg1 = msgTransformer.encode({ type: 'ADD', payload: 1 });
+ const msg2 = msgTransformer.encode({ type: 'ADD', payload: 2 });
+ const msg3 = msgTransformer.encode({ type: 'SUM' });
+
+ const msg4 = msgTransformer.encode({ type: 'ADD', payload: 3 });
+ const msg5 = msgTransformer.encode({ type: 'ADD', payload: 6 });
+ const msg6 = msgTransformer.encode({ type: 'SUM' });
+
+ setTimeout(() => {
+ conn.sendMessage(options.queue, {
+ data: msg1,
+ });
+ }, 1);
+
+ setTimeout(() => {
+ conn.sendMessage(options.queue, {
+ data: msg2,
+ });
+ }, 2);
+
+ setTimeout(() => {
+ conn.sendMessage(options.queue, {
+ data: msg3,
+ replyTo: options.queue + '_1',
+ correlationId: createUuid(),
+ });
+ }, 3);
+
+ setTimeout(() => {
+ conn.sendMessage(options.queue, {
+ data: msg4,
+ });
+ }, 10);
+
+ setTimeout(() => {
+ conn.sendMessage(options.queue, {
+ data: msg5,
+ });
+ }, 11);
+
+ setTimeout(() => {
+ conn.sendMessage(options.queue, {
+ data: msg6,
+ replyTo: options.queue + '_2',
+ correlationId: createUuid(),
});
+ }, 12);
});
});
};

ADD
packages/messaging/src/transport/strategies/amqp.strategy.ts
+import { EMPTY, Subject, from, fromEvent } from 'rxjs';
+import { map, share } from 'rxjs/operators';
+import { Channel, Connection, ConsumeMessage, Options } from 'amqplib';
+import { TransportLayer, TransportMessage } from '../transport.interface';
+
+interface RmqStrategyOptions {
+ host: string;
+ queue: string;
+ queueOptions?: Options.AssertQueue;
+ socketOptions?: any;
+ prefetchCount?: number;
+ isGlobalPrefetchCount?: boolean;
+}
+
+export const createAmqpStrategy = async (options: RmqStrategyOptions): Promise<TransportLayer> => {
+
+ const sendMessage = (channelInstance: Channel) => (queue: string, msg: TransportMessage<Buffer>) => {
+ const { correlationId, replyTo } = msg;
+
+ channelInstance.sendToQueue(queue, msg.data, { correlationId, replyTo });
+
+ return EMPTY;
+ };
+
+ const handleMessage = (channelInstance: Channel) => () => {
+ const msg$ = new Subject<ConsumeMessage>();
+
+ channelInstance.consume(
+ options.queue,
+ msg => msg && msg$.next(msg),
+ { noAck: true },
+ );
+
+ return msg$.asObservable().pipe(
+ share(),
+ map(message => ({
+ data: message.content,
+ replyTo: message.properties.replyTo,
+ correlationId: message.properties.correlationId,
+ raw: message,
+ } as TransportMessage<Buffer>))
+ );
+ };
+
+ const close = (connection: Connection) => () =>
+ from(connection.close());
+
+ const error$ = (connection: Connection) =>
+ fromEvent<Error>(connection, 'error');
+
+ const connect = async () => {
+ const amqplib = await import('amqplib');
+ const connection = await amqplib.connect(options.host);
+ const channel = await connection.createChannel();
+
+ channel.assertQueue(options.queue, options.queueOptions);
+
+ return {
+ sendMessage: sendMessage(channel),
+ handleMessage: handleMessage(channel),
+ close: close(connection),
+ error$: error$(connection),
+ };
+ };
+
+ return { connect };
+};

MODIFY
packages/messaging/src/transport/strategies/nats.strategy.ts
packages/messaging/src/transport/strategies/nats.strategy.ts
-import { TransportServer } from '../transport.interface';
+import { TransportLayer } from '../transport.interface';
-export const createNatsStrategy = (): TransportServer => {
+export const createNatsStrategy = (): TransportLayer => {
// @TODO
return {} as any;
};

DELETE
packages/messaging/src/transport/strategies/rmq.strategy.ts
-import { EMPTY, Subject } from 'rxjs';
-import { map } from 'rxjs/operators';
-import { Channel, Connection, ConsumeMessage } from 'amqplib';
-import { TransportServer } from '../transport.interface';
-
-interface RmqStrategyOptions {
- host: string;
- queue: string;
- queueOptions?: any;
- socketOptions?: any;
- prefetchCount?: number;
- isGlobalPrefetchCount?: boolean;
-}
-
-export const createRmqStrategy = async (options: RmqStrategyOptions): Promise<TransportServer> => {
- const amqplib = await import('amqplib');
- const message$ = new Subject<ConsumeMessage>();
-
- const sendMessage = (channelInstance: Channel) => (message: Buffer) => {
- channelInstance.sendToQueue(options.queue, message);
- return EMPTY;
- };
-
- const handleMessage = (channelInstance: Channel) => () => {
- channelInstance.consume(
- options.queue,
- message => message && message$.next(message),
- { noAck: true },
- );
-
- return message$.pipe(map(message => message.content));
- };
-
- const close = (connection: Connection) => () => {
- return connection.close() as any as Promise<void>;
- };
-
- const connect = async () => {
- const connection = await amqplib.connect(options.host);
- const channel = await connection.createChannel();
- channel.assertQueue(options.queue, options.queueOptions);
-
- return {
- sendMessage: sendMessage(channel),
- handleMessage: handleMessage(channel),
- close: close(connection),
- };
- };
-
- return { connect };
-};

MODIFY
packages/messaging/src/transport/strategies/tcp.strategy.ts
packages/messaging/src/transport/strategies/tcp.strategy.ts
-import { TransportServer } from '../transport.interface';
+import { TransportLayer } from '../transport.interface';
-export const createTcpStrategy = (): TransportServer => {
+export const createTcpStrategy = (): TransportLayer => {
// @TODO
return {} as any;
};

MODIFY
packages/messaging/src/transport/transport.interface.ts
packages/messaging/src/transport/transport.interface.ts
import { Observable } from 'rxjs';
+import { Event } from '@marblejs/core';
export enum Transport {
TCP,
NATS,
- RMQ,
+ AMQP,
REDIS,
MQTT,
GRPC,
}
-export interface TransportServer {
- connect: () => Promise<TransportServerConnection>;
+export interface TransportLayer {
+ connect: () => Promise<TransportLayerConnection>;
}
-export interface TransportServerConnection {
- sendMessage: (msg: Buffer) => Observable<any>;
- handleMessage: () => Observable<any>;
- close: () => Promise<any>;
+export interface TransportLayerConnection {
+ sendMessage: (channel: string, message: TransportMessage<Buffer>) => Observable<any>;
+ handleMessage: () => Observable<TransportMessage<Buffer>>;
+ close: () => Observable<any>;
+ error$: Observable<Error>;
}
export interface TransportMessageTransformer<T> {
decode: (incomingEvent: Buffer) => T;
encode: (outgoingEvent: T) => Buffer;
}
+
+export interface TransportMessage<T = Event> {
+ data: T;
+ raw?: any;
+ replyTo?: string;
+ correlationId?: string;
+}

MODIFY
packages/messaging/src/transport/transport.provider.ts
packages/messaging/src/transport/transport.provider.ts
-import { Transport, TransportServer } from './transport.interface';
-import { createRmqStrategy } from './strategies/rmq.strategy';
+import { Transport, TransportLayer } from './transport.interface';
+import { createAmqpStrategy } from './strategies/amqp.strategy';
// @TODO
import { createTcpStrategy } from './strategies/tcp.strategy';
import { createNatsStrategy } from './strategies/nats.strategy';
-export const provideTransportServer = (transport: Transport, transportOptions: any): Promise<TransportServer> => {
+export const provideTransportLayer = (transport: Transport, transportOptions: any): Promise<TransportLayer> => {
switch (transport) {
- case Transport.RMQ:
- return createRmqStrategy(transportOptions);
+ case Transport.AMQP:
+ return createAmqpStrategy(transportOptions);
default:
// @TODO
- return createRmqStrategy(transportOptions);
+ return createAmqpStrategy(transportOptions);
}
};

feat(messaging): continuous message stream