Hi everyone,

In this post, I am going to give an advanced example on a topic that is quite popular these days: gRPC.
gRPC is a modern open source high performance RPC framework that can run in any environment. It can efficiently connect services in and across data centers with pluggable support for load balancing, tracing, health checking and authentication. It is also applicable in last mile of distributed computing to connect devices, mobile applications and browsers to backend services.[grpc.io]
The following main.proto describes the service Main that uses bi-directional streaming Remote Procedure Call communication and carries raw bytes as the message type.

syntax = "proto3";

package rpc;

service Main {
  rpc RPC (stream Payload) returns (stream Payload) {}
}

message Payload {
  bytes Data = 1;
}

After creating the proto file, we need to generate the source code for the server and the client using this file:

 protoc -I main/ main.proto --go_out=plugins=grpc:main

This command will generate the necessary source code that will be used in your project.

The next step is to create a Payload struct that will carry message between server-client pairs.

package rpc

// Payload defines the service message format.
type Payload struct {
 data []byte
}

// NewPayload creates and returns a new payload with the given byte slice.
func NewPayload(d []byte) *Payload {
 p := new(Payload)
 if d != nil {
  p.Set(d)
 }
 return p
}

// Set sets the payload
func (p *Payload) Set(d []byte) {
 p.data = d
}

// Get returns the payload
func (p *Payload) Get() []byte {
 return p.data
}

Now we have reached to a point where we need to add a custom codec to the gRPC library, so that it does not attempt to serialize & deserialize the bytes we are sending over the wire. The main reason behind this is to make the communication faster, since we do not know the size of the stream between the gRPC server and the client.

package rpc

import (
 "errors"
)

// PayloadCodec allows us to alter Marshal&Unmarshal function behaviours
type PayloadCodec struct{}

// Marshal is used to convert data to wire format
// Since the message format is already []byte, no need to Marshal,
// we can directly stream data
func (cb *PayloadCodec) Marshal(v interface{}) ([]byte, error) {
 p, ok := v.(*Payload)
 if !ok {
  err := errors.New("err: Invalid type of struct")
  return nil, err
 }
 return p.Get(), nil
}

// Unmarshal is used to convert data from wire format
// Since the message format is already []byte, no need to Marshal,
// we can directly stream data
func (cb *PayloadCodec) Unmarshal(data []byte, v interface{}) error {
 p, ok := v.(*Payload)
 if !ok {
  err := errors.New("err: Invalid type of struct")
  return err
 }
 p.Set(data)
 return nil
}

func (cb *PayloadCodec) String() string {
 return "Custom Marshal-Unmarshal Codec"
}

Since we have a codec we can use now, this is how we instruct the gRPC server and the client to use this codec (The server and client portions are left out in this article, since the focus is to show how to use gRPC bi-directional streaming without protocol buffers.) :

Server:

grpc.NewServer(grpc.CustomCodec(&rpc.PayloadCodec{}))
rpc.RegisterMainServer(s.grpcServerHandle, s)

Client:

conn, err := grpc.Dial(addr, grpc.WithInsecure, grpc.WithCodec(&rpc.PayloadCodec{}))

And that allows disabling the built-in protocol buffers, and allows the use of a custom serializer-deserializer in your gRPC server-client pair. Please leave a comment if you have any comments or questions.