gRPC Quick start - Coding with streams and bidirectional streaming

André Freitas — Thu, 13 Jun 2024 23:51:43 +0000

Perhaps you've already heard about gRPC. A few years ago, just the mention of "gRPC" gave me goosebumps. I used to think it was the most complex monster created by the world of microservices. But it's not as scary as I once thought (really far from it).

Let's see a quick start coding guide using Go, Evans, and gRPC

Attention!

This guide step isn't for you if:

You do not know Go
You're a beginner programmer
You like Star Wars The Last Jedi

If you do not fit into any of the previous cases, keep going

What is gRPC?

In a few words, gRPC is a modern framework created by Google (now maintained by Cloud Native Computing) that implements Remote Procedure Call (RPC) that can run in any environment with high performance. It uses protocol buffers (a simplified XML) through HTTP/2.

Thus, with gRPC we got binary files smaller than JSON, lower network consumption, and faster data transmission between services.

Notice that the main concept of gRPC is communication between microservices, not between browsers x backend.

I recommend that you read the official gRPC website. There is a lot of good content and documentation, I'll not be wordy cause already exists better references

Let's Start

In this article I'll use Go and Docker. So, You'll have to install the gRPC plugins. Follow up on this tutorial: Tutorial

But... What will we create?

Let's suppose antennas spread across a land running a service that calculates and returns the distance in kilometers between the antenna and the client.

Then, an object, user, person, extraterrestrial, or anything else wanna contact an antenna to track a register from your moving position using latitude and longitude and receive a response from the antenna (remember that this is an example).

Now that our domain is set up, in our workspace environment, create a folder called "proto". Here we gonna define the contracts of our gRPC service using proto buffer.

Create a file called "registers.proto" inside the proto folder. First, we gonna define the settings from proto (using proto3) and then, create an object that will represent our Register object.

syntax = "proto3";
package pb;
option go_package = "internal/pb";

message Register {
    string id = 1;
    string latitude = 2;
    string longitude = 3;
    string distance = 4;
}

Those numbers represent only the order they should be sent. With this message contract, we defined that Register must be created with an id, latitude, longitude, and distance attributes.

With the proto file defined, run: protoc --go_out=. --go-grpc_out=. proto/register.proto. This command is used to generate Go source files from a Protocol Buffers Documentation here.

Great! Now we have our base object. But supposing we are the extraterrestrial contact, what should be inside of our request to the antenna? Well, our latitude and longitude, right?

Thus:

message CreateRequest {
    string latitude = 1;
    string longitude = 2;
}

Here we defined our request. Quite different from REST.

Finally, let's define the register service on proto

service RegisterService {
    rpc CreateRegister(CreateRequest) returns (Register){}

}

Here we define our service RegisterService with the rpc method CreateRegister receiving a CreateRequest and returning Register

Calm down, don't worry, we're almost there!

create a folder called service and inside, create registers.go file. Here, we have to define the register service

type RegisterService struct {
    pb.UnimplementedRegisterServiceServer
}

func NewRegisterService() *RegisterService {
    return &RegisterService{}
}

var latitude = -5.8623992555733695
var longitude = -35.19111877919574

As we don't have a database or anything else, let's create just with pb.UnimplementedRegisterServiceServer. Notice that I defined global variables from latitude and longitude. It will represent the localization from a specific antenna. You're free to decide your latitude and longitude. I decided to set the coordinates from my favorite restaurant :)

Finally, let's create our server. I'll create a folder cmd containing another folder called grpc that contains the main.go file.

package main

import (
    "andrefsilveira/grpc-quick-start/internal/pb"
    "andrefsilveira/grpc-quick-start/service"
    "fmt"
    "net"

    "google.golang.org/grpc"
    "google.golang.org/grpc/reflection"
)

func main() {

    registerService := service.NewRegisterService()
    grpcServer := grpc.NewServer()
    pb.RegisterRegisterServiceServer(grpcServer, registerService)
    reflection.Register(grpcServer)

    var port = ":50051"
    listen, err := net.Listen("tcp", port)
    if err != nil {
        panic(err)
    }
    fmt.Printf("Server starting on port %s", port)
    if err := grpcServer.Serve(listen); err != nil {
        panic(err)
    }
}

