This is an automated email from the ASF dual-hosted git repository.
sanjeevrk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 61a0bb0 adding protobuf schema (#1908)
61a0bb0 is described below
commit 61a0bb0fb7fad61577890adf347b47eade5803dc
Author: Boyang Jerry Peng <[email protected]>
AuthorDate: Tue Jun 5 14:38:28 2018 -0700
adding protobuf schema (#1908)
* adding protobuf schema
* cleaning up
* adjusting pulsar client pom
* improve protobuf parsing
* remove extra space
---
pulsar-client/pom.xml | 14 ++++
.../pulsar/client/impl/schema/ProtobufSchema.java | 81 ++++++++++++++++++++++
.../pulsar/client/schemas/ProtobufSchemaTest.java | 41 ++++++-----
.../apache/pulsar/common/schema/SchemaType.java | 4 +-
4 files changed, 120 insertions(+), 20 deletions(-)
diff --git a/pulsar-client/pom.xml b/pulsar-client/pom.xml
index 653c17d..16ff286 100644
--- a/pulsar-client/pom.xml
+++ b/pulsar-client/pom.xml
@@ -90,6 +90,13 @@
</exclusion>
</exclusions>
</dependency>
+
+ <dependency>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ <version>${protobuf3.version}</version>
+ <scope>provided</scope>
+ </dependency>
<!-- httpclient-hostname-verification depends on below dependencies -->
<dependency>
@@ -108,6 +115,13 @@
<artifactId>jackson-module-jsonSchema</artifactId>
</dependency>
+ <!-- Testing dependencies -->
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-functions-proto</artifactId>
+ <version>${project.parent.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufSchema.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufSchema.java
new file mode 100644
index 0000000..42c9331
--- /dev/null
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/ProtobufSchema.java
@@ -0,0 +1,81 @@
+/**
+ * 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.
+ */
+package org.apache.pulsar.client.impl.schema;
+
+import com.google.protobuf.Parser;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SchemaSerializationException;
+import org.apache.pulsar.common.schema.SchemaInfo;
+import org.apache.pulsar.common.schema.SchemaType;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.Collections;
+import java.util.Map;
+
+public class ProtobufSchema<T extends com.google.protobuf.GeneratedMessageV3>
implements Schema<T> {
+
+ private SchemaInfo schemaInfo;
+ private Parser<T> tParser;
+
+ private ProtobufSchema(SchemaInfo schemaInfo, Class<T> pojo) {
+ this.schemaInfo = schemaInfo;
+ try {
+ T protoMessageInstance = (T)
pojo.getMethod("getDefaultInstance").invoke(null);
+ tParser = (Parser<T>) protoMessageInstance.getParserForType();
+ } catch (IllegalAccessException | InvocationTargetException |
NoSuchMethodException e) {
+ throw new IllegalArgumentException(e);
+ }
+ }
+
+ @Override
+ public byte[] encode(T message) {
+ return message.toByteArray();
+ }
+
+ @Override
+ public T decode(byte[] bytes) {
+ try {
+ return this.tParser.parseFrom(bytes);
+ } catch (Exception e) {
+ throw new RuntimeException(new SchemaSerializationException(e));
+ }
+ }
+
+ @Override
+ public SchemaInfo getSchemaInfo() {
+ return schemaInfo;
+ }
+
+ public static <T extends com.google.protobuf.GeneratedMessageV3>
ProtobufSchema<T> of(Class<T> pojo) {
+ return of(pojo, Collections.emptyMap());
+ }
+
+ public static <T extends com.google.protobuf.GeneratedMessageV3>
ProtobufSchema<T> of(
+ Class<T> pojo, Map<String, String> properties){
+
+ SchemaInfo info = new SchemaInfo();
+ info.setName("");
+ info.setProperties(properties);
+ info.setType(SchemaType.PROTOBUF);
+
+ //TODO determine best method to extract schema from a protobuf message
+ info.setSchema(null);
+ return new ProtobufSchema<>(info, pojo);
+ }
+}
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java
b/pulsar-client/src/test/java/org/apache/pulsar/client/schemas/ProtobufSchemaTest.java
similarity index 51%
copy from
pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java
copy to
pulsar-client/src/test/java/org/apache/pulsar/client/schemas/ProtobufSchemaTest.java
index ad56bf2..fe2c116 100644
---
a/pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java
+++
b/pulsar-client/src/test/java/org/apache/pulsar/client/schemas/ProtobufSchemaTest.java
@@ -16,24 +16,27 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.common.schema;
+package org.apache.pulsar.client.schemas;
-/**
- * Types of supported schema for Pulsar messages
- */
-public enum SchemaType {
- /**
- * No schema defined
- */
- NONE,
-
- /**
- * Simple String encoding with UTF-8
- */
- STRING,
-
- /**
- * JSON object encoding and validation
- */
- JSON
+import org.apache.pulsar.client.impl.schema.ProtobufSchema;
+import org.apache.pulsar.functions.proto.Function;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class ProtobufSchemaTest {
+
+ private static final String NAME = "foo";
+
+ @Test
+ public void testEncodeAndDecode() {
+ Function.FunctionDetails functionDetails =
Function.FunctionDetails.newBuilder().setName(NAME).build();
+
+ ProtobufSchema<Function.FunctionDetails> protobufSchema =
ProtobufSchema.of(Function.FunctionDetails.class);
+
+ byte[] bytes = protobufSchema.encode(functionDetails);
+
+ Function.FunctionDetails message = protobufSchema.decode(bytes);
+
+ Assert.assertEquals(message.getName(), NAME);
+ }
}
diff --git
a/pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java
b/pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java
index ad56bf2..44d78c9 100644
---
a/pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java
+++
b/pulsar-common/src/main/java/org/apache/pulsar/common/schema/SchemaType.java
@@ -35,5 +35,7 @@ public enum SchemaType {
/**
* JSON object encoding and validation
*/
- JSON
+ JSON,
+
+ PROTOBUF
}
--
To stop receiving notification emails like this one, please contact
[email protected].
