Add listener for umb

Signed-off-by: Andrei Stepanov <astepano@redhat.com>

.gitignore

@@ -4,6 +4,9 @@
 /node_modules
 /.pnp
 .pnp.js
+env*
+messages-queue
+.vscode
 
 # testing
 /coverage

assets/config-default.yaml

@@ -1,7 +1,27 @@
 ---
 listener:
+  broker_umb:
+    # client_name must match CN from certificate Subject
+    client_name: ~
+    # if subscription_id is empty it will be auto-generated and have uuid form
+    subscription_id: ~
+    prefetch: 200
+    connection:
+      # maps to ConnectionOptions 'rhea-promise' package https://github.com/amqp/rhea
+      transport: 'tls'
+      host: ~
+      port: 5671
+      ca: ~
+      key: ~
+      cert: ~
+    failover:
+      # entries in form of: "host:port"
+      set: []
+    topics:
+      set:
+        - 'VirtualTopic.eng.brew.task.closed'
   # https://fedora-messaging.readthedocs.io/en/stable/quick-start.html
-  broker:
+  broker_rabbitmq:
     #
     # Take server's default value: channelMax
     # Default value for heartbeat doesn't work.
@@ -23,19 +43,19 @@ listener:
     # https://www.rabbitmq.com/consumer-prefetch.html
     #
     prefetch: 200
+    # * (star) can substitute for exactly one word.
+    # # (hash) can substitute for zero or more words.
+    topics:
+      set:
+        # https://fedora-fedmsg.readthedocs.io/en/latest/topics.html#buildsys
+        - 'org.centos.prod.buildsys.tag'
+        # - 'org.fedoraproject.prod.copr.#'
+        # - 'org.fedoraproject.prod.buildsys.tag'
+        # - 'org.centos.prod.ci.koji-build.test.*'
   # where to dump received messages
   # can be relative path to process.cwd()
   # the directroy will be created
   file_queue_path: 'messages-queue'
-  # * (star) can substitute for exactly one word.
-  # # (hash) can substitute for zero or more words.
-  topics:
-    set:
-      # https://fedora-fedmsg.readthedocs.io/en/latest/topics.html#buildsys
-      - 'org.centos.prod.buildsys.tag'
-      # - 'org.fedoraproject.prod.copr.#'
-      # - 'org.fedoraproject.prod.buildsys.tag'
-      # - 'org.centos.prod.ci.koji-build.test.*'
 loader:
   db:
     db_name: 'ci-messages'
@@ -109,17 +129,36 @@ env_to_config_map:
       db_collection_invalid: LOADER_DB_COLLECTION_INVALID
     file_queue_path: LOADER_FILE_QUEUE_PATH
   listener:
-    broker:
-      url: LISTENER_BROKER_URL
-      keypath: LISTENER_BROKER_KEYPATH
-      certpath: LISTENER_BROKER_CERTPATH
-      cacertpath: LISTENER_BROKER_CACERTPATH
-      exchange_name: LISTENER_BROKER_EXCHANGE_NAME
-      prefetch: LISTENER_BROKER_PREFETCH
-    topics:
-      set:
-        # use new line to separate
-        - LISTENER_TOPICS_SET
+    broker_umb:
+      client_name: LISTENER_BROKER_UMB_CLIENT_NAME
+      subscription_id: LISTENER_BROKER_UMB_SUBSCRIPTION_ID
+      prefetch: LISTENER_BROKER_UMB_PREFETCH
+      connection:
+        host: LISTENER_BROKER_UMB_CONNECTION_HOST
+        port: LISTENER_BROKER_UMB_CONNECTION_PORT
+        ca: LISTENER_BROKER_UMB_CONNECTION_CA
+        key: LISTENER_BROKER_UMB_CONNECTION_KEY
+        cert: LISTENER_BROKER_UMB_CONNECTION_CERT
+        transport: LISTENER_BROKER_UMB_TRANSPORT
+      failover:
+        set:
+          # use new line to separate
+          - LISTENER_BROKER_UMB_FAILOVER_SET
+      topics:
+        set:
+          # use new line to separate
+          - LISTENER_BROKER_UMB_TOPICS_SET
+    broker_rabbitmq:
+      url: LISTENER_BROKER_RABBITMQ_URL
+      keypath: LISTENER_BROKER_RABBITMQ_KEYPATH
+      certpath: LISTENER_BROKER_RABBITMQ_CERTPATH
+      cacertpath: LISTENER_BROKER_RABBITMQ_CACERTPATH
+      exchange_name: LISTENER_BROKER_RABBITMQ_EXCHANGE_NAME
+      prefetch: LISTENER_BROKER_RABBITMQ_PREFETCH
+      topics:
+        set:
+          # use new line to separate
+          - LISTENER_BROKER_RABBITMQ_TOPICS_SET
     file_queue_path: LISTENER_FILE_QUEUE_PATH
   koji_fp:
     host: KOJI_FP_HOST

dev-listener-fedora.json

