Spring Boot 3 + gRPC Bidirectional Streaming Example

In previous tutorial we looked at Spring Boot 3 + gRPC Client Streaming Example. In this tutorial we will be implementing spring boot + gRPC Bidirectional Streaming call. Bidirectional streaming gRPC is a communication pattern in which both the client and server can send multiple messages to each other in a continuous stream. It allows real-time, two-way communication between the client and server.
This pattern is commonly used in scenarios where there is a need for ongoing, interactive communication. For example, in a gaming application where multiple gamers may need to continously interact with each other. Other scenario is the autocomplete feature in google search, where the user types some word and is given some possible options which he may select.
We will be implementing the following scenario. Consider a banking service where the client chats with bank customer service. The client can send multiple requests and receive multiple responses. Similarly the bank customer service can receive multiple requests and send multiple responses.

We will implement the following conversation.

• 1. When the client sends the message "Hi", the server responds with "Hello how can i help you today."
• 2. If the client requests information about their account balance by sending the message "I need to know my account balance", the server acknowledges this request with the message "Ok. Please type the OTP received on your registered phone".
• 3. Upon receiving the OTP "5050" from the client, the server responds with "Ok. The account balance is 300".
• 4. In case the client sends an OTP other than "5050", the server responds with "Invalid OTP".

gRPC - Table of Contents
Spring Boot+ gRPC Hello World Example
Spring Boot gRPC Server + C# gRPC Client Example
Spring Boot 3 + gRPC - Types of gRPC
Spring Boot 3 + gRPC Unary Example
Spring Boot 3 + gRPC Server Streaming Example
Spring Boot 3 + gRPC Client Streaming Example
Spring Boot 3 + gRPC Bidirectional Streaming Example
Spring Boot + gRPC Deadline Example
Spring Boot + gRPC Error Handling Example
Spring Boot + gRPC Error Handling - Using Trailer Metadata
Spring Boot + gRPC Error Handling - Global Exception Handler Using GrpcAdvice

Implementation

Spring Boot Bidirectional Streaming gRPC Server

We will be creating a maven project as follows -

The pom.xml will be as follows-

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>com.javainuse</groupId>
	<artifactId>boot-bidirectional-streaming-grpc-server</artifactId>
	<version>0.0.1-SNAPSHOT</version>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>3.2.0</version>
		<relativePath /> <!-- lookup parent from repository -->
	</parent>

	<properties>
		<protobuf.version>3.17.3</protobuf.version>
		<protobuf-plugin.version>0.6.1</protobuf-plugin.version>
		<grpc.version>1.59.0</grpc.version>
	</properties>
	<dependencies>

		<dependency>
			<groupId>net.devh</groupId>
			<artifactId>grpc-server-spring-boot-starter</artifactId>
			<version>2.15.0.RELEASE</version>
		</dependency>

		<dependency>
			<groupId>io.grpc</groupId>
			<artifactId>grpc-stub</artifactId>
			<version>${grpc.version}</version>
		</dependency>
		<dependency>
			<groupId>io.grpc</groupId>
			<artifactId>grpc-protobuf</artifactId>
			<version>${grpc.version}</version>
		</dependency>

		<dependency>
			<groupId>org.apache.tomcat</groupId>
			<artifactId>annotations-api</artifactId>
			<version>6.0.53</version>
			<scope>provided</scope>
		</dependency>

	</dependencies>

	<build>
		<extensions>
			<extension>
				<groupId>kr.motd.maven</groupId>
				<artifactId>os-maven-plugin</artifactId>
				<version>1.7.1</version>
			</extension>
		</extensions>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
			<plugin>
				<groupId>org.xolstice.maven.plugins</groupId>
				<artifactId>protobuf-maven-plugin</artifactId>
				<version>0.6.1</version>
				<configuration>
					<protocArtifact>
						com.google.protobuf:protoc:3.24.0:exe:${os.detected.classifier}
					</protocArtifact>
					<pluginId>grpc-java</pluginId>
					<pluginArtifact>
						io.grpc:protoc-gen-grpc-java:1.59.0:exe:${os.detected.classifier}
					</pluginArtifact>
					<protoSourceRoot>
						${basedir}/src/main/proto/
					</protoSourceRoot>
				</configuration>
				<executions>
					<execution>
						<goals>
							<goal>compile</goal>
							<goal>compile-custom</goal>
						</goals>
					</execution>
				</executions>
			</plugin>
		</plugins>
	</build>
</project>

Next create the proto file named bank-service.proto in src/main/proto folder. A proto file in gRPC is a special type of file that helps define the structure and communication between different software components. It acts like a blueprint for creating and interacting with these components. The file also defines a service ChatService for bidirectional streaming in a chat system.

syntax = "proto3";

package banking;

option java_multiple_files = true;
option java_package = "com.javainuse.banking";

