- Switch from the flatbuffers to the protobuf

Signed-off-by: Valery Piashchynski <[email protected]>

6 files changed, 250 insertions, 214 deletions

diff --git a/pkg/pubsub/message.fbs b/pkg/pubsub/message.fbs
deleted file mode 100644
index 7e975894..00000000
--- a/pkg/pubsub/message.fbs
+++ /dev/null
@@ -1,14 +0,0 @@
-namespace message;
-
-table Message {
-    command:string;
-    broker:string;
-    topics:[string];
-    payload:[byte];
-}
-
-table Messages {
-    messages:[Message];
-}
-
-root_type Messages;
diff --git a/pkg/pubsub/message.go b/pkg/pubsub/message.go
deleted file mode 100644
index 74348722..00000000
--- a/pkg/pubsub/message.go
+++ /dev/null
@@ -1,15 +0,0 @@
-package pubsub
-
-type Message struct {
-	// Command (join, leave, headers)
-	Command string `json:"command"`
-
-	// Broker (redis, memory)
-	Broker string `json:"broker"`
-
-	// Topic message been pushed into.
-	Topics []string `json:"topic"`
-
-	// Payload to be broadcasted
-	Payload []byte `json:"payload"`
-}
diff --git a/pkg/pubsub/message.proto b/pkg/pubsub/message.proto
new file mode 100644
index 00000000..772e7611
--- /dev/null
+++ b/pkg/pubsub/message.proto
@@ -0,0 +1,15 @@
+syntax = "proto3";
+
+package pubsub.v1;
+option go_package = "./message";
+
+message Message {
+    string command = 1;
+    string broker = 2;
+    repeated string topics = 3;
+    bytes payload = 4;
+}
+
+message Messages {
+    repeated Message messages = 1;
+}
diff --git a/pkg/pubsub/message/Message.go b/pkg/pubsub/message/Message.go
deleted file mode 100644
index 26bbd12c..00000000
--- a/pkg/pubsub/message/Message.go
+++ /dev/null
@@ -1,118 +0,0 @@
-// Code generated by the FlatBuffers compiler. DO NOT EDIT.
-
-package message
-
-import (
-	flatbuffers "github.com/google/flatbuffers/go"
-)
-
-type Message struct {
-	_tab flatbuffers.Table
-}
-
-func GetRootAsMessage(buf []byte, offset flatbuffers.UOffsetT) *Message {
-	n := flatbuffers.GetUOffsetT(buf[offset:])
-	x := &Message{}
-	x.Init(buf, n+offset)
-	return x
-}
-
-func GetSizePrefixedRootAsMessage(buf []byte, offset flatbuffers.UOffsetT) *Message {
-	n := flatbuffers.GetUOffsetT(buf[offset+flatbuffers.SizeUint32:])
-	x := &Message{}
-	x.Init(buf, n+offset+flatbuffers.SizeUint32)
-	return x
-}
-
-func (rcv *Message) Init(buf []byte, i flatbuffers.UOffsetT) {
-	rcv._tab.Bytes = buf
-	rcv._tab.Pos = i
-}
-
-func (rcv *Message) Table() flatbuffers.Table {
-	return rcv._tab
-}
-
-func (rcv *Message) Command() []byte {
-	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
-	if o != 0 {
-		return rcv._tab.ByteVector(o + rcv._tab.Pos)
-	}
-	return nil
-}
-
-func (rcv *Message) Broker() []byte {
-	o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
-	if o != 0 {
-		return rcv._tab.ByteVector(o + rcv._tab.Pos)
-	}
-	return nil
-}
-
-func (rcv *Message) Topics(j int) []byte {
-	o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
-	if o != 0 {
-		a := rcv._tab.Vector(o)
-		return rcv._tab.ByteVector(a + flatbuffers.UOffsetT(j*4))
-	}
-	return nil
-}
-
-func (rcv *Message) TopicsLength() int {
-	o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
-	if o != 0 {
-		return rcv._tab.VectorLen(o)
-	}
-	return 0
-}
-
-func (rcv *Message) Payload(j int) int8 {
-	o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
-	if o != 0 {
-		a := rcv._tab.Vector(o)
-		return rcv._tab.GetInt8(a + flatbuffers.UOffsetT(j*1))
-	}
-	return 0
-}
-
-func (rcv *Message) PayloadLength() int {
-	o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
-	if o != 0 {
-		return rcv._tab.VectorLen(o)
-	}
-	return 0
-}
-
-func (rcv *Message) MutatePayload(j int, n int8) bool {
-	o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
-	if o != 0 {
-		a := rcv._tab.Vector(o)
-		return rcv._tab.MutateInt8(a+flatbuffers.UOffsetT(j*1), n)
-	}
-	return false
-}
-
-func MessageStart(builder *flatbuffers.Builder) {
-	builder.StartObject(4)
-}
-func MessageAddCommand(builder *flatbuffers.Builder, command flatbuffers.UOffsetT) {
-	builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(command), 0)
-}
-func MessageAddBroker(builder *flatbuffers.Builder, broker flatbuffers.UOffsetT) {
-	builder.PrependUOffsetTSlot(1, flatbuffers.UOffsetT(broker), 0)
-}
-func MessageAddTopics(builder *flatbuffers.Builder, topics flatbuffers.UOffsetT) {
-	builder.PrependUOffsetTSlot(2, flatbuffers.UOffsetT(topics), 0)
-}
-func MessageStartTopicsVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
-	return builder.StartVector(4, numElems, 4)
-}
-func MessageAddPayload(builder *flatbuffers.Builder, payload flatbuffers.UOffsetT) {
-	builder.PrependUOffsetTSlot(3, flatbuffers.UOffsetT(payload), 0)
-}
-func MessageStartPayloadVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
-	return builder.StartVector(1, numElems, 1)
-}
-func MessageEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
-	return builder.EndObject()
-}
diff --git a/pkg/pubsub/message/Messages.go b/pkg/pubsub/message/Messages.go
deleted file mode 100644
index 633b367d..00000000
--- a/pkg/pubsub/message/Messages.go
+++ /dev/null
@@ -1,67 +0,0 @@
-// Code generated by the FlatBuffers compiler. DO NOT EDIT.
-
-package message
-
-import (
-	flatbuffers "github.com/google/flatbuffers/go"
-)
-
-type Messages struct {
-	_tab flatbuffers.Table
-}
-
-func GetRootAsMessages(buf []byte, offset flatbuffers.UOffsetT) *Messages {
-	n := flatbuffers.GetUOffsetT(buf[offset:])
-	x := &Messages{}
-	x.Init(buf, n+offset)
-	return x
-}
-
-func GetSizePrefixedRootAsMessages(buf []byte, offset flatbuffers.UOffsetT) *Messages {
-	n := flatbuffers.GetUOffsetT(buf[offset+flatbuffers.SizeUint32:])
-	x := &Messages{}
-	x.Init(buf, n+offset+flatbuffers.SizeUint32)
-	return x
-}
-
-func (rcv *Messages) Init(buf []byte, i flatbuffers.UOffsetT) {
-	rcv._tab.Bytes = buf
-	rcv._tab.Pos = i
-}
-
-func (rcv *Messages) Table() flatbuffers.Table {
-	return rcv._tab
-}
-
-func (rcv *Messages) Messages(obj *Message, j int) bool {
-	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
-	if o != 0 {
-		x := rcv._tab.Vector(o)
-		x += flatbuffers.UOffsetT(j) * 4
-		x = rcv._tab.Indirect(x)
-		obj.Init(rcv._tab.Bytes, x)
-		return true
-	}
-	return false
-}
-
-func (rcv *Messages) MessagesLength() int {
-	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
-	if o != 0 {
-		return rcv._tab.VectorLen(o)
-	}
-	return 0
-}
-
-func MessagesStart(builder *flatbuffers.Builder) {
-	builder.StartObject(1)
-}
-func MessagesAddMessages(builder *flatbuffers.Builder, messages flatbuffers.UOffsetT) {
-	builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(messages), 0)
-}
-func MessagesStartMessagesVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
-	return builder.StartVector(4, numElems, 4)
-}
-func MessagesEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
-	return builder.EndObject()
-}
diff --git a/pkg/pubsub/message/message.pb.go b/pkg/pubsub/message/message.pb.go
new file mode 100644
index 00000000..3a73c39c
--- /dev/null
+++ b/pkg/pubsub/message/message.pb.go
@@ -0,0 +1,235 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.26.0
+// 	protoc        v3.16.0
+// source: message.proto
+
+package message
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+type Message struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Command string   `protobuf:"bytes,1,opt,name=command,proto3" json:"command,omitempty"`
+	Broker  string   `protobuf:"bytes,2,opt,name=broker,proto3" json:"broker,omitempty"`
+	Topics  []string `protobuf:"bytes,3,rep,name=topics,proto3" json:"topics,omitempty"`
+	Payload []byte   `protobuf:"bytes,4,opt,name=payload,proto3" json:"payload,omitempty"`
+}
+
+func (x *Message) Reset() {
+	*x = Message{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_message_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *Message) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Message) ProtoMessage() {}
+
+func (x *Message) ProtoReflect() protoreflect.Message {
+	mi := &file_message_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Message.ProtoReflect.Descriptor instead.
+func (*Message) Descriptor() ([]byte, []int) {
+	return file_message_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *Message) GetCommand() string {
+	if x != nil {
+		return x.Command
+	}
+	return ""
+}
+
+func (x *Message) GetBroker() string {
+	if x != nil {
+		return x.Broker
+	}
+	return ""
+}
+
+func (x *Message) GetTopics() []string {
+	if x != nil {
+		return x.Topics
+	}
+	return nil
+}
+
+func (x *Message) GetPayload() []byte {
+	if x != nil {
+		return x.Payload
+	}
+	return nil
+}
+
+type Messages struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Messages []*Message `protobuf:"bytes,1,rep,name=messages,proto3" json:"messages,omitempty"`
+}
+
+func (x *Messages) Reset() {
+	*x = Messages{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_message_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *Messages) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Messages) ProtoMessage() {}
+
+func (x *Messages) ProtoReflect() protoreflect.Message {
+	mi := &file_message_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Messages.ProtoReflect.Descriptor instead.
+func (*Messages) Descriptor() ([]byte, []int) {
+	return file_message_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *Messages) GetMessages() []*Message {
+	if x != nil {
+		return x.Messages
+	}
+	return nil
+}
+
+var File_message_proto protoreflect.FileDescriptor
+
+var file_message_proto_rawDesc = []byte{
+	0x0a, 0x0d, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12,
+	0x09, 0x70, 0x75, 0x62, 0x73, 0x75, 0x62, 0x2e, 0x76, 0x31, 0x22, 0x6d, 0x0a, 0x07, 0x4d, 0x65,
+	0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x63, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x63, 0x6f, 0x6d, 0x6d, 0x61, 0x6e, 0x64, 0x12,
+	0x16, 0x0a, 0x06, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x06, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x6f, 0x70, 0x69, 0x63,
+	0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x73, 0x12,
+	0x18, 0x0a, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0c,
+	0x52, 0x07, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x22, 0x3a, 0x0a, 0x08, 0x4d, 0x65, 0x73,
+	0x73, 0x61, 0x67, 0x65, 0x73, 0x12, 0x2e, 0x0a, 0x08, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65,
+	0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x70, 0x75, 0x62, 0x73, 0x75, 0x62,
+	0x2e, 0x76, 0x31, 0x2e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, 0x08, 0x6d, 0x65, 0x73,
+	0x73, 0x61, 0x67, 0x65, 0x73, 0x42, 0x0b, 0x5a, 0x09, 0x2e, 0x2f, 0x6d, 0x65, 0x73, 0x73, 0x61,
+	0x67, 0x65, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_message_proto_rawDescOnce sync.Once
+	file_message_proto_rawDescData = file_message_proto_rawDesc
+)
+
+func file_message_proto_rawDescGZIP() []byte {
+	file_message_proto_rawDescOnce.Do(func() {
+		file_message_proto_rawDescData = protoimpl.X.CompressGZIP(file_message_proto_rawDescData)
+	})
+	return file_message_proto_rawDescData
+}
+
+var file_message_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
+var file_message_proto_goTypes = []interface{}{
+	(*Message)(nil),  // 0: pubsub.v1.Message
+	(*Messages)(nil), // 1: pubsub.v1.Messages
+}
+var file_message_proto_depIdxs = []int32{
+	0, // 0: pubsub.v1.Messages.messages:type_name -> pubsub.v1.Message
+	1, // [1:1] is the sub-list for method output_type
+	1, // [1:1] is the sub-list for method input_type
+	1, // [1:1] is the sub-list for extension type_name
+	1, // [1:1] is the sub-list for extension extendee
+	0, // [0:1] is the sub-list for field type_name
+}
+
+func init() { file_message_proto_init() }
+func file_message_proto_init() {
+	if File_message_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_message_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*Message); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_message_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*Messages); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_message_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   2,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_message_proto_goTypes,
+		DependencyIndexes: file_message_proto_depIdxs,
+		MessageInfos:      file_message_proto_msgTypes,
+	}.Build()
+	File_message_proto = out.File
+	file_message_proto_rawDesc = nil
+	file_message_proto_goTypes = nil
+	file_message_proto_depIdxs = nil
+}