Implement bidirectional streaming gRPC - Go
Tags: go, tutorial, microservices, beginners
Published: 2020-05-04 20:01:05

Hi everyone!

Today we’re gonna learn how to implement the last type of gRPC: **bidirectional streaming**, or **bidi-streaming**. This streaming allows client and server to send multiple requests and multiple responses to each other in parallel.

Here's the link to the [full gRPC course playlist on Youtube](https://www.youtube.com/playlist?list=PLy_6D98if3UJd5hxWNfAqKMr15HZqFnqf)
Github repository: [pcbook-go](https://github.com/techschool/pcbook-go) and [pcbook-java](https://github.com/techschool/pcbook-java)
Gitlab repository: [pcbook-go](https://gitlab.com/techschool/pcbook) and [pcbook-java](https://gitlab.com/techschool/pcbook-java)

{% youtube hjTI35iKMyQ %}

In this lecture, we will write an API for client to rate a stream of laptops with score from 1 to 10, and the server will respond with a stream of average scores for each of the laptops.

Alright, let’s start!

## 1. Define bidi-streaming gRPC protobuf

The first thing we need to do is to define a new bidi-streaming RPC in the `laptop_service.proto` file.

We define the `RateLaptopRequest` with 2 fields: the laptop ID and the score.

```protobuf
message RateLaptopRequest {
  string laptop_id = 1;
  double score = 2;
}
```

Then the `RateLaptopResponse` with 3 fields: the laptop ID, the number of time this laptop was rated, and the average rated score.

```protobuf
message RateLaptopResponse {
  string laptop_id = 1;
  uint32 rated_count = 2;
  double average_score = 3;
}
```

Now we define the `RateLaptop` RPC with input is a stream of `RateLaptopRequest`, and output is a stream of `RateLaptopResponse`.

```protobuf
service LaptopService {
  ...
  rpc RateLaptop(stream RateLaptopRequest) returns (stream RateLaptopResponse) {};
}
```

After we run `make gen` in the terminal to regenerate the codes, we can see an error inside the `server/main.go` file. This is because the `LaptopServiceServer` interface now requires 1 more method: `RateLaptop`, which should be implemented by the `LaptopServer` struct. We can find the signature of this method inside the `pb/laptop_service.pb.go` file. 

```go
type LaptopServiceServer interface {
	CreateLaptop(context.Context, *CreateLaptopRequest) (*CreateLaptopResponse, error)
	SearchLaptop(*SearchLaptopRequest, LaptopService_SearchLaptopServer) error
	UploadImage(LaptopService_UploadImageServer) error
	RateLaptop(LaptopService_RateLaptopServer) error
}
```

So let’s copy it, and paste in the `laptop_server.go` file. Let's just return `nil` for now. We will come back to implement this method later.

```go
func (server *LaptopServer) RateLaptop(stream pb.LaptopService_RateLaptopServer) error {
    return nil
}
```

## 2. Implement the rating store

Now we need to create a new rating store to save the laptop ratings.

I will define a `RatingStore` interface. It has 1 function `Add` that takes a laptop ID and a score as input, and returns the updated laptop rating or an error.

```go
type RatingStore interface {
	Add(laptopID string, score float64) (*Rating, error)
}
```

The rating consists of 2 fields: one is count, which is the number of times the laptop is rated, and the other is the sum of all rated scores.

```go
type Rating struct {
	Count uint32
	Sum   float64
}
```

Then we will write an in-memory rating store that implements the interface. Similar to the in-memory laptop store, here we will need a mutex to handle concurrent access. And we have a rating map with key is the laptop ID, and value is the rating object.

```go
type InMemoryRatingStore struct {
	mutex  sync.RWMutex
	rating map[string]*Rating
}
```

Then we define a function to create a new in-memory rating store. In this function, we just need to initialize the rating map.

```go
func NewInMemoryRatingStore() *InMemoryRatingStore {
	return &InMemoryRatingStore{
		rating: make(map[string]*Rating),
	}
}
```

OK, now let’s implement the `Add` function! 

As we’re going to change the internal data of the store, we have to acquire a lock. Then we get the rating of the laptop ID from the map. 

If the rating is not found, we just create a new object with count is 1 and sum is the input score. Else, we increase the rating count by 1 and add the score to the sum.

```go
func (store *InMemoryRatingStore) Add(laptopID string, score float64) (*Rating, error) {
	store.mutex.Lock()
	defer store.mutex.Unlock()

	rating := store.rating[laptopID]
	if rating == nil {
		rating = &Rating{
			Count: 1,
			Sum:   score,
		}
	} else {
		rating.Count++
		rating.Sum += score
	}

	store.rating[laptopID] = rating
	return rating, nil
}
```

Finally we put the updated rating back to the map and return it to the caller. 

Then we’re done with the store. Now let’s go back to implement the server.

## 3. Implement the bidi-streaming gRPC server

We add a new `ratingStore` to the `LaptopServer` struct and the `NewLaptopServer()` function.

```go
type LaptopServer struct {
	laptopStore LaptopStore
	imageStore  ImageStore
	ratingStore RatingStore
}

func NewLaptopServer(laptopStore LaptopStore, imageStore ImageStore, ratingStore RatingStore) *LaptopServer {
	return &LaptopServer{laptopStore, imageStore, ratingStore}
}
```

Now let’s implement the `RateLaptop` function!

Since we will receive multiple requests from the stream, we must use a for loop here. Similar to what we did on the client-streaming RPC, before doing anything, let’s check the context error to see if it’s already canceled or deadline exceeded or not.

```go
func (server *LaptopServer) RateLaptop(stream pb.LaptopService_RateLaptopServer) error {
	for {
		err := contextError(stream.Context())
		if err != nil {
			return err
		}

		req, err := stream.Recv()
		if err == io.EOF {
			log.Print("no more data")
			break
		}
		if err != nil {
			return logError(status.Errorf(codes.Unknown, "cannot receive stream request: %v", err))
        }
        ...
    }

    return nil
}
```

Then we call stream.Recv() to get a request from the stream. If error is end of file (EOF), then there’s no more data, we simply break the loop. Else if error is not nil, we log it and return the error with status code unknown to the client.

Otherwise, we can get the laptop ID and the score from the request. Let’s write a log here saying that we have received a request with this laptop ID and score.

```go
func (server *LaptopServer) RateLaptop(stream pb.LaptopService_RateLaptopServer) error {
	for {
        ...
        laptopID := req.GetLaptopId()
		score := req.GetScore()

        log.Printf("received a rate-laptop request: id = %s, score = %.2f", laptopID, score)
        
        found, err := server.laptopStore.Find(laptopID)
		if err != nil {
			return logError(status.Errorf(codes.Internal, "cannot find laptop: %v", err))
		}
		if found == nil {
			return logError(status.Errorf(codes.NotFound, "laptopID %s is not found", laptopID))
        }
        
        ...
    }

    return nil
}
```

Then we should check if this laptop ID really exists or not by using the `laptopStore.Find()` function. If an error occurs, we return it with the status code `Internal`. If the laptop is not found, we return the status code NotFound to the client.

If everything goes well, we call `ratingStore.Add()` to add the new laptop score to the store and get back the updated rating object.

```go
func (server *LaptopServer) RateLaptop(stream pb.LaptopService_RateLaptopServer) error {
	for {
        ...

        rating, err := server.ratingStore.Add(laptopID, score)
		if err != nil {
			return logError(status.Errorf(codes.Internal, "cannot add rating to the store: %v", err))
		}

		res := &pb.RateLaptopResponse{
			LaptopId:     laptopID,
			RatedCount:   rating.Count,
			AverageScore: rating.Sum / float64(rating.Count),
		}

		err = stream.Send(res)
		if err != nil {
			return logError(status.Errorf(codes.Unknown, "cannot send stream response: %v", err))
		}
    }

    return nil
}
```

If there’s an error, we return `Internal` status code. Else, we create a `RateLaptopResponse` with laptop ID is the input laptop ID, rated count taken from the rating object, and average score is computed using the sum and count of the rating.

Then we call `stream.Send()` to send the response to the client. If error is not `nil`, we log it and return status code `Unknown`.

And that’s it! We’re done with the server.

## 3. Implement the bidi-streaming gRPC client

Now before moving to the client, I will add a new function to generate a random laptop score in the `sample` package. To be simple, let’s say it’s gonna be a random integer between 1 and 10.

```go
func RandomLaptopScore() float64 {
	return float64(randomInt(1, 10))
}
```

Alright, now let’s implement the client!

First we define a `rateLaptop()` function with 3 input parameters: a laptop client, a list of laptop IDs and their corresponding scores. In this function, we create a new context with timeout after 5 seconds. Then we call `laptopClient.RateLaptop()` with the created context.

```go
func rateLaptop(laptopClient pb.LaptopServiceClient, laptopIDs []string, scores []float64) error {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	stream, err := laptopClient.RateLaptop(ctx)
	if err != nil {
		return fmt.Errorf("cannot rate laptop: %v", err)
    }
    
    ...
}
```

The output is a stream or an error. If error is not nil, we just return it. Else, we will have to make a channel to wait for the responses from the server. The `waitResponse` channel will receive an error when it occurs, or a `nil` if all responses are received successfully.

Note that the requests and responses are sent concurrently, so we have to start a new go routine to receive the responses. In the go routine, we use a for loop, and call `stream.Recv()` to get a response from the server.

```go
func rateLaptop(laptopClient pb.LaptopServiceClient, laptopIDs []string, scores []float64) error {
    ...

    waitResponse := make(chan error)

	// go routine to receive responses
	go func() {
		for {
			res, err := stream.Recv()
			if err == io.EOF {
				log.Print("no more responses")
				waitResponse <- nil
				return
			}
			if err != nil {
				waitResponse <- fmt.Errorf("cannot receive stream response: %v", err)
				return
			}

			log.Print("received response: ", res)
		}
    }()
    
    ...
}
```

If error is EOF, it means there’s no more responses, so we send `nil` to the `waitResponse` channel, and return. Else, if error is not `nil`, we send the error to the `waitResponse` channel, and return as well. If no errors occur, we just write a simple log.

OK, now after this go routine, we can start sending requests to the server. Let’s iterate through the list of the laptops and create a new request for each of them with the input laptop ID and the corresponding input scores.

```go
func rateLaptop(laptopClient pb.LaptopServiceClient, laptopIDs []string, scores []float64) error {
    ...

	// send requests
	for i, laptopID := range laptopIDs {
		req := &pb.RateLaptopRequest{
			LaptopId: laptopID,
			Score:    scores[i],
		}

		err := stream.Send(req)
		if err != nil {
			return fmt.Errorf("cannot send stream request: %v - %v", err, stream.RecvMsg(nil))
		}

		log.Print("sent request: ", req)
	}

    ...
}
```

Then we call `stream.Send()` to send the request to the server. If we get an error, just return it. Note that here we call `stream.RecvMsg()` to get the real error, just like what we did in [the previous lecture with client-streaming RPC](https://dev.to/techschoolguru/upload-file-in-chunks-with-client-streaming-grpc-golang-4loc). If no error occurs, we write a log saying the request is sent. 

Now one important thing that we must do after sending all requests, which is, to call `stream.CloseSend()` to tell the server that we won’t send any more data. And finally read from the `waitResponse` channel and return the received error.

```go
func rateLaptop(laptopClient pb.LaptopServiceClient, laptopIDs []string, scores []float64) error {
    ...

	err = stream.CloseSend()
	if err != nil {
		return fmt.Errorf("cannot close send: %v", err)
	}

	err = <-waitResponse
	return err
}
```

The `rateLaptop()` function is completed. Now we will write a `testRateLaptop()` function to call it.

Let’s say we want to rate 3 laptops, so we declare a slice to keep the laptop IDs. We use a for loop to generate a random laptop, save its ID to the slice, and call `createLaptop()` function to create it on the server.

```go
func testRateLaptop(laptopClient pb.LaptopServiceClient) {
	n := 3
	laptopIDs := make([]string, n)

	for i := 0; i < n; i++ {
		laptop := sample.NewLaptop()
		laptopIDs[i] = laptop.GetId()
		createLaptop(laptopClient, laptop)
	}

    ...
}
```

Then we also make a slice to keep the scores. I want to rate these 3 laptops in multiple rounds, so I will use a for loop here and ask if we want to do another round of rating or not.

```go
func testRateLaptop(laptopClient pb.LaptopServiceClient) {
    ...

    scores := make([]float64, n)
	for {
		fmt.Print("rate laptop (y/n)? ")
		var answer string
		fmt.Scan(&answer)

		if strings.ToLower(answer) != "y" {
			break
		}

		for i := 0; i < n; i++ {
			scores[i] = sample.RandomLaptopScore()
		}

		err := rateLaptop(laptopClient, laptopIDs, scores)
		if err != nil {
			log.Fatal(err)
		}
	}
}
```

If the answer is no, we break the loop. Else we generate a new set of scores for the laptops and call `rateLaptop()` function to rate them with the generated scores.

If an error occurs, we write a fatal log. In the main function, just call `testRateLaptop()`, and we’re all set.

```go
func main() {
	serverAddress := flag.String("address", "", "the server address")
	flag.Parse()
	log.Printf("dial server %s", *serverAddress)

	conn, err := grpc.Dial(*serverAddress, grpc.WithInsecure())
	if err != nil {
		log.Fatal("cannot dial server: ", err)
	}

	laptopClient := pb.NewLaptopServiceClient(conn)
	testRateLaptop(laptopClient)
}
```

## 4. Run the bidi-streaming gRPC server and client

Let’s run the server, then run the client.

![Run the client](https://dev-to-uploads.s3.amazonaws.com/i/vgevs7ichxlamfjjsz5q.png)

3 laptops are created. Let's press `y` to rate these laptops.

![Rate round 1](https://dev-to-uploads.s3.amazonaws.com/i/56orycut0t6g6h1fxzdb.png)

As you can see, we sent 3 requests with scores of 10, 8 and 4, and received 3 responses with rated count of 1, and average scores of 10, 8 and 4.

Let’s do another rating round! 

![Rate round 2](https://dev-to-uploads.s3.amazonaws.com/i/32pr59h2apc20f7thzih.png)

This time the scores we sent are 6, 1 and 5, and the responses has rated count of 2, and the average scores have been updated to 8, 4.5 and 4.5, which are all correct.

## 5. Test bidi-streaming gRPC

Now I’m gonna show you how to test this bidirectional streaming RPC. Let’s go to `service/laptop_client_test.go` file.

The test setup will be very similar to the upload image test that we've written in the last lecture. 

We just create a new laptop store, new rating store, generate a random laptop and save it to the store.

```go
func TestClientRateLaptop(t *testing.T) {
	t.Parallel()

	laptopStore := service.NewInMemoryLaptopStore()
	ratingStore := service.NewInMemoryRatingStore()

	laptop := sample.NewLaptop()
	err := laptopStore.Save(laptop)
	require.NoError(t, err)

    ...
}
```

Then we start the test laptop server to get the server adress, and use it to create a test laptop client.

```go
func TestClientRateLaptop(t *testing.T) {
    ...
    
    serverAddress := startTestLaptopServer(t, laptopStore, nil, ratingStore)
    laptopClient := newTestLaptopClient(t, serverAddress)
    
	stream, err := laptopClient.RateLaptop(context.Background())
    require.NoError(t, err)

    ...
}
```

After that, we call `laptopClient.RateLaptop()` with a background context to get the stream, and require no error.

For simplicity, we just rate 1 single laptop, but we will rate it 3 times with a score of 8, 7.5 and 10 respectively. So the expected average score after each time should be 8, 7.75 and 8.5.

We define n as the number of rated times, and use a for loop to send multiple requests.

```go
func TestClientRateLaptop(t *testing.T) {
    ...
    
	scores := []float64{8, 7.5, 10}
	averages := []float64{8, 7.75, 8.5}

	n := len(scores)
	for i := 0; i < n; i++ {
		req := &pb.RateLaptopRequest{
			LaptopId: laptop.GetId(),
			Score:    scores[i],
		}

		err := stream.Send(req)
		require.NoError(t, err)
    }
    
    err = stream.CloseSend()
	require.NoError(t, err)

    ...
}
```

Each time we will create a new request with the same laptop ID and a new score. We call `stream.Send()` to send the request to the server, and require no errors to be returned. After sending all the rate laptop requests, we call `stream.CloseSend()` just like what we did in the client code.

To be simple, I don't create a separate go routine to receive the responses. Here I simply use a for loop to receive them, and use an idx variable to count how many responses we have received.

```go
func TestClientRateLaptop(t *testing.T) {
    ...

	for idx := 0; ; idx++ {
		res, err := stream.Recv()
		if err == io.EOF {
			require.Equal(t, n, idx)
			return
		}

		require.NoError(t, err)
		require.Equal(t, laptop.GetId(), res.GetLaptopId())
		require.Equal(t, uint32(idx+1), res.GetRatedCount())
		require.Equal(t, averages[idx], res.GetAverageScore())
	}
}
```

Inside the loop, we call `stream.Recv()` to receive a new response. If error is `EOF`, then it’s the end of the stream, we just require that the number of responses we received must be equal to n, which is the number of requests we sent, and we return immediately.

Else, there should be no error. The response laptop ID should be equal to the input laptop ID. The rated count should be equal to idx + 1. And the average score should be equal to the expected value.

Now let’s run the test.

![Run unit test](https://dev-to-uploads.s3.amazonaws.com/i/g5qioctwletapzu50eoe.png)

It passed. Excellent!

And that wraps up today’s lecture about implementing bidirectional streaming RPC in Go.

Thank you for reading! Happy coding, and I will see you in the next lecture!

---

If you like the article, please [subscribe to our Youtube channel](http://bit.ly/techschoolguru) and [follow us on Twitter](https://twitter.com/techschoolguru) for more tutorials in the future.

---
If you want to join me on my current amazing team at Voodoo, check out [our job openings here](https://bit.ly/voodoojobs). Remote or onsite in Paris/Amsterdam/London/Berlin/Barcelona with visa sponsorship.