// Message representing a client's message in a chat system
message ChatMessage {
  string sender = 1;
  string message = 2;
}

// Message representing a server's response in a chat system
message ChatResponse {
  string sender = 1;
  string message = 2;
}

// Service for bidirectional streaming for chat support
service ChatService {
  rpc StartChat (stream ChatMessage) returns (stream ChatResponse) {}
}

Next create the spring bootstrap class as follows-

package com.javainuse.bank;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class SpringBootGrpcServerExampleApplication {

	public static void main(String[] args) {
		SpringApplication.run(SpringBootGrpcServerExampleApplication.class, args);
	}
}

Run the maven install command which will generate the classes using the proto file.

Create the class named ChatStreamObserver. ChatStreamObserver is a class that implements StreamObserver for ChatMessage. It receives messages from clients, processes them, and sends a response back. It contains methods to handle messages, errors, and completion of the response stream. It takes a responseObserver as a parameter to send responses back to the client.

package com.javainuse.bank.service;

import com.javainuse.banking.ChatMessage;
import com.javainuse.banking.ChatResponse;
import io.grpc.stub.StreamObserver;

public class ChatStreamObserver implements StreamObserver<ChatMessage> {

	private final StreamObserver<ChatResponse> responseObserver;

	public ChatStreamObserver(StreamObserver<ChatResponse> responseObserver) {
		this.responseObserver = responseObserver;
	}

	@Override
	public void onNext(ChatMessage chatMessage) {
		// Handle the client's message
		String sender = chatMessage.getSender();
		String message = chatMessage.getMessage();

		System.out.println("Received message from: " + sender);
		System.out.println("Message: " + message);

		ChatResponse.Builder response = ChatResponse.newBuilder();

		if (message.equals("Hi")) {
			response.setSender("Server");
			response.setMessage("Hello how can i help you today.");
		} else if (message.equals("I need to know my account balance")) {
			response.setSender("Server");
			response.setMessage("Ok. Please type the otp recieved on your registered phone");
		} else if (message.equals("5050")) {
			response.setSender("Server");
			response.setMessage("Ok. The account balance is 300");
		} else {
			response.setSender("Server");
			response.setMessage("Invalid OTP");
		}

		responseObserver.onNext(response.build());
	}

	@Override
	public void onError(Throwable throwable) {
		// Handle any errors
	}

	@Override
	public void onCompleted() {
		// Complete the response stream
		responseObserver.onCompleted();
	}
}

ChatServiceServer is a service for handling chat messages using gRPC technology. It listens for incoming messages and sends responses using StreamObserver. It is annotated with @GrpcService to indicate it is a gRPC service and extends ChatServiceGrpc.ChatServiceImplBase for implementing the chat service functionality.

package com.javainuse.bank.service;

import com.javainuse.banking.ChatMessage;
import com.javainuse.banking.ChatResponse;
import com.javainuse.banking.ChatServiceGrpc;
import io.grpc.stub.StreamObserver;
import net.devh.boot.grpc.server.service.GrpcService;

@GrpcService
public class ChatServiceServer extends ChatServiceGrpc.ChatServiceImplBase {

	@Override
	public StreamObserver<ChatMessage> startChat(StreamObserver<ChatResponse> responseObserver) {
		return new ChatStreamObserver(responseObserver);
	}

}

Finally create a file named application.properties where we specify the grpc port configuration-

grpc.server.port=8090

Test using BloomRPC

Start the spring boot project created. Start BloomRPC and load the proto file. This will create the gRPC client.

Spring Boot Bidirectional Streaming gRPC Client

We will be creating a maven project as follows -

The pom.xml will be as follows-

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>com.javainuse</groupId>
	<artifactId>boot-bidirectional-streaming-grpc-client</artifactId>
	<version>0.0.1-SNAPSHOT</version>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>3.2.0</version>
		<relativePath /> <!-- lookup parent from repository -->
	</parent>

	<properties>
		<protobuf.version>3.17.3</protobuf.version>
		<protobuf-plugin.version>0.6.1</protobuf-plugin.version>
		<grpc.version>1.59.0</grpc.version>
	</properties>
	<dependencies>

		<dependency>
			<groupId>net.devh</groupId>
			<artifactId>grpc-server-spring-boot-starter</artifactId>
			<version>2.15.0.RELEASE</version>
		</dependency>

		<dependency>
			<groupId>io.grpc</groupId>
			<artifactId>grpc-stub</artifactId>
			<version>${grpc.version}</version>
		</dependency>
		<dependency>
			<groupId>io.grpc</groupId>
			<artifactId>grpc-protobuf</artifactId>
			<version>${grpc.version}</version>
		</dependency>

		<dependency>
			<groupId>org.apache.tomcat</groupId>
			<artifactId>annotations-api</artifactId>
			<version>6.0.53</version>
			<scope>provided</scope>
		</dependency>

	</dependencies>

	<build>
		<extensions>
			<extension>
				<groupId>kr.motd.maven</groupId>
				<artifactId>os-maven-plugin</artifactId>
				<version>1.7.1</version>
			</extension>
		</extensions>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
			<plugin>
				<groupId>org.xolstice.maven.plugins</groupId>
				<artifactId>protobuf-maven-plugin</artifactId>
				<version>0.6.1</version>
				<configuration>
					<protocArtifact>
						com.google.protobuf:protoc:3.24.0:exe:${os.detected.classifier}
					</protocArtifact>
					<pluginId>grpc-java</pluginId>
					<pluginArtifact>
						io.grpc:protoc-gen-grpc-java:1.59.0:exe:${os.detected.classifier}
					</pluginArtifact>
					<protoSourceRoot>
						${basedir}/src/main/proto/
					</protoSourceRoot>
				</configuration>
				<executions>
					<execution>
						<goals>
							<goal>compile</goal>
							<goal>compile-custom</goal>
						</goals>
					</execution>
				</executions>
			</plugin>
		</plugins>
	</build>
