Add schema logic in producer and consumer

.golangci.yml

@@ -13,10 +13,8 @@ linters:
     - golint
     - gosimple
     - govet
-    - ineffassign
     - interfacer
     - misspell
-    - staticcheck
     - structcheck
     - stylecheck
     - typecheck

go.mod

@@ -7,10 +7,14 @@ require (
 	github.com/apache/pulsar-client-go/oauth2 v0.0.0-20200715083626-b9f8c5cedefb
 	github.com/beefsack/go-rate v0.0.0-20180408011153-efa7637bb9b6
 	github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b
+	github.com/davecgh/go-spew v1.1.1
 	github.com/gogo/protobuf v1.3.1
+	github.com/golang/protobuf v1.4.2
+	github.com/golang/snappy v0.0.1 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
 	github.com/klauspost/compress v1.10.8
 	github.com/kr/pretty v0.2.0 // indirect
+	github.com/linkedin/goavro/v2 v2.9.8 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.1 // indirect
 	github.com/pierrec/lz4 v2.0.5+incompatible

go.sum

@@ -56,6 +56,8 @@ github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:W
 github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
 github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
+github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
+github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
@@ -88,6 +90,11 @@ github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfn
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/linkedin/goavro v1.0.5 h1:6ds0AI8upkEoafDk0a5r9q1p/xRtMq47jCilZYEqbmg=
+github.com/linkedin/goavro v2.1.0+incompatible h1:DV2aUlj2xZiuxQyvag8Dy7zjY69ENjS66bWkSfdpddY=
+github.com/linkedin/goavro v2.1.0+incompatible/go.mod h1:bBCwI2eGYpUI/4820s67MElg9tdeLbINjLjiM2xZFYM=
+github.com/linkedin/goavro/v2 v2.9.8 h1:jN50elxBsGBDGVDEKqUlDuU1cFwJ11K/yrJCBMe/7Wg=
+github.com/linkedin/goavro/v2 v2.9.8/go.mod h1:UgQUb2N/pmueQYH9bfqFioWxzYCZXSfF8Jw03O5sjqA=
 github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=

integration-tests/license_test.go

@@ -67,6 +67,7 @@ var otherCheck = regexp.MustCompile(`#
 var skip = map[string]bool{
 	"../pulsar/internal/pulsar_proto/PulsarApi.pb.go": true,
 	"../.github/workflows/bot.yaml":                   true,
+	"../integration-tests/pb/hello.pb.go":             true,
 }
 
 func TestLicense(t *testing.T) {

integration-tests/pb/build.sh

@@ -0,0 +1,23 @@
+#!/usr/bin/env bash
+#
+# 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.
+#
+
+pkg=pb
+protoc --go_out=import_path=${pkg}:. hello.proto
+

integration-tests/pb/hello.proto

@@ -0,0 +1,25 @@
+/**
+ * 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.
+ */
+syntax = "proto3";
+package prototest;
+
+message Test {
+    int32 num = 1;
+    string msf = 2;
+}

pulsar/consumer.go

@@ -151,6 +151,8 @@ type ConsumerOptions struct {
 	// A chain of interceptors, These interceptors will be called at some points defined in ConsumerInterceptor interface.
 	Interceptors ConsumerInterceptors
 
+	Schema Schema
+
 	// MaxReconnectToBroker set the maximum retry number of reconnectToBroker. (default: ultimate)
 	MaxReconnectToBroker *uint
 }

pulsar/consumer_impl.go

@@ -103,6 +103,12 @@ func newConsumer(client *client, options ConsumerOptions) (Consumer, error) {
 		options.Name = generateRandomName()
 	}
 
+	if options.Schema != nil && options.Schema.GetSchemaInfo() != nil {
+		if options.Schema.GetSchemaInfo().Type == NONE {
+			options.Schema = NewBytesSchema(nil)
+		}
+	}
+
 	// did the user pass in a message channel?
 	messageCh := options.MessageChannel
 	if options.MessageChannel == nil {
@@ -311,6 +317,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 				interceptors:               c.options.Interceptors,
 				maxReconnectToBroker:       c.options.MaxReconnectToBroker,
 				keySharedPolicy:            c.options.KeySharedPolicy,
+				schema:                     c.options.Schema,
 			}
 			cons, err := newPartitionConsumer(c, c.client, opts, c.messageCh, c.dlq)
 			ch <- ConsumerError{

pulsar/consumer_partition.go

@@ -122,6 +122,7 @@ type partitionConsumerOpts struct {
 	interceptors               ConsumerInterceptors
 	maxReconnectToBroker       *uint
 	keySharedPolicy            *KeySharedPolicy
+	schema                     Schema
 }
 
 type partitionConsumer struct {
@@ -521,6 +522,7 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 				topic:               pc.topic,
 				msgID:               msgID,
 				payLoad:             payload,
+				schema:              pc.options.schema,
 				replicationClusters: msgMeta.GetReplicateTo(),
 				replicatedFrom:      msgMeta.GetReplicatedFrom(),
 				redeliveryCount:     response.GetRedeliveryCount(),
@@ -535,6 +537,7 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 				topic:               pc.topic,
 				msgID:               msgID,
 				payLoad:             payload,
+				schema:              pc.options.schema,
 				replicationClusters: msgMeta.GetReplicateTo(),
 				replicatedFrom:      msgMeta.GetReplicatedFrom(),
 				redeliveryCount:     response.GetRedeliveryCount(),
@@ -862,6 +865,23 @@ func (pc *partitionConsumer) grabConn() error {
 	initialPosition := toProtoInitialPosition(pc.options.subscriptionInitPos)
 	keySharedMeta := toProtoKeySharedMeta(pc.options.keySharedPolicy)
 	requestID := pc.client.rpcClient.NewRequestID()
+
+	pbSchema := new(pb.Schema)
+
+	if pc.options.schema != nil && pc.options.schema.GetSchemaInfo() != nil {
+		tmpSchemaType := pb.Schema_Type(int32(pc.options.schema.GetSchemaInfo().Type))
+		pbSchema = &pb.Schema{
+			Name:       proto.String(pc.options.schema.GetSchemaInfo().Name),
+			Type:       &tmpSchemaType,
+			SchemaData: []byte(pc.options.schema.GetSchemaInfo().Schema),
+			Properties: internal.ConvertFromStringMap(pc.options.schema.GetSchemaInfo().Properties),
+		}
+		pc.log.Debugf("The partition consumer schema name is: %s", pbSchema.Name)
+	} else {
+		pbSchema = nil
+		pc.log.Debug("The partition consumer schema is nil")
+	}
+
 	cmdSubscribe := &pb.CommandSubscribe{
 		Topic:                      proto.String(pc.topic),
 		Subscription:               proto.String(pc.options.subscription),
@@ -873,7 +893,7 @@ func (pc *partitionConsumer) grabConn() error {
 		Durable:                    proto.Bool(pc.options.subscriptionMode == durable),
 		Metadata:                   internal.ConvertFromStringMap(pc.options.metadata),
 		ReadCompacted:              proto.Bool(pc.options.readCompacted),
-		Schema:                     nil,
+		Schema:                     pbSchema,
 		InitialPosition:            initialPosition.Enum(),
 		ReplicateSubscriptionState: proto.Bool(pc.options.replicateSubscriptionState),
 		KeySharedMeta:              keySharedMeta,

pulsar/impl_message.go

@@ -178,6 +178,7 @@ type message struct {
 	replicationClusters []string
 	replicatedFrom      string
 	redeliveryCount     uint32
+	schema              Schema
 }
 
 func (msg *message) Topic() string {
@@ -220,6 +221,10 @@ func (msg *message) GetReplicatedFrom() string {
 	return msg.replicatedFrom
 }
 
+func (msg *message) GetSchemaValue(v interface{}) error {
+	return msg.schema.Decode(msg.payLoad, v)
+}
+
 func (msg *message) ProducerName() string {
 	return msg.producerName
 }

pulsar/message.go

@@ -27,6 +27,9 @@ type ProducerMessage struct {
 	// Payload for the message
 	Payload []byte
 
+	//Value and payload is mutually exclusive, `Value interface{}` for schema message.
+	Value interface{}
+
 	// Key sets the key of the message for routing policy
 	Key string
 
@@ -102,6 +105,9 @@ type Message interface {
 
 	// Get name of cluster, from which the message is replicated.
 	GetReplicatedFrom() string
+
+	//Get the de-serialized value of the message, according the configured
+	GetSchemaValue(v interface{}) error
 }
 
 // MessageID identifier for a particular message