Abhijeet's Logs
Go back

A gentle introduction to gRPC with Golang

On this page
High level overview

Hello!

I have been wanting to learn gRPC for some time and finally got the time to go through it.

After going through the tutorial example there were few still a few missing pieces I couldn’t wrap my head around. In this post I will try to cover points that were unclear to me. Hopefully it will help you too

Before diving into our example, let’s see what kind of communication can happen via gRPC

  • Unary: Client sends a request, server responds
  • Client Stream: Client streams data to server
  • Server Stream: Server streams data to client
  • Bidirectional Stream: Client and server both can send and receive

Now that we know what type of communication we can do, let’s understand what we will be building.

We will build a restaurant management system, which can do the following things:

  • Customers can get menu to see items, this will demonstrate Unary : GetMenu
  • Customer can place orders, this will demonstrate Client Stream: PlaceOrder
  • Kitchen should be able to see the newly placed orders Server Stream: KitchenSubscribe
  • Lastly customer shares a review and we respond with acknowledgement Bidirectional Stream : Review

Here’s the link to project: https://github.com/iamads/go-workbook/tree/main/restaurant_ordering_system

Defining the contract

To define the API contract for our system we will create a protocol buffer file. We will place it in restaurant/restaurant.proto

In the restaurant.proto

  • We define the syntax we will be using to define our contract: syntax = "proto3";
    • proto3 is the newest version of protocol buffer
    • Compared to proto2 the major improvement in proto3 is that we do not have to mention required for every field. All fields are implicitly optional and default zero value is used in case of absence.
    • For detailed info check this: https://protobuf.dev/programming-guides/proto3/
  • package restaurant: In protocol buffer package, this is used to namespace protobuf definition
  • option go_package: In order to generate Go code, the Go package’s import path must be provided for every .proto file. For our case useoption go_package = "github.com/iamads/go-workbook/restaurant_ordering_system/restaurant";
  • service <Name>: We create a service to encompass all the relevant RPCs. Inside the service we can define the different RPCs.

Defining the Contract

GetMenu - This is the unary RPC. Once the client calls GetMenu, we return them the Menu.

One important thing to note is RPCs will only accept 1 argument and return 1 argument too.

In case of GetMenu we don’t actually need any argument, but since we need 1 argument. We accept Empty (using google.protobuf.Empty)

PlaceOrder - This is the client streaming RPC. Here server accepts a stream of orders from the client.

Once the client is finished streaming in all orders, the server responds with an Order Summary.

KitchenSubscribe - This is the server streaming RPC. Here client accepts a stream of orders from the server. It passes on all the newly placed orders to the kitchen.

Review - This is the bidirectional streaming RPC.

Next we define the messages needed for the RPCs

  • MenuItem: contains name, price of item
  • Menu: array of MenuItems
  • Order: Contains tableNum and orderItems
  • OrderSummary: Contains tableNum, total and array of orderItems
  • ReviewChat: Contains msg and tableNum

We define these messages by composing scalars like string and int32. We also use repeated to define an array.

Other scalars that can be used here are: double, float, int32, bool, bytes etc.

Repeated can be used to define arrays and map can be used to define key/value pairs.

We can also mark if a field is optional or required.

Assigning Field Numbers

We assign field numbers to each of the field. We need to make sure that for a message these field numbers are unique and are not repeated. Once fields have been defined and the application is live, we do not change the field number of any field. We always add new fields and don’t delete. This allows protocol buffers to be backwards compatible.

You can find more information about writing the schema here

// restaurant/restaurant.proto

syntax = "proto3";

package restaurant;

import "google/protobuf/empty.proto";

option go_package = "github.com/iamads/go-workbook/restaurant_ordering_system/restaurant";

service Restaurant {

  rpc GetMenu(google.protobuf.Empty) returns (Menu) {}

  rpc PlaceOrder(stream Order) returns (OrderSummary) {}

  rpc KitchenSubscribe(google.protobuf.Empty) returns (stream Order) {}

  rpc Review(stream ReviewChat) returns (stream ReviewChat) {}
}


message MenuItem {
  string name = 1;
  int32 price = 2;
}

message Menu {
  repeated MenuItem items = 1;
}

message Order {
  int32 tableNum = 1;
  MenuItem orderItem = 2;
}