At main.go we are instantiating the register service and also creating a new grpc server. Again, the details are on the documentation. Remember to import all of the necessary packages.

But, are we forgetting something?

Yes, we are. Look, we create the proto file, generate the files from proto buffers, and create the connection and the constructors. But, we have to do something with this request. A few months ago I created a go package that calculates the distance between two coordinates and returns the distance in kilometers. You can read about it here. So, let's use it to calculate our distance from an antenna.

at service/registers.go, create:

func (c *RegisterService) CreateRegister(ctx context.Context, in *pb.CreateRequest) (*pb.Register, error) {
    register := &pb.Register{}

    lat, err := strconv.ParseFloat(in.Latitude, 64)
    if err != nil {
        return nil, err
    }

    lon, err := strconv.ParseFloat(in.Longitude, 64)
    if err != nil {
        return nil, err
    }

    result := haversine.Calculate(latitude, longitude, lat, lon)
    value := strconv.FormatFloat(result, 'f', -1, 64)

    register = &pb.Register{
        Id:        uuid.New().String(),
        Latitude:  register.Latitude,
        Longitude: register.Longitude,
        Distance:  value,
    }

    return register, nil

}

Here, we are receiving the register from input and parsing to float to finally use it in the haversine package. Then, a Register is returned.

Notice that the method name should match the method name declared in the proto file:

service RegisterService {
    rpc CreateRegister(CreateRequest) returns (Register){}

}

So far so good, let's do it!

run: go run cmd/grpc/main.go

You should see the Server starting on port :50051 message. To interact with the gRPC server, we will use Evans, which is an interactive command-line client for gRPC. You can read more about it here and can install using docker docker run --rm --network host -it ghcr.io/ktr0731/evans --host localhost --port 50051 --reflection. This method is not recommended, but in our case, it will be fine. If everything worked correctly, you must see something like this in your terminal:

first, we have to select the package and the service. Remember that we defined at proto file?

syntax = "proto3";
package pb;
option go_package = "internal/pb";

[...]

service RegisterService {
    rpc CreateRegister(CreateRequest) returns (Register){}
}

The package is pband the service is RegisterService. Select them using respectively:

package pb
service RegisterService

Now, we can call the create register method. Run call CreateRegister and type any latitude and longitude. If everything works fine, this should be your output:

pb.RegisterService@localhost:50051> call CreateRegister
latitude (TYPE_STRING) => 150
longitude (TYPE_STRING) => 600
{
  "distance": "10831.805049673263",
  "id": "5421f456-9b77-495e-b358-1184fcc8dc3b"
}

But, let's suppose a scenario that we do not have to receive the response after the request. Take the same example, and let be the "tracking" a non vital information. Thus, we could receive the response only when our connection was closed.

To do this, we gonna create a stream. Streams allow the client or the server to send multiple requests over a unique connection. But keep calm it is very simple.

Create another message called Registers. Notice that it is the same Register but with the tag repeated. This means that we are gonna receive a list of Register messages.

message Registers {
    repeated Register regiters = 1;
}

Then, add the method in RegisterService

service RegisterService {
    rpc CreateRegister(CreateRequest) returns (Register){}
    rpc CreateRegisterStream(stream CreateRequest) returns (Registers) {}

}

And finally, create the CreateRegisterStream method:

func (c *RegisterService) CreateRegisterStream(stream pb.RegisterService_CreateRegisterStreamServer) error {
    registers := &pb.Registers{}

    for {
        register, err := stream.Recv()
        if err == io.EOF {
            return stream.SendAndClose(registers)
        }

        if err != nil {
            return err
        }

        lat, err := strconv.ParseFloat(register.Latitude, 64)
        if err != nil {
            return err
        }

        lon, err := strconv.ParseFloat(register.Longitude, 64)
        if err != nil {
            return err
        }

        result := haversine.Calculate(latitude, longitude, lat, lon)
        value := strconv.FormatFloat(result, 'f', -1, 64)

        registers.Registers = append(registers.Registers, &pb.Register{
            Id:        uuid.New().String(),
            Latitude:  register.Latitude,
            Longitude: register.Longitude,
            Distance:  value,
        })

    }
}

Notice that this method is quite different from the previous one. It's opening a stream with register, err := stream.Recv() and receiving values inside a for loop. This loop will exit only if an error occurs or it reaches the end of file io.EOF. Great. Save it and restart Evans and your gRPC server.

