backend/business-management/bin/tools/broker/PubSubBroker.js
/**
     * Config the broker to listen to several topics
     * Returns an observable that resolves to a stream of subscribed topics
     * @param {Array} topics topics to listen
     */
Config the broker to listen to several topics
Returns an observable that resolves to a stream of subscribed topics
@param {Array} topics topics to listen
Config the broker to listen to several topics Returns an observable that resolves to a stream of subscribed topics @param { Array } topics topics to listen
configMessageListener$(topics) {
        return Rx.Observable.create((observer) => {
            Rx.Observable.from(topics)
                .filter(topicName => Object.keys(this.listeningTopics).indexOf(topicName) === -1)
                .mergeMap(topicName => {
                    const subscriptionName = `${topicName}_business-management`;
                    return this.getSubscription$(topicName, subscriptionName)
                        .map(subsription => { 
                            return { topicName, subsription, subscriptionName };
                         })
                })
                .subscribe(
                    ({ topicName, subsription, subscriptionName }) => {
                        this.listeningTopics[topicName] = subscriptionName;
                        subsription.on(`message`, message => {
                            this.incomingMessages$.next(
                                {
                                    id: message.id,
                                    data: JSON.parse(message.data),
                                    attributes: message.attributes,
                                    correlationId: message.attributes.correlationId,
                                    topic: topicName,
                                    type: message.attributes.type,
                                }
                            );
                            message.ack();
                        });
                        observer.next(topicName);
                    },
                    (err) => {
                        console.error('Failed to obtain emi-gatewayReplies subscription', err);
                        observer.error(err);
                    },
                    () => {
                        observer.complete();
                    }
                )
        })
            .reduce((acc, topic) => {
                acc.push(topic);
                return acc;
            }, []);
        ;
    }
configMessageListener$ ( topics ) { return Rx . Observable . create ( ( observer ) => { Rx . Observable . from ( topics ) . filter ( topicName => Object . keys ( this . listeningTopics ) . indexOf ( topicName ) === - 1 ) . mergeMap ( topicName => { const subscriptionName = ` ${ topicName } ` ; return this . getSubscription$ ( topicName , subscriptionName ) . map ( subsription => { return { topicName , subsription , subscriptionName } ; } ) } ) . subscribe ( ( { topicName , subsription , subscriptionName } ) => { this . listeningTopics [ topicName ] = subscriptionName ; subsription . on ( ` ` , message => { this . incomingMessages$ . next ( { id : message . id , data : JSON . parse ( message . data ) , attributes : message . attributes , correlationId : message . attributes . correlationId , topic : topicName , type : message . attributes . type , } ) ; message . ack ( ) ; } ) ; observer . next ( topicName ) ; } , ( err ) => { console . error ( 'Failed to obtain emi-gatewayReplies subscription' , err ) ; observer . error ( err ) ; } , ( ) => { observer . complete ( ) ; } ) } ) . reduce ( ( acc , topic ) => { acc . push ( topic ) ; return acc ; } , [ ] ) ; ; }
Config the broker to listen to several topics
 Returns an observable that resolves to a stream of subscribed topics
 @param {Array} topics topics to listen
Config the broker to listen to several topics Returns an observable that resolves to a stream of subscribed topics @param { Array } topics topics to listen
// console.log(`Received message ${message.id}:`);