message OrderSummary {
  int32 tableNum = 1;
  repeated MenuItem orderItems = 2;
  int32 total = 3;
}

message ReviewChat {
  string msg = 1;
  int32 tableNum = 2;
}

Once the .proto file is written we can use protoc to generate the go code.

protoc --go_out=. --go_opt=paths=source_relative \
    --go-grpc_out=. --go-grpc_opt=paths=source_relative \
    restaurant/restaurant.proto

Two files get generated

  • restaurant.pb.go - which contains all the protocol buffer code required to handle messaging.
  • restaurant.grpc.go
    • This contains the stub for the clients to call
    • It also contains the UnimplementedRestaurantServer , which is needed to define a forward compatible server.

Adding the go code

Initial Setup

// server/server.go
package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"sync"

	pb "github.com/iamads/go-workbook/restaurant_ordering_system/restaurant"
	"google.golang.org/grpc"
	"google.golang.org/protobuf/types/known/emptypb"
)

var (
	port = 50051
	ch   = make(chan *pb.Order)
)

type restaurentServer struct {
	pb.UnimplementedRestaurantServer

	mu            sync.Mutex
	tableOrderMap map[int][]*pb.Order // record orders for the table
}



func newServer() *restaurentServer {
	s := &restaurentServer{tableOrderMap: make(map[int][]*pb.Order)}
	return s
}

func main() {
	listener, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", port))
	if err != nil {
		log.Fatalf("failed to listed: %v", err)
	}
	grpcServer := grpc.NewServer() // Creates grpc server
	pb.RegisterRestaurantServer(grpcServer, newServer()) // pb.RegisterRestarurantServer , connects grpc server with the implementation

	log.Println("Starting server")
	err = grpcServer.Serve(listener) // start listening for incoming connections
}
	if err != nil {
		log.Fatalf("Failed to start listening: %v", err)
	}
}

Let’s start building the server

  • In server.go, we will add be doing our setup and adding handlers for the different RPCs we will support
  • restaurantServer is the core struct for this server
    • We embed UnimplementedRestaurantServer from the generated code. This includes information about all the different RPCs and message schema.
    • We add a map and mutex to keep order Info in memory from different tables
  • func newServer() creates a new restaurant server and initialises the map.
  • In the main function
    • We create a tcp listener on localhost on our specified port
    • We create a new grpc server.
    • We register restaurant server , grpc server and restaurant server
    • Finally, grpcServer will serve on the listener

Adding the GetMenu rpc to our server

// server/server.go


// Add this to server.go
func (s *restaurentServer) GetMenu(ctx context.Context, _ *emptypb.Empty) (*pb.Menu, error) {
	items := []*pb.MenuItem{
		&pb.MenuItem{Name: "Idli", Price: 20},
		&pb.MenuItem{Name: "Dosa", Price: 50},
		&pb.MenuItem{Name: "Biryani", Price: 200},
		&pb.MenuItem{Name: "Fried Rice", Price: 150},
	}
	menu := pb.Menu{
		Items: items,
	}
	return &menu, nil
}

Now we want to implement our first rpc: GetMenu which has Unary style of communication.

But how do we add this method? How do we connect our go code with generated Go protocol buffer code

We look into restaurant_grpc.pb.go and in the RestaurantServer interface find this definition GetMenu(context.Context, *emptypb.Empty) (*Menu, error)

We implement this method for our restaurantServer

In the above example from lines 6-15 we define a slice of pointers to MenuItems and then create a menu and finally return it.

Now lets set up the client and perform GetMenu call

package main

import (
	"context"
	"log"

	"google.golang.org/grpc"

	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/protobuf/types/known/emptypb"

	"fmt"
	"time"

	pb "github.com/iamads/go-workbook/restaurant_ordering_system/restaurant"
)

var serverAddress = "localhost:50051"

func printMenu(client pb.RestaurantClient) {
	log.Println("Going to get menu")
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	menu, err := client.GetMenu(ctx, &emptypb.Empty{})
	if err != nil {
		log.Fatalf("client.GetMenu failed: %v", err)
	}
	for _, item := range menu.Items {
		fmt.Printf("%s for Rs %d\n", item.Name, item.Price)
	}
}

