gRPC Unidirectional Streaming in Java with Example: A Complete Guide

By /

Overview

gRPC Unidirectional Streaming in Java allows a server to continuously send multiple responses to the client after receiving a single request. The connection remains open, and the client can process the incoming stream of messages in real-time. This type of streaming is highly efficient for real-time systems, such as monitoring applications, chat services, or live data feeds.

What is gRPC Unidirectional Streaming?

gRPC (Google Remote Procedure Call) is an open-source framework that allows for seamless communication between distributed systems. Unlike traditional communication models where a client sends a request and waits for a response, gRPC Unidirectional Streaming in Java enables one-way data flow. The server continuously streams data to the client after an initial request, without waiting for the client to respond after each message. This streamlining makes it ideal for use cases like real-time dashboards, sensor data processing, and live feeds.

Use Cases of gRPC Unidirectional Streaming

  1. Real-Time Data Analytics
    Companies that rely on real-time data processing, like monitoring stock prices or analyzing web traffic, benefit from gRPC unidirectional streaming. With the ability to stream continuous data, the system can analyze trends as they happen.
  2. IoT and Sensor Data
    Imagine a smart city filled with sensors monitoring traffic, weather, or energy consumption. Each sensor streams its data in real time to a central system using gRPC Unidirectional Streaming in Java, which then processes it without interruption.
  3. Live Media Streaming
    gRPC unidirectional streaming is perfect for applications like live video feeds, where the server streams a continuous video or audio feed to thousands of clients. The efficiency and low latency ensure a smooth viewing experience.

How gRPC Unidirectional Streaming Works

In a gRPC unidirectional streaming scenario, communication follows this general flow:

  1. The client sends a request to the server.
  2. The server opens a stream and responds with multiple messages.
  3. The server continues streaming data until all messages are sent or the stream is manually closed.

The client can consume these messages in real-time without having to make additional requests.

Pre-Requisites

Example Use Case Implementation

In our example – we will implement the below use case.
Lets Imagine you’re building a real-time stock price monitoring system. The client (a dashboard) requests stock prices for a particular company, and the server continuously streams the latest price updates as they become available.

Project Structure

gRPC Unidirectional Streaming in Java

1. Dependency management

Make sure to include the gRPC and protobuf dependencies in your pom.xml or build.gradle.

Example pom.xml dependencies:

<!-- gRPC dependencies -->
<dependency>
   <groupId>io.grpc</groupId>
   <artifactId>grpc-netty-shaded</artifactId>
   <version>1.47.0</version>
</dependency>

<dependency>
   <groupId>io.grpc</groupId>
   <artifactId>grpc-protobuf</artifactId>
   <version>1.47.0</version>
</dependency>

<dependency>
   <groupId>io.grpc</groupId>
   <artifactId>grpc-stub</artifactId>
   <version>1.47.0</version>
</dependency>

<!-- Protobuf compiler -->
<dependency>
   <groupId>com.google.protobuf</groupId>
   <artifactId>protobuf-java</artifactId>
   <version>3.21.6</version>
</dependency>

2. Define the gRPC Service in Protobuf

irst, we define the service using Protocol Buffers (protobuf), which gRPC uses for data serialization. Here’s a basic .proto file defining the gRPC Unidirectional Streaming in Java service:

syntax = "proto3";

option java_multiple_files = true;
option java_package = "com.javatecharc.grpc.proto.stock";
option java_outer_classname = "StockPriceProto";
option optimize_for = SPEED;

service StockPriceService {
    rpc StreamStockPrices(StockRequest) returns (stream StockPriceResponse);
}

message StockRequest {
    string stock_symbol = 1;
}

message StockPriceResponse {
    string stock_symbol = 1;
    double price = 2;
    string timestamp = 3;
}
  • Compile the above proto generate Java Source from it
  • The StreamStockPrices method allows the client to request stock prices by providing a stock symbol.
  • The server then responds with a stream of StockPriceResponse messages.

3. Implement the Server

Now, let’s implement the server in gRPC Unidirectional Streaming in Java that streams stock prices. The server will continuously send updated stock prices at regular intervals.

import io.grpc.Server;
import io.grpc.ServerBuilder;
import java.io.IOException;