package pb
service RegisterService
call CreateRegisterStream

Type some coordinates. When you finish, type: cntrl + d, and you should receive something like this:

pb.RegisterService@localhost:50051> call CreateRegisterStream
latitude (TYPE_STRING) => 156
longitude (TYPE_STRING) => 616
latitude (TYPE_STRING) => 616
longitude (TYPE_STRING) => 99
latitude (TYPE_STRING) => 864
longitude (TYPE_STRING) => 151
latitude (TYPE_STRING) => 
{
  "registers": [
    {
      "distance": "12422.520903261275",
      "id": "f083993e-2c73-438e-a906-d1b19b26476f",
      "latitude": "156",
      "longitude": "616"
    },
    {
      "distance": "8286.547137287573",
      "id": "8ca6bc74-af0d-4c38-af54-b1f36bbdecf0",
      "latitude": "616",
      "longitude": "99"
    },
    {
      "distance": "4699.522659470381",
      "id": "af354a5f-91df-4573-b5ad-5fb736187942",
      "latitude": "864",
      "longitude": "151"
    }
  ]
}

Nice!

Moreover, let's suppose another case. Assume that now it is necessary a continuous communication between the user and the antenna. Both must keep in touch with each other, and the information should be transmitted through a binary connection client <-> server.

In this case, we can use bidirectional streams. This type of communication is related to a two-way conversation where both parties can continuously send and receive messages.

Similarly to the previous method, let's create the CreateRegisterBidirectional method:

service RegisterService {
    rpc CreateRegister(CreateRequest) returns (Register){}
    rpc CreateRegisterStream(stream CreateRequest) returns (Registers) {}
    rpc CreateRegisterBidirectional(stream CreateRequest) returns (stream Register) {}
}

Notice that we now put the stream tag in the return statement too. When it has a stream at input and output, it defines a bidirectional method.

Thus, let's create the method:

func (c *RegisterService) CreateRegisterBidirectional(stream pb.RegisterService_CreateRegisterBidirectionalServer) error {
    for {
        register, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }

        lat, err := strconv.ParseFloat(register.Latitude, 64)
        if err != nil {
            return err
        }

        lon, err := strconv.ParseFloat(register.Longitude, 64)
        if err != nil {
            return err
        }

        result := haversine.Calculate(latitude, longitude, lat, lon)
        value := strconv.FormatFloat(result, 'f', -1, 64)

        err = stream.Send(&pb.Register{
            Id:        uuid.New().String(),
            Latitude:  register.Latitude,
            Longitude: register.Longitude,
            Distance:  value,
        })

        if err != nil {
            return err
        }
    }
}

Notice that looks like with the previous method. But instead send and closing only when the connection close, now we gonna receive and send simultaneously. Restart Evans and the gRPC server again.

package pb
service RegisterService
call CreateRegisterBidirectional

Type how many coordinates you want. The output should look like:

pb.RegisterService@localhost:50051> call CreateRegisterBidirectional
latitude (TYPE_STRING) => 161
longitude (TYPE_STRING) => 66
latitude (TYPE_STRING) => {
  "distance": "9052.814145827013",
  "id": "197ed428-3a18-4548-888a-9d2ca4930995",
  "latitude": "161",
  "longitude": "66"
}
latitude (TYPE_STRING) => 6489
longitude (TYPE_STRING) => 116
latitude (TYPE_STRING) => {
  "distance": "16820.47617639521",
  "id": "d356e97f-2a1d-491a-b2ef-37e88bfafb59",
  "latitude": "6489",
  "longitude": "116"
}
latitude (TYPE_STRING) => 616
longitude (TYPE_STRING) => 888
latitude (TYPE_STRING) => {
  "distance": "7930.192513515994",
  "id": "ba561afe-3a92-4f19-8a14-9b6c7ae3de26",
  "latitude": "616",
  "longitude": "888"
}

Notice that now we are sending and receiving information at the same time. To stop it, type cntrl + d.

Uff, so this is it. If you have any doubts or recommendations do not hesitate to reach out at: freitasandre38@gmail.com

Also, you can find the source code here

DEV Community: André Freitas

gRPC Quick start - Coding with streams and bidirectional streaming

What is gRPC?

Let's Start