func main() {
	conn, err := grpc.NewClient(serverAddress, grpc.WithTransportCredentials(insecure.NewCredentials()))

	if err != nil {
		log.Fatalf("failed to dial: %v", err)
	}

	defer conn.Close()
	client := pb.NewRestaurantClient(conn)

	tableNumber := rand.IntN(10) + 1 // We have 10 tables and will be automatically assigned to customer

	menu, err := printMenu(client)

	if err != nil {
		log.Fatalf("Error in getting the menu: %v", err)
	}

	log.Println("Shutting down client!")
}

The client has two main parts now

Connecting to the server

In the main function

  • We create a gRPC client connection using grpc.NewClient, which takes two params serverAddress and opts (Here I am using insecure which disables security)
  • We pass this conn to generated client to get a RestaurantClient
  • We randomly select a table for the customer.
  • Next we pass the client to printMenu function to do the getMenu call.

Doing the actual GetMenu Call

In the printMenu function

  • GetMenu is called just like a function.
  • We create a context, and pass it to client.GetMenu function.
  • We pass &emptypb.Empty{} to satisfy the argument requirement.
  • After this we can directly iterate on the menuItems we get in response.

Note: We don’t have to do any kind of json encoding/decoding, which typically happens with a REST endpoint.

This was an example of Unary Operation in our system, we do a request and we get the response.

Now lets move onto our first streaming RPC: PlaceOrder

Client Streaming: Placing Order

Now that the customer has been assigned a table and they know what items are available on the menu, they can start placing their order.

The client will connect to the server and place order for all the items they want to eat. Once they send a done signal, the server will return them an OrderSummary.

The interface in restaurant_grpc.pb.go looks like this PlaceOrder(grpc.ClientStreamingServer[Order, OrderSummary]) error

So our PlaceOrder handler takes in grpc.ClientStreamingServer represents server side for client streaming. It is a generic on request and response. In our particular use-case, we have Order as request and OrderSummary as response.

grpc.ClientStreamingServer interface embeds ServerStream and exposes

  • Recv() : this is the function used to receive data from client
  • SendAndClose(): this is used to return some data and close the stream. Typically at the end of stream.

Note: We can use this ServerStream to Context, SetHeader etc.

Place Order: Server Side

// server/server.go

var (
	port = 50051
	ch   = make(chan *pb.Order) // added a chan of orders
)

// GetMenu and other setup code


// to place order for a cetain table
func (s *restaurentServer) PlaceOrder(stream pb.Restaurant_PlaceOrderServer) error {
	var tableNum int32
	for {
		order, err := stream.Recv()

		if err == io.EOF {
			s.mu.Lock()
			allOrders := s.tableOrderMap[int(tableNum)]
			s.mu.Unlock()

			orderItems := []*pb.MenuItem{}
			total := 0

			for _, item := range allOrders {
				orderItems = append(orderItems, item.OrderItem)
				total += int(item.OrderItem.Price)
			}
			summary := pb.OrderSummary{
				TableNum:   tableNum,
				OrderItems: orderItems,
				Total:      int32(total),
			}
			return stream.SendAndClose(&summary)
		}

		if err != nil {
			return err
		}

		if tableNum == 0 {
			tableNum = order.TableNum
		}

		s.mu.Lock()
		if _, ok := s.tableOrderMap[int(order.TableNum)]; ok {
			s.tableOrderMap[int(order.TableNum)] = append(s.tableOrderMap[int(order.TableNum)], order)
		} else {
			s.tableOrderMap[int(order.TableNum)] = []*pb.Order{order}
		}
		s.mu.Unlock()
		
		ch <- order // send the order to kitchen
	}
}

On line 12, we have added a PlaceOrder method which stream of type Restaurant_PlaceOrderServer (This is the server stream we talked about just moment’s before)

The main things happening here

  • Listening to incoming Order messages
  • In case we get an EOF error
  • In case we get some other error
  • Process the Order

Let’s do a dry run of

First incoming Order from customer

  • We receive order using stream.Recv()
  • There is no error, we see tableNum is 0, so we change it to what we get from msg
  • Next, we see if we have an entry in orderMap, since it is first order we will have to add a new entry and then add the order Item
  • Then we pass on the order to Kitchen

Any further orders for this customers will just be appended.

When the client is done

  • When the client is done we get an EOF error.
  • Here we can collect the entire summary of all our orders, compute bill etc.
  • Then send OrderSummary to client using SendAndClose.

