Empty string value/key shouldn't be treated as null

e2e/both.spec.js

@@ -306,6 +306,28 @@ describe('Consumer/Producer', function() {
     run_headers_test(done, headers);
   });
 
+  it('should be able to produce and consume messages: empty buffer key and empty value', function(done) {
+    this.timeout(20000);
+    var emptyString = '';
+    var key = Buffer.from(emptyString);
+    var value = Buffer.from('');
+
+    producer.setPollInterval(10);
+
+    consumer.once('data', function(message) {
+      t.notEqual(message.value, null, 'message should not be null');
+      t.equal(value.toString(), message.value.toString(), 'invalid message value');
+      t.equal(emptyString, message.key, 'invalid message key');
+      done();
+    });
+
+    consumer.subscribe([topic]);
+    consumer.consume();
+
+    setTimeout(function() {
+      producer.produce(topic, null, value, key);
+    }, 2000);
+  });
 
   it('should be able to produce and consume messages: empty key and empty value', function(done) {
     this.timeout(20000);

src/producer.cc

@@ -388,6 +388,12 @@ NAN_METHOD(Producer::NodeProduce) {
 
     message_buffer_length = node::Buffer::Length(message_buffer_object);
     message_buffer_data = node::Buffer::Data(message_buffer_object);
+    if (message_buffer_data == NULL) {
+      // empty string message buffer should not end up as null message
+      v8::Local<v8::Object> message_buffer_object_emptystring = Nan::NewBuffer(new char[0], 0).ToLocalChecked();
+      message_buffer_length = node::Buffer::Length(message_buffer_object_emptystring);
+      message_buffer_data = node::Buffer::Data(message_buffer_object_emptystring);
+    }
   }
 
   size_t key_buffer_length;
@@ -412,6 +418,12 @@ NAN_METHOD(Producer::NodeProduce) {
 
     key_buffer_length = node::Buffer::Length(key_buffer_object);
     key_buffer_data = node::Buffer::Data(key_buffer_object);
+    if (key_buffer_data == NULL) {
+      // empty string key buffer should not end up as null key
+        v8::Local<v8::Object> key_buffer_object_emptystring = Nan::NewBuffer(new char[0], 0).ToLocalChecked();
+        key_buffer_length = node::Buffer::Length(key_buffer_object_emptystring);
+        key_buffer_data = node::Buffer::Data(key_buffer_object_emptystring);
+    }
   } else {
     // If it was a string just use the utf8 value.
     v8::Local<v8::String> val = Nan::To<v8::String>(info[3]).ToLocalChecked();