Implement possibility to get URL / path of current request and write it to message

Author: drublic

index.ts

@@ -1,6 +1,7 @@
 import * as merge from "lodash/merge";
 
 import ConfigInterface from "./lib/interfaces/ConfigInterface";
+import ProducerMessageInterface from "./lib/interfaces/ProducerMessageInterface";
 
 import Connector from "./lib/Connector";
 
@@ -35,6 +36,8 @@ const defaultOptions = {
   "retry.backoff.ms": 200,
   "socket.keepalive.enable": true,
   "workerPerPartition": 1,
+
+  "getPath": (message: ProducerMessageInterface): string => message.url,
 };
 
 export default (options: ConfigInterface) => {

lib/Consumer.ts

@@ -93,6 +93,7 @@ export default class Consumer extends EventEmitter {
 
       await this.publish(message.key, {
         content,
+        url: messageContent.url,
       });
     } catch (err) {
       this.handleError(err);

lib/Producer.ts

@@ -42,7 +42,10 @@ export default class Producer extends EventEmitter {
   public async produce(key: string, message: ProducerMessageInterface): Promise<void> {
     try {
       // With version = 1
-      message.path = "/missing"; // TODO: make this set-able via transform callback from config
+      if (this.config.getPath) {
+        message.path = this.config.getPath(message);
+      }
+
       await this.producer.buffer(this.config.produceTo, key, message, null);
 
       super.emit("info", `Message produced with id ${key}`);

lib/interfaces/ConfigInterface.ts

@@ -1,5 +1,6 @@
 import ConsumerContentInterface from "./ConsumerContentInterface";
 import LoggerInterface from "./LoggerInterface";
+import ProducerMessageInterface from "./ProducerMessageInterface";
 
 export default interface ConfigInterface {
   consumeFrom: string;
@@ -44,6 +45,7 @@ export default interface ConfigInterface {
     noBatchCommits?: boolean;
   };
   transformer: (message: ConsumerContentInterface) => Promise<string>;
+  getPath?: (message: ProducerMessageInterface) => string;
   "batch.num.messages"?: number;
   "compression.codec"?: "snappy";
   "dr_cb"?: boolean;

lib/interfaces/ProducerMessageInterface.ts

@@ -1,4 +1,5 @@
 export default interface ProducerMessageInterface {
   content: string;
   path?: string;
+  url: string;
 }