Use protocolbuffers instead of gogo (#895)

* Use protocolbuffers instead of gogo Signed-off-by: Zixuan Liu <nodeces@gmail.com> * Improve Signed-off-by: Zixuan Liu <nodeces@gmail.com>
apache · Dec 15, 2022 · 504e589 · 504e589
1 parent 1fe9b62
commit 504e589
Show file tree

Hide file tree

Showing 22 changed files with 7,348 additions and 27,159 deletions.
diff --git a/go.mod b/go.mod
@@ -8,9 +8,7 @@ require (
 	github.com/DataDog/zstd v1.5.0
 	github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b
 	github.com/davecgh/go-spew v1.1.1
-	github.com/gogo/protobuf v1.3.2
 	github.com/golang-jwt/jwt v3.2.1+incompatible
-	github.com/golang/protobuf v1.5.2
 	github.com/google/uuid v1.1.2
 	github.com/klauspost/compress v1.14.4
 	github.com/linkedin/goavro/v2 v2.9.8
@@ -27,4 +25,5 @@ require (
 	go.uber.org/atomic v1.7.0
 	golang.org/x/mod v0.5.1
 	golang.org/x/oauth2 v0.0.0-20210402161424-2e8d93401602
+	google.golang.org/protobuf v1.26.0
 )
diff --git a/go.sum b/go.sum
@@ -118,7 +118,6 @@ github.com/godbus/dbus v0.0.0-20190726142602-4481cbc300e2 h1:ZpnhV/YsD2/4cESfV5+
 github.com/godbus/dbus v0.0.0-20190726142602-4481cbc300e2/go.mod h1:bBOAhwG1umN6/6ZUMtDFBMQR8jRg9O75tm9K00oMsK4=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
-github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang-jwt/jwt v3.2.1+incompatible h1:73Z+4BJcrTC+KczS6WvTPvRGOp1WmfEP4Q1lOd9Z/+c=
 github.com/golang-jwt/jwt v3.2.1+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I=
@@ -419,7 +418,6 @@ golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.1/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.4.2 h1:Gz96sIWK3OalVv/I/qNygP42zyoKp3xptRVCWRFEBvo=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.5.1 h1:OJxoQ/rynoF0dcCdI7cLPktw/hR2cueqYfjm43oqK38=
 golang.org/x/mod v0.5.1/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro=

diff --git a/integration-tests/pb/build.sh b/integration-tests/pb/build.sh
@@ -18,6 +18,4 @@
 # under the License.
 #
 
-pkg=pb
-protoc --go_out=import_path=${pkg}:. hello.proto
-
+protoc --go_out=. --go_opt=paths=source_relative hello.proto
diff --git a/integration-tests/pb/hello.pb.go b/integration-tests/pb/hello.pb.go
diff --git a/integration-tests/pb/hello.proto b/integration-tests/pb/hello.proto
@@ -19,6 +19,8 @@
 syntax = "proto3";
 package prototest;
 
+option go_package = "./pb";
+
 message Test {
     int32 num = 1;
     string msf = 2;

diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
@@ -28,7 +28,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/gogo/protobuf/proto"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/apache/pulsar-client-go/pulsar/crypto"
 	"github.com/apache/pulsar-client-go/pulsar/internal"

diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
@@ -34,10 +34,10 @@ import (
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	plog "github.com/apache/pulsar-client-go/pulsar/log"
-	"github.com/gogo/protobuf/proto"
 	"github.com/google/uuid"
 	"github.com/pierrec/lz4"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
 )
 
 var (

diff --git a/pulsar/helper.go b/pulsar/helper.go
@@ -22,7 +22,7 @@ import (
 
 	pkgerrors "github.com/pkg/errors"
 
-	"github.com/gogo/protobuf/proto"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"

diff --git a/pulsar/impl_message.go b/pulsar/impl_message.go
@@ -26,7 +26,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/gogo/protobuf/proto"
+	"google.golang.org/protobuf/proto"
 
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 )

diff --git a/pulsar/internal/batch_builder.go b/pulsar/internal/batch_builder.go
@@ -21,7 +21,7 @@ import (
 	"bytes"
 	"time"
 
-	"github.com/gogo/protobuf/proto"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal/compression"
 	"github.com/apache/pulsar-client-go/pulsar/internal/crypto"

diff --git a/pulsar/internal/commands.go b/pulsar/internal/commands.go
@@ -22,7 +22,7 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/gogo/protobuf/proto"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal/compression"
 	"github.com/apache/pulsar-client-go/pulsar/internal/crypto"
@@ -226,11 +226,11 @@ func baseCommand(cmdType pb.BaseCommand_Type, msg proto.Message) *pb.BaseCommand
 }
 
 func addSingleMessageToBatch(wb Buffer, smm *pb.SingleMessageMetadata, payload []byte) {
-	metadataSize := uint32(smm.Size())
+	metadataSize := uint32(proto.Size(smm))
 	wb.WriteUint32(metadataSize)
 
 	wb.ResizeIfNeeded(metadataSize)
-	_, err := smm.MarshalToSizedBuffer(wb.WritableSlice()[:metadataSize])
+	err := MarshalToSizedBuffer(smm, wb.WritableSlice()[:metadataSize])
 	if err != nil {
 		panic(fmt.Sprintf("Protobuf serialization error: %v", err))
 	}
@@ -282,7 +282,7 @@ func serializeMessage(wb Buffer,
 	// Write cmd
 	wb.WriteUint32(cmdSize)
 	wb.ResizeIfNeeded(cmdSize)
-	_, err = cmdSend.MarshalToSizedBuffer(wb.WritableSlice()[:cmdSize])
+	err = MarshalToSizedBuffer(cmdSend, wb.WritableSlice()[:cmdSize])
 	if err != nil {
 		panic(fmt.Sprintf("Protobuf error when serializing cmdSend: %v", err))
 	}
@@ -297,7 +297,7 @@ func serializeMessage(wb Buffer,
 	metadataStartIdx := wb.WriterIndex()
 	wb.WriteUint32(msgMetadataSize)
 	wb.ResizeIfNeeded(msgMetadataSize)
-	_, err = msgMetadata.MarshalToSizedBuffer(wb.WritableSlice()[:msgMetadataSize])
+	err = MarshalToSizedBuffer(msgMetadata, wb.WritableSlice()[:msgMetadataSize])
 	if err != nil {
 		panic(fmt.Sprintf("Protobuf error when serializing msgMetadata: %v", err))
 	}

diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
@@ -29,7 +29,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/gogo/protobuf/proto"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal/auth"
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
@@ -488,7 +488,7 @@ func (c *connection) internalWriteData(data Buffer) {
 func (c *connection) writeCommand(cmd *pb.BaseCommand) {
 	// Wire format
 	// [FRAME_SIZE] [CMD_SIZE][CMD]
-	cmdSize := uint32(cmd.Size())
+	cmdSize := uint32(proto.Size(cmd))
 	frameSize := cmdSize + 4
 
 	c.writeBufferLock.Lock()
@@ -499,7 +499,7 @@ func (c *connection) writeCommand(cmd *pb.BaseCommand) {
 
 	c.writeBuffer.WriteUint32(cmdSize)
 	c.writeBuffer.ResizeIfNeeded(cmdSize)
-	_, err := cmd.MarshalToSizedBuffer(c.writeBuffer.WritableSlice()[:cmdSize])
+	err := MarshalToSizedBuffer(cmd, c.writeBuffer.WritableSlice()[:cmdSize])
 	if err != nil {
 		c.log.WithError(err).Error("Protobuf serialization error")
 		panic("Protobuf serialization error")

diff --git a/pulsar/internal/connection_reader.go b/pulsar/internal/connection_reader.go
@@ -23,7 +23,7 @@ import (
 	"io"
 
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
-	"github.com/gogo/protobuf/proto"
+	"google.golang.org/protobuf/proto"
 )
 
 type connectionReader struct {

diff --git a/pulsar/internal/lookup_service.go b/pulsar/internal/lookup_service.go
@@ -22,7 +22,7 @@ import (
 	"fmt"
 	"net/url"
 
-	"github.com/gogo/protobuf/proto"
+	"google.golang.org/protobuf/proto"
 
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	"github.com/apache/pulsar-client-go/pulsar/log"