rpc: Support binary request messages

This feature will be useful for possible optimizations of protocol
communication.

Signed-off-by: Leonard Lyubich <[email protected]>

rpc/client/call_options.go

@@ -9,6 +9,8 @@ type CallOption func(*callParameters)
 
 type callParameters struct {
 	ctx context.Context
+
+	allowBinarySendingOnly bool
 }
 
 func defaultCallParameters() *callParameters {
@@ -27,3 +29,15 @@ func WithContext(ctx context.Context) CallOption {
 		prm.ctx = ctx
 	}
 }
+
+// AllowBinarySendingOnly allows only [MessageWriter.WriteMessage] method's
+// arguments that are convertible to binary gRPC messages ([]byte). For example,
+// [BinaryMessage] may be used for such write. By default, only arguments
+// convertible to [proto.Message] may be used. Use this option when binary
+// message transmission is needed. Note that only [proto.Message] convertible
+// response messages are supported even with this option.
+func AllowBinarySendingOnly() CallOption {
+	return func(prm *callParameters) {
+		prm.allowBinarySendingOnly = true
+	}
+}

rpc/client/init.go

@@ -1,6 +1,7 @@
 package client
 
 import (
+	"fmt"
 	"io"
 
 	"github.com/nspcc-dev/neofs-api-go/v2/rpc/common"
@@ -67,12 +68,39 @@ func (g rwGRPC) WriteMessage(m message.Message) error {
 	return g.MessageReadWriter.WriteMessage(m.ToGRPCMessage())
 }
 
+// BinaryMessage represents binary [message.Message] that can be used with
+// [AllowBinarySendingOnly] option.
+type BinaryMessage []byte
+
+func (x BinaryMessage) ToGRPCMessage() grpc.Message {
+	return []byte(x)
+}
+
+func (x BinaryMessage) FromGRPCMessage(m grpc.Message) error {
+	bMsg, ok := m.([]byte)
+	if ok {
+		copy(x, bMsg)
+		return nil
+	}
+
+	return fmt.Errorf("message is not of type %T", bMsg)
+}
+
 func (c *Client) initGRPC(info common.CallMethodInfo, prm *callParameters) (MessageReadWriter, error) {
 	if err := c.createGRPCClient(prm.ctx); err != nil {
 		return nil, err
 	}
 
-	rw, err := c.gRPCClient.Init(info, grpc.WithContext(prm.ctx))
+	var grpcCallOpts []grpc.CallOption
+	ctxCallOpt := grpc.WithContext(prm.ctx)
+
+	if prm.allowBinarySendingOnly {
+		grpcCallOpts = []grpc.CallOption{ctxCallOpt, grpc.AllowBinarySendingOnly()}
+	} else {
+		grpcCallOpts = []grpc.CallOption{ctxCallOpt}
+	}
+
+	rw, err := c.gRPCClient.Init(info, grpcCallOpts...)
 	if err != nil {
 		return nil, err
 	}

rpc/grpc/call_options.go

@@ -9,6 +9,8 @@ type CallOption func(*callParameters)
 
 type callParameters struct {
 	ctx context.Context
+
+	allowBinarySendingOnly bool
 }
 
 func defaultCallParameters() *callParameters {
@@ -23,3 +25,14 @@ func WithContext(ctx context.Context) CallOption {
 		prm.ctx = ctx
 	}
 }
+
+// AllowBinarySendingOnly allows to pass []byte argument only to
+// [MessageReadWriter.WriteMessage] method. By default, only [proto.Message]
+// instances may be used. Use this option when binary message transmission is
+// needed. Note that only [proto.Message] response messages are supported even
+// with this option.
+func AllowBinarySendingOnly() CallOption {
+	return func(prm *callParameters) {
+		prm.allowBinarySendingOnly = true
+	}
+}

rpc/grpc/init.go

@@ -2,11 +2,14 @@ package grpc
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"time"
 
 	"github.com/nspcc-dev/neofs-api-go/v2/rpc/common"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/encoding"
+	"google.golang.org/grpc/encoding/proto"
 )
 
 // Message represents raw gRPC message.
@@ -69,6 +72,26 @@ func (w *streamWrapper) withTimeout(closure func() error) error {
 	}
 }
 
+type onlyBinarySendingCodec struct{}
+
+func (x onlyBinarySendingCodec) Name() string {
+	// may be any non-empty, conflicts are unlikely to arise
+	return "neofs_binary_sender"
+}
+
+func (x onlyBinarySendingCodec) Marshal(msg interface{}) ([]byte, error) {
+	bMsg, ok := msg.([]byte)
+	if ok {
+		return bMsg, nil
+	}
+
+	return nil, fmt.Errorf("message is not of type %T", bMsg)
+}
+
+func (x onlyBinarySendingCodec) Unmarshal(raw []byte, msg interface{}) error {
+	return encoding.GetCodec(proto.Name).Unmarshal(raw, msg)
+}
+
 // Init initiates a messaging session within the RPC configured by options.
 func (c *Client) Init(info common.CallMethodInfo, opts ...CallOption) (MessageReadWriter, error) {
 	prm := defaultCallParameters()
@@ -77,12 +100,17 @@ func (c *Client) Init(info common.CallMethodInfo, opts ...CallOption) (MessageRe
 		opt(prm)
 	}
 
+	var grpcCallOpts []grpc.CallOption
+	if prm.allowBinarySendingOnly {
+		grpcCallOpts = []grpc.CallOption{grpc.ForceCodec(onlyBinarySendingCodec{})}
+	}
+
 	ctx, cancel := context.WithCancel(prm.ctx)
 	stream, err := c.con.NewStream(ctx, &grpc.StreamDesc{
 		StreamName:    info.Name,
 		ServerStreams: info.ServerStream(),
 		ClientStreams: info.ClientStream(),
-	}, toMethodName(info))
+	}, toMethodName(info), grpcCallOpts...)
 	if err != nil {
 		cancel()
 		return nil, err