Merge pull request #22 from radiofrance/prePublish

Add prePublish option
radiofrance · Apr 19, 2021 · dbbe813 · dbbe813
2 parents 1fa6c96 + 233a290
commit dbbe813
Show file tree

Hide file tree

Showing 4 changed files with 42 additions and 0 deletions.
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
@@ -38,12 +38,14 @@ jobs:
       - run: curl -i -u guest:guest -H "content-type:application/json" -XPUT http://127.0.0.1:15672/api/queues/%2f/thirdQueue
       - run: curl -i -u guest:guest -H "content-type:application/json" -XPUT http://127.0.0.1:15672/api/queues/%2f/fourthQueue
       - run: curl -i -u guest:guest -H "content-type:application/json" -XPUT http://127.0.0.1:15672/api/queues/%2f/fithQueue
+      - run: curl -i -u guest:guest -H "content-type:application/json" -XPUT http://127.0.0.1:15672/api/queues/%2f/sixthQueue
       - run: curl -i -u guest:guest -H "content-type:application/json" -XPUT http://127.0.0.1:15672/api/queues/%2f/unackedQueue
       - run: curl -i -u guest:guest -H "content-type:application/json" -XPOST -d '{"routing_key":"*.random.routingKey.*"}' http://127.0.0.1:15672/api/bindings/%2f/e/rf/q/firstQueue
       - run: curl -i -u guest:guest -H "content-type:application/json" -XPOST -d '{"routing_key":"*.randomBis.routingKey.*"}' http://127.0.0.1:15672/api/bindings/%2f/e/rf/q/secondQueue
       - run: curl -i -u guest:guest -H "content-type:application/json" -XPOST -d '{"routing_key":"test3.*.routingKey.test3"}' http://127.0.0.1:15672/api/bindings/%2f/e/rf/q/thirdQueue
       - run: curl -i -u guest:guest -H "content-type:application/json" -XPOST -d '{"routing_key":"test4.*.routingKey.test4"}' http://127.0.0.1:15672/api/bindings/%2f/e/rf/q/fourthQueue
       - run: curl -i -u guest:guest -H "content-type:application/json" -XPOST -d '{"routing_key":"test5.*.routingKey.test5"}' http://127.0.0.1:15672/api/bindings/%2f/e/rf/q/fithQueue
+      - run: curl -i -u guest:guest -H "content-type:application/json" -XPOST -d '{"routing_key":"test6.*.routingKey.test6"}' http://127.0.0.1:15672/api/bindings/%2f/e/rf/q/sixthQueue
       - run: curl -i -u guest:guest -H "content-type:application/json" -XPOST -d '{"routing_key":"*.randomUnacked.routingKey.*"}' http://127.0.0.1:15672/api/bindings/%2f/e/rf/q/unackedQueue
 
       - run: npm install

diff --git a/README.md b/README.md
@@ -220,6 +220,15 @@ Parameters: `message` (see below), `subscriberResult` (ACK/NACK/REJECT)
 
 Function run after each message treatment.
 
+##### prePublish
+
+Type: `function`<br>
+Default: null
+Parameters: `routingKey`, `content`, `properties` (see publish method)
+Must return an object with {routingKey, content, properties}
+
+Function run before each publish call
+
 ### rabQ.start()
 
 Starts a connection.

diff --git a/index.js b/index.js
@@ -49,6 +49,7 @@ class RabQ extends EventEmitter {
 
     this.beforeHook = opts.beforeHook || (() => {});
     this.afterHook = opts.afterHook || (() => {});
+    this.prePublish = opts.prePublish || null;
 
     _connection.set(this, undefined);
     _channel.set(this, undefined);
@@ -192,6 +193,10 @@ class RabQ extends EventEmitter {
       properties.headers['x-query-token'] = uuid.v4();
     }
 
+    if (this.prePublish) {
+      ({routingKey, content, properties} = this.prePublish(routingKey, content, properties));
+    }
+
     this.emit('log', {
       level: 'info',
       uuid: content.uuid,

diff --git a/test/set-consumer.js b/test/set-consumer.js
@@ -201,6 +201,7 @@ test('set before and after hooks', async t => {
   c.afterHook = (msg, result) => {
     afterHookMsg = msg.test + result;
   };
+
   const p = await makeRabQ(c);
 
   p.subscribesTo(/test5\.random\.routingKey\.test5/, message => {
@@ -215,3 +216,28 @@ test('set before and after hooks', async t => {
       t.is(afterHookMsg, 'beforeACK');
     });
 });
+
+test('set prePublish', async t => {
+  t.plan(2);
+
+  const contentToSend = {toto: 'tata'};
+
+  const c = Object.assign({}, minimalOptions);
+  c.queues = 'sixthQueue';
+  c.prePublish = (routingKey, content, properties) => {
+    t.is(routingKey, 'test6.random.routingKey.test6');
+    properties.headers.test = 'TEST';
+    return {routingKey, content, properties};
+  };
+
+  const p = await makeRabQ(c);
+
+  p.subscribesTo(/test6\.random\.routingKey\.test6/, message => {
+    t.is(message.originMsg.properties.headers.test, 'TEST');
+    return Promise.resolve(message.ACK);
+  });
+
+  p.publish('test6.random.routingKey.test6', contentToSend);
+
+  return delay(1000);
+});