refactor(go): Introduce Command interface to improve Go client request handling

foreign/go/binary_serialization/binary_response_deserializer.go

@@ -545,12 +545,12 @@ func DeserializeClient(payload []byte) *iggcon.ClientInfoDetails {
 		for i := uint32(0); i < clientInfo.ConsumerGroupsCount; i++ {
 			streamId := binary.LittleEndian.Uint32(payload[position : position+4])
 			topicId := binary.LittleEndian.Uint32(payload[position+4 : position+8])
-			consumerGroupId := binary.LittleEndian.Uint32(payload[position+8 : position+12])
+			groupId := binary.LittleEndian.Uint32(payload[position+8 : position+12])
 
 			consumerGroup := iggcon.ConsumerGroupInfo{
-				StreamId:        streamId,
-				TopicId:         topicId,
-				ConsumerGroupId: consumerGroupId,
+				StreamId: streamId,
+				TopicId:  topicId,
+				GroupId:  groupId,
 			}
 			consumerGroups = append(consumerGroups, consumerGroup)
 			position += 12

foreign/go/client/tcp/tcp_access_token_management.go

@@ -23,11 +23,10 @@ import (
 )
 
 func (c *IggyTcpClient) CreatePersonalAccessToken(name string, expiry uint32) (*iggcon.RawPersonalAccessToken, error) {
-	message := binaryserialization.SerializeCreatePersonalAccessToken(iggcon.CreatePersonalAccessTokenRequest{
+	buffer, err := c.do(&iggcon.CreatePersonalAccessToken{
 		Name:   name,
 		Expiry: expiry,
 	})
-	buffer, err := c.sendAndFetchResponse(message, iggcon.CreateAccessTokenCode)
 	if err != nil {
 		return nil, err
 	}
@@ -36,15 +35,14 @@ func (c *IggyTcpClient) CreatePersonalAccessToken(name string, expiry uint32) (*
 }
 
 func (c *IggyTcpClient) DeletePersonalAccessToken(name string) error {
-	message := binaryserialization.SerializeDeletePersonalAccessToken(iggcon.DeletePersonalAccessTokenRequest{
+	_, err := c.do(&iggcon.DeletePersonalAccessToken{
 		Name: name,
 	})
-	_, err := c.sendAndFetchResponse(message, iggcon.DeleteAccessTokenCode)
 	return err
 }
 
 func (c *IggyTcpClient) GetPersonalAccessTokens() ([]iggcon.PersonalAccessTokenInfo, error) {
-	buffer, err := c.sendAndFetchResponse([]byte{}, iggcon.GetAccessTokensCode)
+	buffer, err := c.do(&iggcon.GetPersonalAccessTokens{})
 	if err != nil {
 		return nil, err
 	}

foreign/go/client/tcp/tcp_clients_management.go

@@ -23,7 +23,7 @@ import (
 )
 
 func (c *IggyTcpClient) GetClients() ([]iggcon.ClientInfo, error) {
-	buffer, err := c.sendAndFetchResponse([]byte{}, iggcon.GetClientsCode)
+	buffer, err := c.do(&iggcon.GetClients{})
 	if err != nil {
 		return nil, err
 	}
@@ -32,8 +32,7 @@ func (c *IggyTcpClient) GetClients() ([]iggcon.ClientInfo, error) {
 }
 
 func (c *IggyTcpClient) GetClient(clientId uint32) (*iggcon.ClientInfoDetails, error) {
-	message := binaryserialization.SerializeUint32(clientId)
-	buffer, err := c.sendAndFetchResponse(message, iggcon.GetClientCode)
+	buffer, err := c.do(&iggcon.GetClient{ClientID: clientId})
 	if err != nil {
 		return nil, err
 	}

foreign/go/client/tcp/tcp_consumer_group_management.go

@@ -24,8 +24,10 @@ import (
 )
 
 func (c *IggyTcpClient) GetConsumerGroups(streamId, topicId iggcon.Identifier) ([]iggcon.ConsumerGroup, error) {
-	message := binaryserialization.SerializeIdentifiers(streamId, topicId)
-	buffer, err := c.sendAndFetchResponse(message, iggcon.GetGroupsCode)
+	buffer, err := c.do(&iggcon.GetConsumerGroups{
+		StreamId: streamId,
+		TopicId:  topicId,
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -34,8 +36,13 @@ func (c *IggyTcpClient) GetConsumerGroups(streamId, topicId iggcon.Identifier) (
 }
 
 func (c *IggyTcpClient) GetConsumerGroup(streamId, topicId, groupId iggcon.Identifier) (*iggcon.ConsumerGroupDetails, error) {
-	message := binaryserialization.SerializeIdentifiers(streamId, topicId, groupId)
-	buffer, err := c.sendAndFetchResponse(message, iggcon.GetGroupCode)
+	buffer, err := c.do(&iggcon.GetConsumerGroup{
+		TopicPath: iggcon.TopicPath{
+			StreamId: streamId,
+			TopicId:  topicId,
+		},
+		GroupId: groupId,
+	})
 	if err != nil {
 		return nil, err
 	}
@@ -51,12 +58,13 @@ func (c *IggyTcpClient) CreateConsumerGroup(streamId iggcon.Identifier, topicId
 	if MaxStringLength < len(name) || len(name) == 0 {
 		return nil, ierror.ErrInvalidConsumerGroupName
 	}
-	message := binaryserialization.CreateGroup(iggcon.CreateConsumerGroupRequest{
-		StreamId: streamId,
-		TopicId:  topicId,
-		Name:     name,
+	buffer, err := c.do(&iggcon.CreateConsumerGroup{
+		TopicPath: iggcon.TopicPath{
+			StreamId: streamId,
+			TopicId:  topicId,
+		},
+		Name: name,
 	})
-	buffer, err := c.sendAndFetchResponse(message, iggcon.CreateGroupCode)
 	if err != nil {
 		return nil, err
 	}
@@ -65,19 +73,34 @@ func (c *IggyTcpClient) CreateConsumerGroup(streamId iggcon.Identifier, topicId
 }
 
 func (c *IggyTcpClient) DeleteConsumerGroup(streamId iggcon.Identifier, topicId iggcon.Identifier, groupId iggcon.Identifier) error {
-	message := binaryserialization.SerializeIdentifiers(streamId, topicId, groupId)
-	_, err := c.sendAndFetchResponse(message, iggcon.DeleteGroupCode)
+	_, err := c.do(&iggcon.DeleteConsumerGroup{
+		TopicPath: iggcon.TopicPath{
+			StreamId: streamId,
+			TopicId:  topicId,
+		},
+		GroupId: groupId,
+	})
 	return err
 }
 
 func (c *IggyTcpClient) JoinConsumerGroup(streamId iggcon.Identifier, topicId iggcon.Identifier, groupId iggcon.Identifier) error {
-	message := binaryserialization.SerializeIdentifiers(streamId, topicId, groupId)
-	_, err := c.sendAndFetchResponse(message, iggcon.JoinGroupCode)
+	_, err := c.do(&iggcon.JoinConsumerGroup{
+		TopicPath: iggcon.TopicPath{
+			StreamId: streamId,
+			TopicId:  topicId,
+		},
+		GroupId: groupId,
+	})
 	return err
 }
 
 func (c *IggyTcpClient) LeaveConsumerGroup(streamId iggcon.Identifier, topicId iggcon.Identifier, groupId iggcon.Identifier) error {
-	message := binaryserialization.SerializeIdentifiers(streamId, topicId, groupId)
-	_, err := c.sendAndFetchResponse(message, iggcon.LeaveGroupCode)
+	_, err := c.do(&iggcon.LeaveConsumerGroup{
+		TopicPath: iggcon.TopicPath{
+			StreamId: streamId,
+			TopicId:  topicId,
+		},
+		GroupId: groupId,
+	})
 	return err
 }

foreign/go/client/tcp/tcp_messaging.go

@@ -36,13 +36,13 @@ func (c *IggyTcpClient) SendMessages(
 	if len(messages) == 0 {
 		return ierror.ErrInvalidMessagesCount
 	}
-	serializedRequest := binaryserialization.TcpSendMessagesRequest{
+	_, err := c.do(&iggcon.SendMessages{
+		Compression:  c.MessageCompression,
 		StreamId:     streamId,
 		TopicId:      topicId,
 		Partitioning: partitioning,
 		Messages:     messages,
-	}
-	_, err := c.sendAndFetchResponse(serializedRequest.Serialize(c.MessageCompression), iggcon.SendMessagesCode)
+	})
 	return err
 }
 
@@ -55,16 +55,15 @@ func (c *IggyTcpClient) PollMessages(
 	autoCommit bool,
 	partitionId *uint32,
 ) (*iggcon.PolledMessage, error) {
-	serializedRequest := binaryserialization.TcpFetchMessagesRequest{
+	buffer, err := c.do(&iggcon.PollMessages{
 		StreamId:    streamId,
 		TopicId:     topicId,
 		Consumer:    consumer,
 		AutoCommit:  autoCommit,
 		Strategy:    strategy,
 		Count:       count,
 		PartitionId: partitionId,
-	}
-	buffer, err := c.sendAndFetchResponse(serializedRequest.Serialize(), iggcon.PollMessagesCode)
+	})
 	if err != nil {
 		return nil, err
 	}

foreign/go/client/tcp/tcp_offset_management.go

@@ -23,13 +23,12 @@ import (
 )
 
 func (c *IggyTcpClient) GetConsumerOffset(consumer iggcon.Consumer, streamId iggcon.Identifier, topicId iggcon.Identifier, partitionId *uint32) (*iggcon.ConsumerOffsetInfo, error) {
-	message := binaryserialization.GetOffset(iggcon.GetConsumerOffsetRequest{
+	buffer, err := c.do(&iggcon.GetConsumerOffset{
 		StreamId:    streamId,
 		TopicId:     topicId,
 		Consumer:    consumer,
 		PartitionId: partitionId,
 	})
-	buffer, err := c.sendAndFetchResponse(message, iggcon.GetOffsetCode)
 	if err != nil {
 		return nil, err
 	}
@@ -38,24 +37,22 @@ func (c *IggyTcpClient) GetConsumerOffset(consumer iggcon.Consumer, streamId igg
 }
 
 func (c *IggyTcpClient) StoreConsumerOffset(consumer iggcon.Consumer, streamId iggcon.Identifier, topicId iggcon.Identifier, offset uint64, partitionId *uint32) error {
-	message := binaryserialization.UpdateOffset(iggcon.StoreConsumerOffsetRequest{
+	_, err := c.do(&iggcon.StoreConsumerOffsetRequest{
 		StreamId:    streamId,
 		TopicId:     topicId,
 		Offset:      offset,
 		Consumer:    consumer,
 		PartitionId: partitionId,
 	})
-	_, err := c.sendAndFetchResponse(message, iggcon.StoreOffsetCode)
 	return err
 }
 
 func (c *IggyTcpClient) DeleteConsumerOffset(consumer iggcon.Consumer, streamId iggcon.Identifier, topicId iggcon.Identifier, partitionId *uint32) error {
-	message := binaryserialization.DeleteOffset(iggcon.DeleteConsumerOffsetRequest{
+	_, err := c.do(&iggcon.DeleteConsumerOffset{
 		Consumer:    consumer,
 		StreamId:    streamId,
 		TopicId:     topicId,
 		PartitionId: partitionId,
 	})
-	_, err := c.sendAndFetchResponse(message, iggcon.DeleteConsumerOffsetCode)
 	return err
 }

foreign/go/client/tcp/tcp_partition_management.go

@@ -18,26 +18,23 @@
 package tcp
 
 import (
-	binaryserialization "github.com/apache/iggy/foreign/go/binary_serialization"
 	iggcon "github.com/apache/iggy/foreign/go/contracts"
 )
 
 func (c *IggyTcpClient) CreatePartitions(streamId iggcon.Identifier, topicId iggcon.Identifier, partitionsCount uint32) error {
-	message := binaryserialization.CreatePartitions(iggcon.CreatePartitionsRequest{
+	_, err := c.do(&iggcon.CreatePartitions{
 		StreamId:        streamId,
 		TopicId:         topicId,
 		PartitionsCount: partitionsCount,
 	})
-	_, err := c.sendAndFetchResponse(message, iggcon.CreatePartitionsCode)
 	return err
 }
 
 func (c *IggyTcpClient) DeletePartitions(streamId iggcon.Identifier, topicId iggcon.Identifier, partitionsCount uint32) error {
-	message := binaryserialization.DeletePartitions(iggcon.DeletePartitionsRequest{
+	_, err := c.do(&iggcon.DeletePartitions{
 		StreamId:        streamId,
 		TopicId:         topicId,
 		PartitionsCount: partitionsCount,
 	})
-	_, err := c.sendAndFetchResponse(message, iggcon.DeletePartitionsCode)
 	return err
 }