feat(messaging): transport - rmq.strategy (WIP)
MODIFY
packages/@integration/package.json
packages/@integration/package.json
},
"homepage": "https://github.com/marblejs/marble#readme",
"dependencies": {
+<<<<<<< HEAD
"@marblejs/core": "^2.3.1",
"@marblejs/middleware-body": "^2.3.1",
"@marblejs/middleware-cors": "^2.3.1",
"@marblejs/middleware-logger": "^2.3.1",
"@marblejs/middleware-multipart": "^2.3.1",
"@marblejs/websockets": "^2.3.1",
+=======
+ "@marblejs/core": "^2.1.0",
+ "@marblejs/middleware-body": "^2.1.0",
+ "@marblejs/middleware-cors": "^2.1.0",
+ "@marblejs/middleware-io": "^2.1.0",
+ "@marblejs/middleware-logger": "^2.1.0",
+ "@marblejs/websockets": "^2.1.0",
+ "@marblejs/messaging": "^2.1.0",
+>>>>>>> feat(messaging): transport - rmq.strategy (WIP)
"@types/node": "^10.1.2",
"rxjs": "^6.3.3",
"typescript": "~3.0.3"

MODIFY
packages/messaging/package.json
packages/messaging/package.json
"dependencies": {
},
"devDependencies": {
- "@marblejs/core": "^2.1.0"
+ "@marblejs/core": "^2.1.0",
+ "amqplib": "^0.5.3",
+ "@types/amqplib": "^0.5.11"
},
"publishConfig": {
"access": "private"

RENAME
packages/messaging/src/index.tsx
packages/messaging/src/index.ts
export { messagingListener } from './listener/messaging.listener';
+export * from './transport/transport.interface';

ADD
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 { jsonTransformer } from '../transport/transport.transformer';
+
+export interface MessagingListenerConfig {
+ effects?: any[]; // @TODO
+ middlewares?: any[]; // @TODO
+ transport?: Transport;
+ messageTransformer?: TransportMessageTransformer<any>;
+ options?: any; // @TODO
+}
+
+export const messagingListener = (config: MessagingListenerConfig = {}) => {
+ const {
+ effects = [],
+ middlewares = [],
+ transport = Transport.TCP,
+ messageTransformer = jsonTransformer,
+ options = {},
+ } = config;
+
+ const transportServer = provideTransportServer(transport, options);
+
+ return (): ContextReader => reader.map(() => {
+ // @TODO
+ transportServer.then(server => {
+ server.connect().then(con => {
+ con.handleMessage().pipe(
+ map(messageTransformer.decode),
+ ).subscribe(console.log);
+
+ const encodedMessage = messageTransformer.encode({ type: 'TEST', payload: 'test_message' });
+ con.sendMessage(encodedMessage);
+ });
+ });
+ });
+};

DELETE
packages/messaging/src/listener/messaging.listener.tsx
-import { reader, ContextReader } from '@marblejs/core';
-import { Transport } from '../server/server.interface';
-
-export interface MessagingListenerConfig {
- effects?: any[];
- middlewares?: any[];
- transport?: Transport;
-}
-
-export const messagingListener = (config: MessagingListenerConfig = {}) => {
- const { effects = [], middlewares = [], transport = Transport.TCP } = config;
-
- return (): ContextReader => reader.map(ask => {
- return {}; // @TODO
- });
-};

DELETE
packages/messaging/src/server/server.interface.tsx
-import { Observable } from 'rxjs';
-
-export enum Transport {
- TCP,
- NATS,
- RABBITMQ,
- REDIS,
- MQTT,
- GRPC,
-}
-
-export interface TransportServer {
- listen: () => any;
- close: () => any;
- sendMessage: <T>(msg: T, channel: any) => Observable<any>;
- handleMessage: <T>(msg: T) => Observable<any>;
-}

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

ADD
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 };
+};

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