</project>

Next create the proto file named bank-service.proto in src/main/proto folder

syntax = "proto3";

package banking;

option java_multiple_files = true;
option java_package = "com.javainuse.banking";

// Message representing a client's message in a chat system
message ChatMessage {
  string sender = 1;
  string message = 2;
}

// Message representing a server's response in a chat system
message ChatResponse {
  string sender = 1;
  string message = 2;
}

// Service for bidirectional streaming for chat support
service ChatService {
  rpc StartChat (stream ChatMessage) returns (stream ChatResponse) {}
}

Next create the spring bootstrap class as follows-

package com.javainuse.bank;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;

@SpringBootApplication
public class SpringBootGrpcClientExampleApplication {

	public static void main(String[] args) throws InterruptedException {
		SpringApplication.run(SpringBootGrpcClientExampleApplication.class, args);
	}
}

Run the maven install command which will generate the classes using the proto file.
The ChatServiceClient class is a service that sends a chat message to a server using gRPC technology. It creates a channel to connect to the server, creates a chat message with sender and message content, and sends it asynchronously. It handles server responses, errors, and completion of the communication.

package com.javainuse.bank.service;

import com.javainuse.banking.ChatMessage;
import com.javainuse.banking.ChatResponse;
import com.javainuse.banking.ChatServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import org.springframework.stereotype.Service;

@Service
public class ChatServiceClient {

	public void sendMessage() { // Create the channel
		ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 8090).usePlaintext().build();

		// Create an asynchronous stub
		ChatServiceGrpc.ChatServiceStub asyncStub = ChatServiceGrpc.newStub(channel);

		StreamObserver<ChatResponse> responseObserver = new StreamObserver<ChatResponse>() {
			@Override
			public void onNext(ChatResponse chatResponse) {
				// Handle the server's response
				System.out.println("Received response from: " + chatResponse.getSender());
				System.out.println("Response: " + chatResponse.getMessage());
			}

			@Override
			public void onError(Throwable throwable) {
				// Handle any errors
			}

			@Override
			public void onCompleted() {
				// Complete the communication
				channel.shutdown();
			}
		};

		// Create the ChatMessage
		ChatMessage chatMessageHi = ChatMessage.newBuilder().setSender("client").setMessage("Hi").build();
		ChatMessage chatMessageBalance = ChatMessage.newBuilder().setSender("client")
				.setMessage("I need to know my account balance").build();
		ChatMessage chatMessageOTP5050 = ChatMessage.newBuilder().setSender("client").setMessage("5050").build();
		ChatMessage chatMessageInvalidOTP = ChatMessage.newBuilder().setSender("client").setMessage("1234").build();

		// Send the chat messages to the server
		asyncStub.startChat(responseObserver).onNext(chatMessageHi);
		asyncStub.startChat(responseObserver).onNext(chatMessageBalance);
		asyncStub.startChat(responseObserver).onNext(chatMessageOTP5050);
		asyncStub.startChat(responseObserver).onNext(chatMessageInvalidOTP);
	}
}

Finally modify the the spring bootstrap class to call the sendMessage of the ChatServiceClient-

package com.javainuse.bank;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;

import com.javainuse.bank.service.BankService;

@SpringBootApplication
public class SpringBootGrpcClientExampleApplication {

	public static void main(String[] args) {
		ApplicationContext context = SpringApplication.run(SpringBootGrpcClientExampleApplication.class, args);
		ChatServiceClient bankService = context.getBean(ChatServiceClient.class);
		bankService.sendMessage();
	}
}

Run the client to get the response.

Download Source Code

Download it - Spring Boot + gRPC Bidirectional Streaming Client Example
Download it - Spring Boot + gRPC Bidirectional Streaming Server Example