Enhance HighLevelProducer to take schema serializers (#92)

Author: rayokota

lib/producer/high-level-producer.js

@@ -43,7 +43,37 @@ function createSerializer(serializer) {
   // operation if the number of parameters the function accepts is > 1
   return {
     apply: applyFn,
-    async: serializer.length > 1
+    async: serializer.length > 1,
+    needsTopic: false
+  };
+}
+
+/**
+ * Create a serializer that additionally takes the topic name
+ *
+ * Method simply wraps a serializer provided by a user
+ * so it adds context to the error
+ *
+ * @returns {function} Serialization function
+ */
+function createTopicSerializer(serializer) {
+  var applyFn = function serializationWrapper(t, v, cb) {
+    try {
+      return cb ? serializer(t, v, cb) : serializer(t, v);
+    } catch (e) {
+      var modifiedError = new Error('Could not serialize value: ' + e.message);
+      modifiedError.value = v;
+      modifiedError.serializer = serializer;
+      throw modifiedError;
+    }
+  };
+
+  // We can check how many parameters the function has and activate the asynchronous
+  // operation if the number of parameters the function accepts is > 2
+  return {
+    apply: applyFn,
+    async: serializer.length > 2,
+    needsTopic: true
   };
 }
 
@@ -256,10 +286,20 @@ HighLevelProducer.prototype._modifiedProduce = function(topic, partition, messag
 
   try {
     if (this.valueSerializer.async) {
-      // If this is async we need to give it a callback
-      this.valueSerializer.apply(message, valueSerializerCallback);
+      if (this.valueSerializer.needsTopic) {
+        // If this is async we need to give it a callback
+        this.valueSerializer.apply(topic, message, valueSerializerCallback);
+      } else {
+        // If this is async we need to give it a callback
+        this.valueSerializer.apply(message, valueSerializerCallback);
+      }
     } else {
-      var serializedValue = this.valueSerializer.apply(message);
+      var serializedValue;
+      if (this.valueSerializer.needsTopic) {
+        serializedValue = this.valueSerializer.apply(topic, message);
+      } else {
+        serializedValue = this.valueSerializer.apply(message);
+      }
       // Check if we were returned a promise in order to support promise behavior
       if (serializedValue &&
           typeof serializedValue.then === 'function' &&
@@ -272,10 +312,20 @@ HighLevelProducer.prototype._modifiedProduce = function(topic, partition, messag
     }
 
     if (this.keySerializer.async) {
-      // If this is async we need to give it a callback
-      this.keySerializer.apply(key, keySerializerCallback);
+      if (this.valueSerializer.needsTopic) {
+        // If this is async we need to give it a callback
+        this.keySerializer.apply(topic, key, keySerializerCallback);
+      } else {
+        // If this is async we need to give it a callback
+        this.keySerializer.apply(key, keySerializerCallback);
+      }
     } else {
-      var serializedKey = this.keySerializer.apply(key);
+      var serializedKey;
+      if (this.valueSerializer.needsTopic) {
+        serializedKey = this.keySerializer.apply(topic, key);
+      } else {
+        serializedKey = this.keySerializer.apply(key);
+      }
       // Check if we were returned a promise in order to support promise behavior
       if (serializedKey &&
           typeof serializedKey.then === 'function' &&
@@ -319,3 +369,21 @@ HighLevelProducer.prototype.setKeySerializer = function(serializer) {
 HighLevelProducer.prototype.setValueSerializer = function(serializer) {
   this.valueSerializer = createSerializer(serializer);
 };
+
+/**
+ * Set the topic-key serializer
+ *
+ * A serializer that takes the topic name in addition to the key.
+ */
+HighLevelProducer.prototype.setTopicKeySerializer = function(serializer) {
+  this.keySerializer = createTopicSerializer(serializer);
+};
+
+/**
+ * Set the topic-value serializer
+ *
+ * A serializer that takes the topic name in addition to the value.
+ */
+HighLevelProducer.prototype.setTopicValueSerializer = function(serializer) {
+  this.valueSerializer = createTopicSerializer(serializer);
+};

types/rdkafka.d.ts

@@ -296,6 +296,10 @@ export class HighLevelProducer extends Producer {
   setKeySerializer(serializer: (key: any) => MessageKey | Promise<MessageKey>): void;
   setValueSerializer(serializer: (value: any, cb: (err: any, value: MessageValue) => void) => void): void;
   setValueSerializer(serializer: (value: any) => MessageValue | Promise<MessageValue>): void;
+  setTopicKeySerializer(serializer: (topic: string, key: any, cb: (err: any, key: MessageKey) => void) => void): void;
+  setTopicKeySerializer(serializer: (topic: string, key: any) => MessageKey | Promise<MessageKey>): void;
+  setTopicValueSerializer(serializer: (topic: string, value: any, cb: (err: any, value: MessageValue) => void) => void): void;
+  setTopicValueSerializer(serializer: (topic: string, value: any) => MessageValue | Promise<MessageValue>): void;
 }
 
 export const features: string[];