Place Order: Client Side



// client/customer/customer.go

// This initiates customers ordering flow
// Customer can stream their orders
// Once they are done they receive a bill
func customerOrder(client pb.RestaurantClient, menu *pb.Menu, tableNumber int) {
	log.Println("Pls place as many orders as you want!")
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	stream, err := client.PlaceOrder(ctx)
	if err != nil {
		log.Fatalf("client.PlaceOrder failed: %v", err)
	}
	myOrders := pickRandomOrders(menu, tableNumber) // A helper method I have added to pick random order

	for _, order := range myOrders {
		if err := stream.Send(order); err != nil {
			log.Fatalf("client.PlaceOrder failed at stream.send for order(%v) with error: %v", order, err)
		}
		time.Sleep(500 * time.Millisecond)
	}
	summary, err := stream.CloseAndRecv()
	if err != nil {
		log.Fatalf("client.Place Order failed at stream close: %v", err)
	}
	log.Printf("Total bill for Table %d is %d \n", summary.TableNum, summary.Total)
}

// just an helper method
func pickRandomOrders(menu *pb.Menu, tableNumber int) []*pb.Order {
	order := []*pb.Order{}

	// random number of items to be ordered

	itemCount := rand.IntN(7) + 1

	for i := 0; i < itemCount; i++ {
		order = append(order, &pb.Order{
			TableNum:  int32(tableNumber),
			OrderItem: menu.Items[rand.IntN(len(menu.Items))],
		})
	}
	return order
}

Compared to the server side, client call is simple.

  • We call client.PlaceOrder to create a stream.
  • We push events to server by using stream.Send
  • Once we are done, we call stream.CloseAndRecv(). This is what will cause the EOF error at the server and is ready to receive the final response from the server.

Server streaming: Kitchen Subscribe

To understand Server Streaming we will look at what’s happening in the Kitchen as example.

In the Kitchen, whatever food gets ordered the cook makes it. In the workflow kitchen as an entity does not care about order placement and customer. It is listening to stream of events from the server and it may (or may not) take actions on the incoming messages.

Here the Kitchen subscribes to the new order stream from the server.

Here’s the RPC signature

KitchenSubscribe(*emptypb.Empty, grpc.ServerStreamingServer[Order]) error

Server Side of Kitchen Subscribe


// server/server.go

// ...

// Newly placed order will be streamed to client
func (s *restaurentServer) KitchenSubscribe(_ *emptypb.Empty, stream pb.Restaurant_KitchenSubscribeServer) error {
	for order := range ch {
		if err := stream.Send(order); err != nil {
			return err
		}
	}

	return nil
}

This is super straight forward

  • We get order from channel ch (Remember when placing orders we were pushing order to this channel at the very end)
  • And then stream it to caller

Client Side of Kitchen Subscribe


package main

import (
	"context"
	"io"
	"log"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/protobuf/types/known/emptypb"

	"time"

	pb "github.com/iamads/go-workbook/restaurant_ordering_system/restaurant"
)

var serverAddress = "localhost:50051"

func printOrders(client pb.RestaurantClient) {
	log.Printf("Will stream in new orders as soon as they have been placed")
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Hour) // Change to deadline
	defer cancel()

	stream, err := client.KitchenSubscribe(ctx, &emptypb.Empty{}) 

	if err != nil {
		log.Fatalf("client.KitchenSubscribe failed: %v", err)
	}

	for {
		order, err := stream.Recv()
		if err == io.EOF {
			break
		}

		if err != nil {
			log.Fatalf("client.Kitchen subscribe failed while receiving: %v", err)
		}
		log.Printf("New Order received: %s, for table %d", order.OrderItem.Name, order.TableNum)
	}
}

func main() {
	conn, err := grpc.NewClient(serverAddress, grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatalf("failed to dial: %v", err)
	}
	defer conn.Close()
	client := pb.NewRestaurantClient(conn)

	printOrders(client)
}

I would like you to focus on printOrders function

  • We start by creating a context with relevant deadline/timeout(closing time)
  • Call kitchen subscribe rpc with the context to get the stream
  • Do the usual go err check
  • Now the main part, inside a for loop call stream.Recv() to get the order or err
  • We follow the same pattern as we saw before, if error is EOF that means it has been disconnected correctly and break out of the loop.
  • On any other error we log and exit.
  • If everything is cool, we have received a new Order to make, we log it out.