ADD
packages/messaging/src/transport/transport.interface.ts
+import { Observable } from 'rxjs';
+
+export enum Transport {
+ TCP,
+ NATS,
+ RMQ,
+ REDIS,
+ MQTT,
+ GRPC,
+}
+
+export interface TransportServer {
+ connect: () => Promise<TransportServerConnection>;
+}
+
+export interface TransportServerConnection {
+ sendMessage: (msg: Buffer) => Observable<any>;
+ handleMessage: () => Observable<any>;
+ close: () => Promise<any>;
+}
+
+export interface TransportMessageTransformer<T> {
+ decode: (incomingEvent: Buffer) => T;
+ encode: (outgoingEvent: T) => Buffer;
+}

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

ADD
packages/messaging/src/transport/transport.transformer.ts
+import { Event } from '@marblejs/core';
+import { compose } from 'fp-ts/lib/function';
+import { TransportMessageTransformer } from './transport.interface';
+
+export const jsonTransformer: TransportMessageTransformer<Event> = {
+ decode: event => JSON.parse(event.toString()),
+ encode: event => compose(Buffer.from, JSON.stringify)(event),
+};

MODIFY
yarn.lock
yarn.lock
@@ -1216,6 +1216,15 @@ binary-extensions@^1.0.0:
version "1.12.0"
resolved "https://registry.yarnpkg.com/binary-extensions/-/binary-extensions-1.12.0.tgz#c2d780f53d45bba8317a8902d4ceeaf3a6385b14"
+bitsyntax@~0.1.0:
+ version "0.1.0"
+ resolved "https://registry.yarnpkg.com/bitsyntax/-/bitsyntax-0.1.0.tgz#b0c59acef03505de5a2ed62a2f763c56ae1d6205"
+ integrity sha512-ikAdCnrloKmFOugAfxWws89/fPc+nw0OOG1IzIE72uSOg/A3cYptKCjSUhDTuj7fhsJtzkzlv7l3b8PzRHLN0Q==
+ dependencies:
+ buffer-more-ints "~1.0.0"
+ debug "~2.6.9"
+ safe-buffer "~5.1.2"
+
block-stream@*:
version "0.0.9"
resolved "https://registry.yarnpkg.com/block-stream/-/block-stream-0.0.9.tgz#13ebfe778a03205cfe03751481ebb4b3300c126a"
@@ -1226,6 +1235,11 @@ bluebird@^3.5.0, bluebird@^3.5.1, bluebird@^3.5.3:
version "3.5.3"
resolved "https://registry.yarnpkg.com/bluebird/-/bluebird-3.5.3.tgz#7d01c6f9616c9a51ab0f8c549a79dfe6ec33efa7"
+bluebird@^3.5.2:
+ version "3.5.4"
+ resolved "https://registry.yarnpkg.com/bluebird/-/bluebird-3.5.4.tgz#d6cc661595de30d5b3af5fcedd3c0b3ef6ec5714"
+ integrity sha512-FG+nFEZChJrbQ9tIccIfZJBz3J7mLrAhxakAbnrJWn8d7aKOC+LWifa0G+p4ZqKp4y13T7juYvdhq9NzKdsrjw==
+
boxen@^1.2.1:
version "1.3.0"
resolved "https://registry.yarnpkg.com/boxen/-/boxen-1.3.0.tgz#55c6c39a8ba58d9c61ad22cd877532deb665a20b"
@@ -1851,7 +1865,7 @@ debug@3.1.0:
dependencies:
ms "2.0.0"
-debug@^2.1.2, debug@^2.2.0, debug@^2.3.3, debug@^2.6.8, debug@^2.6.9:
+debug@^2.1.2, debug@^2.2.0, debug@^2.3.3, debug@^2.6.8, debug@^2.6.9, debug@~2.6.9:
version "2.6.9"
resolved "https://registry.yarnpkg.com/debug/-/debug-2.6.9.tgz#5d128515df134ff327e90a4c93f4e077a536341f"
dependencies:
@@ -3062,7 +3076,7 @@ inflight@^1.0.4:
once "^1.3.0"
wrappy "1"
-inherits@2, inherits@^2.0.1, inherits@^2.0.3, inherits@~2.0.0, inherits@~2.0.3:
+inherits@2, inherits@^2.0.1, inherits@^2.0.3, inherits@~2.0.0, inherits@~2.0.1, inherits@~2.0.3:
version "2.0.3"
resolved "https://registry.yarnpkg.com/inherits/-/inherits-2.0.3.tgz#633c2c83e3da42a502f52466022480f4208261de"
integrity sha1-Yzwsg+PaQqUC9SRmAiSA9CCCYd4=
@@ -3415,6 +3429,11 @@ is-windows@^1.0.2:
version "1.0.2"
resolved "https://registry.yarnpkg.com/is-windows/-/is-windows-1.0.2.tgz#d1850eb9791ecd18e6182ce12a30f396634bb19d"
+isarray@0.0.1:
+ version "0.0.1"
+ resolved "https://registry.yarnpkg.com/isarray/-/isarray-0.0.1.tgz#8a18acfca9a8f4177e09abfc6038939b05d1eedf"
+ integrity sha1-ihis/Kmo9Bd+Cav8YDiTmwXR7t8=
+
isarray@1.0.0, isarray@~1.0.0:
version "1.0.0"
resolved "https://registry.yarnpkg.com/isarray/-/isarray-1.0.0.tgz#bb935d48582cba168c06834957a54a3e07124f11"
@@ -5404,6 +5423,11 @@ qs@~6.5.2:
version "6.5.2"
resolved "https://registry.yarnpkg.com/qs/-/qs-6.5.2.tgz#cb3ae806e8740444584ef154ce8ee98d403f3e36"
+querystringify@^2.0.0:
+ version "2.1.1"
+ resolved "https://registry.yarnpkg.com/querystringify/-/querystringify-2.1.1.tgz#60e5a5fd64a7f8bfa4d2ab2ed6fdf4c85bad154e"
+ integrity sha512-w7fLxIRCRT7U8Qu53jQnJyPkYZIaR4n5151KMfcJlO/A9397Wxb1amJvROTK6TOnp7PfoAmg/qXiNHI+08jRfA==
+
quick-lru@^1.0.0:
version "1.1.0"
resolved "https://registry.yarnpkg.com/quick-lru/-/quick-lru-1.1.0.tgz#4360b17c61136ad38078397ff11416e186dcfbb8"
@@ -5500,6 +5524,16 @@ read@1, read@~1.0.1:
string_decoder "~1.1.1"
util-deprecate "~1.0.1"
+"readable-stream@1.x >=1.1.9":
+ version "1.1.14"
+ resolved "https://registry.yarnpkg.com/readable-stream/-/readable-stream-1.1.14.tgz#7cf4c54ef648e3813084c636dd2079e166c081d9"
+ integrity sha1-fPTFTvZI44EwhMY23SB54WbAgdk=
+ dependencies:
+ core-util-is "~1.0.0"
+ inherits "~2.0.1"
+ isarray "0.0.1"
+ string_decoder "~0.10.x"
+
readdir-scoped-modules@^1.0.0:
version "1.0.2"
resolved "https://registry.yarnpkg.com/readdir-scoped-modules/-/readdir-scoped-modules-1.0.2.tgz#9fafa37d286be5d92cbaebdee030dc9b5f406747"
@@ -6558,6 +6592,14 @@ url-parse-lax@^1.0.0:
dependencies:
prepend-http "^1.0.1"
+url-parse@~1.4.3:
+ version "1.4.6"
+ resolved "https://registry.yarnpkg.com/url-parse/-/url-parse-1.4.6.tgz#baf91d6e6783c8a795eb476892ffef2737fc0456"
+ integrity sha512-/B8AD9iQ01seoXmXf9z/MjLZQIdOoYl/+gvsQF6+mpnxaTfG9P7srYaiqaDMyKkR36XMXfhqSHss5MyFAO8lew==
+ dependencies:
+ querystringify "^2.0.0"
+ requires-port "^1.0.0"
+
use@^3.1.0:
version "3.1.1"
resolved "https://registry.yarnpkg.com/use/-/use-3.1.1.tgz#d50c8cac79a19fbc20f2911f56eb973f4e10070f"

feat(messaging): transport - rmq.strategy (WIP)