Merge pull request #39 from hapinessjs/next

Next into master

Author: fauvillej

README.md

@@ -453,6 +453,8 @@ To set up your development environment:
 [Back to top](#table-of-contents)
 
 ## Change History
+* v1.5.0 (2018-08-24)
+    * Add possibility to provide a custom MessageRouter
 * v1.4.3 (2018-08-20)
     * Emit RETRY_LIMIT_EXCEEDED error on ConnectionManager
 * v1.4.2 (2018-06-11)

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@hapiness/rabbitmq",
-  "version": "1.4.3",
+  "version": "1.5.0",
   "description": "Hapiness module for rabbitmq",
   "main": "commonjs/index.js",
   "types": "index.d.ts",

package.json

@@ -0,0 +1,18 @@
+import { createDecorator, CoreDecorator, Type } from '@hapiness/core';
+import { Options } from 'amqplib';
+import { ExchangeType, ChannelOptions } from '../interfaces';
+
+export interface ExchangeDecoratorInterface {
+    name: string;
+    type: ExchangeType;
+    options?: Options.AssertExchange;
+    channel?: ChannelOptions;
+    providers?: Array<Type<any> | any>;
+}
+export const Exchange: CoreDecorator<ExchangeDecoratorInterface> = createDecorator<ExchangeDecoratorInterface>('Exchange', {
+    name: undefined,
+    type: undefined,
+    options: undefined,
+    channel: undefined,
+    providers: [],
+});

src/module/decorators.ts

@@ -0,0 +1,3 @@
+export * from './exchange.decorator';
+export * from './message.decorator';
+export * from './queue.decorator';

src/module/decorators/exchange.decorator.ts

@@ -0,0 +1,21 @@
+import { Type } from '@hapiness/core';
+import { createDecorator, CoreDecorator } from '@hapiness/core';
+
+export interface MessageDecoratorInterface {
+    queue: Type<any>;
+    exchange?: Type<any>;
+    routingKey?: string | RegExp;
+    filter?: {
+        [key: string]: string | RegExp;
+    };
+    is_fallback?: boolean;
+    providers?: Array<Type<any> | any>;
+}
+export const Message: CoreDecorator<MessageDecoratorInterface> = createDecorator<MessageDecoratorInterface>('Message', {
+    queue: undefined,
+    exchange: undefined,
+    routingKey: undefined,
+    filter: undefined,
+    is_fallback: false,
+    providers: [],
+});

src/module/decorators/index.ts

@@ -0,0 +1,21 @@
+import { Type } from '@hapiness/core';
+import { createDecorator, CoreDecorator } from '@hapiness/core';
+import { Options } from 'amqplib';
+import { Bind, ChannelOptions } from '../interfaces';
+
+export interface QueueDecoratorInterface {
+    name: string;
+    binds?: Array<Bind>;
+    options?: Options.AssertQueue;
+    channel?: ChannelOptions;
+    force_json_decode?: boolean;
+    providers?: Array<Type<any> | any>;
+}
+export const Queue: CoreDecorator<QueueDecoratorInterface> = createDecorator<QueueDecoratorInterface>('Queue', {
+    name: undefined,
+    binds: undefined,
+    options: undefined,
+    channel: undefined,
+    force_json_decode: false,
+    providers: [],
+});

src/module/decorators/message.decorator.ts

@@ -0,0 +1,23 @@
+import { Observable } from 'rxjs/Observable';
+import { CoreModule, DependencyInjection } from '@hapiness/core';
+import { ConnectionManager } from '../managers/connection-manager';
+import { metadataFromDeclarations } from '../utils';
+import { ExchangeDecoratorInterface } from '../decorators';
+import { ExchangeManager } from '../managers/exchange-manager';
+import { ExchangeWrapper } from '../managers/exchange-wrapper';
+
+export default function buildExchanges(modules: CoreModule[], connection: ConnectionManager): Observable<any> {
+    return Observable.from(modules)
+        .filter(_module => !!_module)
+        .flatMap(_module =>
+            metadataFromDeclarations<ExchangeDecoratorInterface>(_module.declarations, 'Exchange')
+                .map(metadata => ({ metadata, _module }))
+        )
+        .flatMap(({ metadata, _module }) => DependencyInjection.instantiateComponent(metadata.token, _module.di)
+            .map(instance => ({ instance, _module, metadata })))
+        .flatMap(({ instance, _module, metadata }) => {
+            const exchange = new ExchangeManager(connection.defaultChannel, new ExchangeWrapper(instance, metadata.data));
+            return exchange.assert();
+        })
+        .toArray();
+}

src/module/decorators/queue.decorator.ts

@@ -0,0 +1,72 @@
+import { CoreModule, DependencyInjection, extractMetadataByDecorator, Type } from '@hapiness/core';
+import { ConnectionManager } from '../managers/connection-manager';
+import { Observable } from 'rxjs/Observable';
+import { metadataFromDeclarations } from '../utils';
+import { QueueDecoratorInterface, ExchangeDecoratorInterface } from '../decorators';
+import { getChannel } from './get-channel';
+import { QueueManager } from '../managers/queue-manager';
+import { QueueWrapper } from '../managers/queue-wrapper';
+import registerMessages from './register-messages';
+import { MessageRouterInterface } from '../interfaces/message-router';
+import { consumeQueue } from './consume-queue';
+
+export default function buildQueues(
+    modules: CoreModule[], connection: ConnectionManager, MessageRouter: Type<MessageRouterInterface>
+): Observable<any> {
+    return Observable.from(modules)
+        .filter(_module => !!_module)
+        .flatMap(_module =>
+            metadataFromDeclarations<QueueDecoratorInterface>(_module.declarations, 'Queue')
+                .map(metadata => ({ metadata, _module }))
+        )
+        .flatMap(({ metadata, _module }) =>
+            DependencyInjection.instantiateComponent(metadata.token, _module.di)
+                .map(instance => ({ instance, _module, metadata})))
+        // Assert queue
+        .mergeMap(({ instance, _module, metadata }) =>
+            metadata.data.channel ?
+                    getChannel(connection, metadata.data.channel)
+                    .map(channel => ({ instance, metadata, channel, _module })) :
+                Observable.of({ instance, metadata, channel: connection.defaultChannelManager, _module }))
+        .mergeMap(({ instance, metadata, channel, _module }) => {
+            const queue = new QueueManager(channel, new QueueWrapper(instance, metadata.data));
+            return Observable.forkJoin(queue.assert(), Observable.of(metadata), Observable.of(_module));
+        })
+        // Bind queue
+        .mergeMap(([queue, metadata, _module]) => {
+            if (Array.isArray(metadata.data.binds)) {
+                return Observable.forkJoin(
+                    metadata.data.binds.map(bind => {
+                        if (Array.isArray(bind.pattern)) {
+                            return Observable.forkJoin(bind.pattern.map(pattern => queue.bind(
+                                extractMetadataByDecorator<ExchangeDecoratorInterface>(bind.exchange, 'Exchange').name, pattern)));
+                        }
+
+                        return queue.bind(
+                            extractMetadataByDecorator<ExchangeDecoratorInterface>(bind.exchange, 'Exchange').name, bind.pattern
+                        );
+                    })).map(() => ({ queue, _module }));
+            }
+
+            return Observable.of(({ queue, _module }));
+        })
+        // Register messages related to queue
+        // Consume queue
+        // Dont consume queue if there are no messages or consume() method on queue
+        .mergeMap(({ queue, _module }) => {
+            const messageRouter = new MessageRouter();
+            return registerMessages(modules, queue, messageRouter)
+                .defaultIfEmpty(null)
+                .filter(item => !!item)
+                .toArray()
+                .switchMap((registeredMessages) => {
+                    const _queue = queue.getQueue();
+                    if (registeredMessages.length || typeof _queue['onMessage'] === 'function') {
+                        return consumeQueue(queue, messageRouter);
+                    }
+
+                    return Observable.of(null);
+                });
+        })
+        .toArray();
+}