Note: since we are using Unbuffered channel, while you are running the code it may get stuck if you don’t run the client/kitchen/kitchen.go.

Onwards

Bidirectional Streaming: Review Chat

We use this to allow the customer to review the food and the restaurant can acknowledge their feedback.

Here’s the RPC signature

Review(grpc.BidiStreamingServer[ReviewChat, ReviewChat]) error

Notice that we are now using BidiStreamingServer (Bidirectional) with ReviewChat as both the request and Response.

The BidiStreamingClient interface exposes Send and Recv. It embeds the ClientStreaminterface. Similarly BidiStreamingServer also exposes Send and Recv methods, it embeds the ServerStream interface.

Let’s look into the server code


// server/server.go


// We create a bidirectional stream for collecting Review
// customer sends their review
// we thank them for response and clear out their table related info from orderMap
func (s *restaurentServer) Review(stream pb.Restaurant_ReviewServer) error {
	for {
		in, err := stream.Recv()

		if err == io.EOF {
			return nil
		}

		if err != nil {
			return err
		}

		log.Println("Customer says: ", in.Msg)

		if err := stream.Send(&pb.ReviewChat{Msg: "Thanks for sharing your experience!"}); err != nil {
			return err
		} else {
			s.mu.Lock()
			delete(s.tableOrderMap, int(in.TableNum))
			s.mu.Unlock()
		}
	}
}

The flow is very similar to other streaming examples we have already seen.

  • In the for loop we use stream.Recv()
  • In case, we get an EOF error that means the connection has been closed by other side.
  • In case of no error, we have successfully received a msg from server.
  • For our example scenario, we thank them for sharing their experience (Bidirectional because received client’s review and sent review.)
  • If no error, we delete clients order and mark that table as free. (line 25)

Let’s move on to the client side of this


// We use this function to review food
// This is a bidirectional rpc
// In this implementation customer sends a review and receives acknowlegment
// Note: Msg ordering is independent of client and server roles
func customerReview(client pb.RestaurantClient, tableNum int) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	stream, err := client.Review(ctx)
	if err != nil {
		log.Fatalf("client.Review failed: %v", err)
	}

	wg := sync.WaitGroup{}

	wg.Add(1)
	go func() {
		defer wg.Done()

		for {
			in, err := stream.Recv()
			if err == io.EOF {
				return
			}
			if err != nil {
				log.Fatalf("client.customerReview failed: %v", err)
			}
			log.Printf("Msg to customer: %s \n", in.Msg)
			if err = stream.CloseSend(); err != nil {
				log.Fatalf("Could not close stream after receving msg to customer: %v", err)
			}
		}
	}()

	review := getRandomReview()
	if err := stream.Send(&review); err != nil {
		log.Fatalf("client.Review: stream.Send(%v) failed: %v", review.Msg, err)
	}
	wg.Wait()
}

func getRandomReview() pb.ReviewChat {
	msgs := []string{"good!", "ok", "great!", "bad", "it was really bad"}
	selectedMsg := msgs[rand.IntN(len(msgs))]
	return pb.ReviewChat{Msg: selectedMsg}
}

The Client Review follows the following flow

  • We create the context with relevant timeout, the pass that on to review rpc call.
  • From the review rpc call we get our bidirectional stream.
  • Since we have to both listen and send messages we spin up a separate goroutine to receive message.
  • From the current goroutine, we select a random review and send it to our server.
  • In the new goroutine, we wait to recv msgs from server. Here we wait till we wait till we recv a message, the stream is disconnected(In case of EOF) or an error happens.

Here this demonstrates how a Bidirectional streaming will work. Although not a great example, I hope it gives you an idea how Bidirectional streaming might work.

Conclusion

I hope this blog post was helpful in understanding gRPC and how to use it with Golang. Pls feel free to share your thoughts in the comments or email.

I am planning to start working on my database project. Here are the previous blog posts on those if you are interested:

Share this post:
Share this post on Linkedin Share this post on Whatsapp Share this post on Facebook Share this post on X Share this post on Telegram Share this post on Pinterest Share this post via email
Next Post
Just concurrency will not make your go code faster