public class StockPriceServer {
    public static void main(String[] args) throws IOException, InterruptedException {
        Server server = ServerBuilder.forPort(18090)
                .addService(new StockPriceService())
                .build();

        System.out.println("Starting server on port 18090...");
        server.start();
        server.awaitTermination();
    }
}

4. Implement the Service Endpoint

Implement gRPC Unidirectional Streaming in Java endpoint, which will be having implementation of business service.

package com.javatecharc.demo.grpc.unidirectional;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;

import com.javatecharc.grpc.proto.stock.*;
import com.javatecharc.grpc.proto.stock.StockPriceServiceGrpc;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class StockPriceService extends StockPriceServiceGrpc.StockPriceServiceImplBase {

    @Override
    public void streamStockPrices(StockRequest request, StreamObserver<StockPriceResponse> responseObserver) {
        String stockSymbol = request.getStockSymbol();
        Random random = new Random();
        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

        for (int i = 0; i < 10; i++) { // Stream 10 stock prices
            double price = 100 + (random.nextDouble() * 50); // Random price between 100 and 150
            String timestamp = formatter.format(new Date());

            StockPriceResponse stockPrice = StockPriceResponse.newBuilder()
                    .setStockSymbol(stockSymbol)
                    .setPrice(price)
                    .setTimestamp(timestamp)
                    .build();

            // Send the price to the client
            responseObserver.onNext(stockPrice);

            try {
                Thread.sleep(1000); // Simulate time interval
            } catch (InterruptedException e) {
                log.error(e.getMessage());
            }
        }

        responseObserver.onCompleted(); // Close the stream when done
    }
}

This service:

  • Streams 10 random stock prices to the client for a given stock symbol.
  • Uses StreamObserver to send messages (onNext()) and closes the stream with onCompleted() after the data is fully transmitted.

5. Implement the gRPC Client

The client will send a request to the server and continuously receive stock prices as they are streamed.

package com.javatecharc.demo.grpc.unidirectional;

import com.javatecharc.grpc.proto.stock.StockPriceResponse;
import com.javatecharc.grpc.proto.stock.StockPriceServiceGrpc;
import com.javatecharc.grpc.proto.stock.StockRequest;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class StockPriceClient {
    public static void main(String[] args) {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 18090)
                .usePlaintext()
                .build();

        StockPriceServiceGrpc.StockPriceServiceStub asyncStub = StockPriceServiceGrpc.newStub(channel);

        StockRequest request = StockRequest.newBuilder()
                .setStockSymbol("JAVATECHARC")
                .build();

        StreamObserver<StockPriceResponse> responseObserver = new StreamObserver<>() {
            @Override
            public void onNext(StockPriceResponse value) {
                System.out.println("Received stock price: " +
                        value.getStockSymbol() + " $" +
                        value.getPrice() + " at " +
                        value.getTimestamp());
            }

            @Override
            public void onError(Throwable t) {
                System.err.println("Error: " + t.getMessage());
            }

            @Override
            public void onCompleted() {
                System.out.println("Stream completed.");
                channel.shutdown();
            }
        };

        asyncStub.streamStockPrices(request, responseObserver);

        try {
            Thread.sleep(12000); // Wait for streaming to finish
        } catch (InterruptedException e) {
            log.error(e.getMessage());
        }
    }
}

This client:

  • Sends a request to the server for stock prices with a specific symbol (e.g., “JAVATECHARC”).
  • Implements a StreamObserver to receive and print the stream of StockPrice messages.
  • Shuts down the channel once the server finishes streaming the data.

Benefits of gRPC Unidirectional Streaming

  • Real-Time Data Delivery: The server can continuously send data without requiring the client to make new requests, which is perfect for real-time applications.
  • Lower Latency: By keeping a persistent connection open, gRPC streaming reduces the time spent on establishing new connections for each piece of data.
  • Efficient Use of Resources: Since messages are streamed over a single connection, it reduces network overhead and makes better use of available resources.

Conclusion

gRPC Unidirectional Streaming in Java is an ideal solution for scenarios where continuous, real-time data needs to be sent from a server to a client. It helps reduce latency, optimize network usage, and handle large-scale, time-sensitive data applications with ease. By leveraging this approach, you can build efficient, scalable systems for anything from live data feeds to sensor networks.

The source code of the examples can be found on github.

Related Articles

Leave a Comment

Your email address will not be published. Required fields are marked *

Table of Contents

→ Index
Scroll to Top