Support protobuf native schema

Author: shibd

.eslintignore

@@ -19,3 +19,4 @@
 
 examples
 perf
+tests/protobuf_schema/generated

index.d.ts

@@ -225,6 +225,21 @@ export interface SchemaInfo {
   properties?: Record<string, string>;
 }
 
+export namespace ProtobufNativeSchema {
+  interface CreateSchemaInfoFromRootOptions {
+    root: any;
+    rootMessageTypeName: string;
+    rootFileDescriptorName: string;
+    schemaType?: 'ProtobufNative';
+    syntax?: 'proto3' | 'proto2' | string;
+    name?: string;
+    properties?: Record<string, string>;
+  }
+
+  function createRootFromJson(rootJson: Record<string, unknown>): any;
+  function createSchemaInfoFromRoot(options: CreateSchemaInfoFromRootOptions): SchemaInfo;
+}
+
 export interface DeadLetterPolicy {
   deadLetterTopic: string;
   maxRedeliverCount?: number;
@@ -385,6 +400,7 @@ export type SchemaType =
   'Float32' |
   'Float64' |
   'KeyValue' |
+  'ProtobufNative' |
   'Bytes' |
   'AutoConsume' |
   'AutoPublish';

index.js

@@ -24,6 +24,7 @@ const AuthenticationToken = require('./src/AuthenticationToken');
 const AuthenticationOauth2 = require('./src/AuthenticationOauth2');
 const AuthenticationBasic = require('./src/AuthenticationBasic');
 const Client = require('./src/Client');
+const ProtobufNativeSchema = require('./src/ProtobufNativeSchema');
 
 const LogLevel = {
   DEBUG: 0,
@@ -43,6 +44,7 @@ const Pulsar = {
   AuthenticationToken,
   AuthenticationOauth2,
   AuthenticationBasic,
+  ProtobufNativeSchema,
   LogLevel,
 };
 

package-lock.json

@@ -61,7 +61,8 @@
     "license-checker": "^25.0.1",
     "lodash": "^4.17.21",
     "typedoc": "^0.23.28",
-    "typescript": "^4.9.5"
+    "typescript": "^4.9.5",
+    "protobufjs": "^8.4.2"
   },
   "dependencies": {
     "@mapbox/node-pre-gyp": "^2.0.3",

package.json

@@ -0,0 +1,76 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+require('protobufjs/ext/descriptor');
+
+const protobuf = require('protobufjs');
+const descriptor = require('protobufjs/ext/descriptor');
+
+const normalizeTypeName = (typeName) => typeName.replace(/^\./, '');
+
+const createSchemaInfoFromRoot = ({
+  root,
+  rootMessageTypeName,
+  rootFileDescriptorName,
+  schemaType = 'ProtobufNative',
+  syntax = 'proto3',
+  name = rootMessageTypeName,
+  properties = {},
+}) => {
+  if (!root) {
+    throw new Error('root is required');
+  }
+  if (!rootMessageTypeName) {
+    throw new Error('rootMessageTypeName is required');
+  }
+  if (!rootFileDescriptorName) {
+    throw new Error('rootFileDescriptorName is required');
+  }
+
+  const normalizedTypeName = normalizeTypeName(rootMessageTypeName);
+  const rootMessageType = root.lookupType(normalizedTypeName);
+  const packageName = normalizedTypeName.split('.').slice(0, -1).join('.');
+  const namespace = packageName ? root.lookup(packageName) : root;
+
+  // protobufjs reflection JSON does not retain the source file name. Set it
+  // before exporting a FileDescriptorSet, mirroring descriptor->file()->name().
+  namespace.filename = rootFileDescriptorName;
+  root.resolveAll();
+
+  const fileDescriptorSet = root.toDescriptor(syntax);
+  const fileDescriptorSetBytes = descriptor.FileDescriptorSet.encode(fileDescriptorSet).finish();
+
+  return {
+    schemaType,
+    name,
+    schema: JSON.stringify({
+      fileDescriptorSet: Buffer.from(fileDescriptorSetBytes).toString('base64'),
+      rootMessageTypeName: normalizeTypeName(rootMessageType.fullName),
+      rootFileDescriptorName,
+    }),
+    properties,
+  };
+};
+
+const createRootFromJson = (rootJson) => protobuf.Root.fromJSON(rootJson);
+
+module.exports = {
+  createRootFromJson,
+  createSchemaInfoFromRoot,
+};

src/ProtobufNativeSchema.js

@@ -17,35 +17,47 @@
  * under the License.
  */
 #include "SchemaInfo.h"
+#include <pulsar/ConsumerConfiguration.h>
+#include <pulsar/ProducerConfiguration.h>
 #include <map>
 
 static const std::string CFG_SCHEMA_TYPE = "schemaType";
 static const std::string CFG_NAME = "name";
 static const std::string CFG_SCHEMA = "schema";
 static const std::string CFG_PROPS = "properties";
 
-static const std::map<std::string, pulsar_schema_type> SCHEMA_TYPE = {{"None", pulsar_None},
-                                                                      {"String", pulsar_String},
-                                                                      {"Json", pulsar_Json},
-                                                                      {"Protobuf", pulsar_Protobuf},
-                                                                      {"Avro", pulsar_Avro},
-                                                                      {"Boolean", pulsar_Boolean},
-                                                                      {"Int8", pulsar_Int8},
-                                                                      {"Int16", pulsar_Int16},
-                                                                      {"Int32", pulsar_Int32},
-                                                                      {"Int64", pulsar_Int64},
-                                                                      {"Float32", pulsar_Float32},
-                                                                      {"Float64", pulsar_Float64},
-                                                                      {"KeyValue", pulsar_KeyValue},
-                                                                      {"Bytes", pulsar_Bytes},
-                                                                      {"AutoConsume", pulsar_AutoConsume},
-                                                                      {"AutoPublish", pulsar_AutoPublish}};
+struct _pulsar_producer_configuration {
+  pulsar::ProducerConfiguration conf;
+};
 
-SchemaInfo::SchemaInfo(const Napi::Object &schemaInfo) : cSchemaType(pulsar_Bytes), name("BYTES"), schema() {
-  this->cProperties = pulsar_string_map_create();
+struct _pulsar_consumer_configuration {
+  pulsar::ConsumerConfiguration consumerConfiguration;
+};
+
+static const std::map<std::string, pulsar::SchemaType> SCHEMA_TYPE = {
+    {"None", static_cast<pulsar::SchemaType>(0)},
+    {"String", static_cast<pulsar::SchemaType>(1)},
+    {"Json", static_cast<pulsar::SchemaType>(2)},
+    {"Protobuf", static_cast<pulsar::SchemaType>(3)},
+    {"Avro", static_cast<pulsar::SchemaType>(4)},
+    {"Boolean", static_cast<pulsar::SchemaType>(5)},
+    {"Int8", static_cast<pulsar::SchemaType>(6)},
+    {"Int16", static_cast<pulsar::SchemaType>(7)},
+    {"Int32", static_cast<pulsar::SchemaType>(8)},
+    {"Int64", static_cast<pulsar::SchemaType>(9)},
+    {"Float32", static_cast<pulsar::SchemaType>(10)},
+    {"Float64", static_cast<pulsar::SchemaType>(11)},
+    {"KeyValue", static_cast<pulsar::SchemaType>(15)},
+    {"ProtobufNative", static_cast<pulsar::SchemaType>(20)},
+    {"Bytes", static_cast<pulsar::SchemaType>(-1)},
+    {"AutoConsume", static_cast<pulsar::SchemaType>(-3)},
+    {"AutoPublish", static_cast<pulsar::SchemaType>(-4)}};
+
+SchemaInfo::SchemaInfo(const Napi::Object &schemaInfo)
+    : schemaType(static_cast<pulsar::SchemaType>(-1)), name("BYTES"), schema() {
   if (schemaInfo.Has(CFG_SCHEMA_TYPE) && schemaInfo.Get(CFG_SCHEMA_TYPE).IsString()) {
     this->name = schemaInfo.Get(CFG_SCHEMA_TYPE).ToString().Utf8Value();
-    this->cSchemaType = SCHEMA_TYPE.at(schemaInfo.Get(CFG_SCHEMA_TYPE).ToString().Utf8Value());
+    this->schemaType = SCHEMA_TYPE.at(schemaInfo.Get(CFG_SCHEMA_TYPE).ToString().Utf8Value());
   }
   if (schemaInfo.Has(CFG_NAME) && schemaInfo.Get(CFG_NAME).IsString()) {
     this->name = schemaInfo.Get(CFG_NAME).ToString().Utf8Value();
@@ -60,19 +72,19 @@ SchemaInfo::SchemaInfo(const Napi::Object &schemaInfo) : cSchemaType(pulsar_Byte
     for (int i = 0; i < size; i++) {
       Napi::String key = arr.Get(i).ToString();
       Napi::String value = propObj.Get(key).ToString();
-      pulsar_string_map_put(this->cProperties, key.Utf8Value().c_str(), value.Utf8Value().c_str());
+      this->properties[key.Utf8Value()] = value.Utf8Value();
     }
   }
 }
 
 void SchemaInfo::SetProducerSchema(std::shared_ptr<pulsar_producer_configuration_t> cProducerConfiguration) {
-  pulsar_producer_configuration_set_schema_info(cProducerConfiguration.get(), this->cSchemaType,
-                                                this->name.c_str(), this->schema.c_str(), this->cProperties);
+  cProducerConfiguration->conf.setSchema(
+      pulsar::SchemaInfo(this->schemaType, this->name, this->schema, this->properties));
 }
 
 void SchemaInfo::SetConsumerSchema(std::shared_ptr<pulsar_consumer_configuration_t> cConsumerConfiguration) {
-  pulsar_consumer_configuration_set_schema_info(cConsumerConfiguration.get(), this->cSchemaType,
-                                                this->name.c_str(), this->schema.c_str(), this->cProperties);
+  cConsumerConfiguration->consumerConfiguration.setSchema(
+      pulsar::SchemaInfo(this->schemaType, this->name, this->schema, this->properties));
 }
 
-SchemaInfo::~SchemaInfo() { pulsar_string_map_free(this->cProperties); }
+SchemaInfo::~SchemaInfo() {}

src/SchemaInfo.cc

@@ -20,9 +20,13 @@
 #ifndef SCHEMA_INFO_H
 #define SCHEMA_INFO_H
 
+#include <map>
+#include <memory>
 #include <napi.h>
-#include <pulsar/c/producer_configuration.h>
+#include <pulsar/Schema.h>
 #include <pulsar/c/consumer_configuration.h>
+#include <pulsar/c/producer_configuration.h>
+#include <string>
 
 class SchemaInfo {
  public:
@@ -32,10 +36,10 @@ class SchemaInfo {
   void SetConsumerSchema(std::shared_ptr<pulsar_consumer_configuration_t> cConsumerConfiguration);
 
  private:
-  pulsar_schema_type cSchemaType;
+  pulsar::SchemaType schemaType;
   std::string name;
   std::string schema;
-  pulsar_string_map_t *cProperties;
+  std::map<std::string, std::string> properties;
 };
 
 #endif

src/SchemaInfo.h

@@ -0,0 +1,96 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+const Pulsar = require('../index');
+const { pulsar } = require('./protobuf_schema/generated/user_event_pb');
+const userEventRootJson = require('./protobuf_schema/generated/user_event_root.json');
+
+(() => {
+  describe('ProtobufNativeSchema', () => {
+    test('produce and consume protobuf native messages', async () => {
+      const topic = `persistent://public/default/protobuf-native-schema-${Date.now()}`;
+      const subscription = `protobuf-native-sub-${Date.now()}`;
+      const userEventRoot = Pulsar.ProtobufNativeSchema.createRootFromJson(userEventRootJson);
+      const schema = Pulsar.ProtobufNativeSchema.createSchemaInfoFromRoot({
+        root: userEventRoot,
+        rootMessageTypeName: 'pulsar.example.UserEvent',
+        rootFileDescriptorName: 'user_event.proto',
+      });
+
+      expect(schema.schemaType).toBe('ProtobufNative');
+
+      const client = new Pulsar.Client({
+        serviceUrl: 'pulsar://localhost:6650',
+        operationTimeoutSeconds: 30,
+      });
+
+      let producer;
+      let consumer;
+
+      try {
+        consumer = await client.subscribe({
+          topic,
+          subscription,
+          subscriptionType: 'Shared',
+          ackTimeoutMs: 10000,
+          schema,
+        });
+
+        producer = await client.createProducer({
+          topic,
+          sendTimeoutMs: 30000,
+          batchingEnabled: true,
+          schema,
+        });
+
+        const sent = [];
+        for (let i = 0; i < 5; i += 1) {
+          const userEvent = pulsar.example.UserEvent.create({
+            id: `user-${i}`,
+            name: `User ${i}`,
+            age: 20 + i,
+            tags: ['nodejs', 'protobuf'],
+          });
+          const data = Buffer.from(pulsar.example.UserEvent.encode(userEvent).finish());
+          sent.push(pulsar.example.UserEvent.toObject(userEvent));
+          await producer.send({ data });
+        }
+        await producer.flush();
+
+        const received = [];
+        for (let i = 0; i < sent.length; i += 1) {
+          const msg = await consumer.receive();
+          const userEvent = pulsar.example.UserEvent.decode(msg.getData());
+          received.push(pulsar.example.UserEvent.toObject(userEvent));
+          await consumer.acknowledge(msg);
+        }
+
+        expect(received).toEqual(sent);
+      } finally {
+        if (producer) {
+          await producer.close();
+        }
+        if (consumer) {
+          await consumer.close();
+        }
+        await client.close();
+      }
+    }, 30000);
+  });
+})();