@@ -0,0 +1,6 @@
+{
+  "watch": ["src"],
+  "ext": ".ts,.js",
+  "ignore": [],
+  "exec": "source ./env.sh && ts-node ./src/listener_fedora.ts | ./node_modules/.bin/pino-pretty"
+}

dev-listener-umb.json

@@ -0,0 +1,6 @@
+{
+  "watch": ["src"],
+  "ext": ".ts,.js",
+  "ignore": [],
+  "exec": "source ./env.sh && ts-node ./src/listener_umb.ts | ./node_modules/.bin/pino-pretty"
+}

package.json

@@ -5,8 +5,10 @@
   "description": "JS implementation of kai",
   "main": "index.js",
   "scripts": {
-    "dev:listener": "nodemon --config dev-listener.json",
-    "start:listener": "source ./env.sh && node build/listener.js | ./node_modules/.bin/pino-pretty",
+    "dev:listener-umb": "nodemon --config dev-listener-umb.json",
+    "start:listener-umb": "source ./env.sh && node build/listener_umb.js | ./node_modules/.bin/pino-pretty",
+    "dev:listener-fedora": "nodemon --config dev-listener-fedora.json",
+    "start:listener-fedora": "source ./env.sh && node build/listener_fedora.js | ./node_modules/.bin/pino-pretty",
     "dev:loader": "nodemon --config dev-loader.json",
     "start:loader": "source ./env.sh && node build/loader.js | ./node_modules/.bin/pino-pretty",
     "build": "rimraf ./build && tsc"
@@ -30,6 +32,8 @@
     "pino": "^6.12.0",
     "pino-pretty": "^4.8.0",
     "retry": "^0.12.0",
+    "rhea": "^2.0.3",
+    "rhea-promise": "^2.1.0",
     "uuid": "^8.3.2",
     "xmlrpc": "^1.3.2"
   },

src/cfg.ts

@@ -24,8 +24,7 @@ import path from 'path';
 import debug from 'debug';
 import yaml from 'js-yaml';
 import assert from 'assert';
-import { error } from 'console';
-import { SrvRecord } from 'dns';
+import { ConnectionOptions } from 'rhea-promise';
 
 const log = debug('kaijs:cfg');
 /** Default config must present */
@@ -70,7 +69,10 @@ const mk_config_from_env: any = _.flow(
               _.partial(_.get, process.env, _, undefined),
               _.cond([
                 [_.isUndefined, _.stubArray],
-                [_.stubTrue, _.partial(_.split, _, '\n')],
+                [
+                  _.stubTrue,
+                  _.flow(_.ary(_.trim, 1), _.partial(_.split, _, '\n')),
+                ],
               ])
             ),
           ]),
@@ -123,9 +125,8 @@ class Config {
     } catch (err) {
       console.warn('Cannot proceed default configuration: ', DEF_CFG_PATH);
       throw err;
-      //throw new Error(`Cannot proceed default configuration: ${DEF_CFG_PATH}`);
     }
-    log('Default config: %s', '\n' + yaml.dump(this.config_default));
+    //log('Default config: %s', '\n' + yaml.dump(this.config_default));
     var override_cfg_path: string;
     for (override_cfg_path of OVERRIDE_CFG_LOOKUP_PATHS) {
       log(override_cfg_path);
@@ -147,7 +148,10 @@ class Config {
     if (this.config_default != null && typeof this.config_default == 'object') {
       const env_to_config_map = _.get(this.config_default, 'env_to_config_map');
       this.config_from_env = mk_config_from_env(env_to_config_map);
-      log('Environment config: %s', '\n' + yaml.dump(this.config_from_env));
+      /**
+       * Uncomment to print Environment config
+       */
+      //log('Environment config: %s', '\n' + yaml.dump(this.config_from_env));
     }
     /** Priority order */
     _.defaultsDeep(
@@ -157,7 +161,10 @@ class Config {
       this.config_default
     );
     _.unset(this.config_active, 'env_to_config_map');
-    log('Active config: %s', '\n' + yaml.dump(this.config_active));
+    /**
+     * Uncomment to print whole active config
+     */
+    // log('Active config: %s', '\n' + yaml.dump(this.config_active));
     /** constructor in javascript returns this object automatically
      * constructor returns the type of the class, the constructor implicitly returns 'this'
      * Even though you technically can't extend a proxy, there is a way to force a class
@@ -179,16 +186,28 @@ export const getcfg = _.once((): Cfg => {
 
 export interface Cfg {
   listener: {
-    broker: {
+    broker_umb: {
+      client_name: string;
+      subscription_id: string;
+      connection: ConnectionOptions;
+      prefetch: number;
+      failover: {
+        set: string[];
+      };
+      topics: {
+        set: string[];
+      };
+    };
+    broker_rabbitmq: {
       url: string;
       keypath: string;
       certpath: string;
       cacertpath: string;
       exchange_name: string;
       prefetch: string;
-    };
-    topics: {
-      set: string[];
+      topics: {
+        set: string[];
+      };
     };
     file_queue_path: string;
   };