0Pricing
gRPC & High Performance APIs · Lesson

Client Streaming Explained

Learn how to implement client-side streaming, allowing a client to send a sequence of messages to the server.

Client Streaming Explained is a free gRPC & High Performance APIs lesson on CoddyKit — lesson 2 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the gRPC & High Performance APIs learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

What is Client Streaming?

Welcome to client streaming! In gRPC, client streaming is a communication pattern where the client sends a sequence of messages to the server.

Unlike a simple unary RPC (request-response), the client doesn't just send one message. Instead, it sends a stream of messages, and the server processes them, then sends back a single response at the end.

How Client Streaming Works

Imagine uploading a large file by sending it in many small chunks. The server collects all chunks, rebuilds the file, and then sends a single "upload complete" confirmation.

  • The client initiates the RPC.
  • The client sends multiple messages asynchronously.
  • The server receives and processes these messages.
  • Once the client finishes sending (signals completion), the server sends back a single response.

Defining Client Stream in Protobuf

To define a client-streaming method in your .proto file, you use the stream keyword for the request type, but not for the response type.

Here's an example for a log upload service:

syntax = "proto3";

package client_streaming;

service LogService {
  rpc UploadLogs (stream LogEntry) returns (UploadSummary);
}

message LogEntry {
  string message = 1;
  int64 timestamp = 2;
}

message UploadSummary {
  int32 uploaded_count = 1;
  string status_message = 2;
}

Server: Handling the Client Stream

On the server side, your method will receive a StreamObserver for the client's incoming messages and will use another StreamObserver to send its single response.

The server's StreamObserver will have onNext() for each incoming message, onError() for errors, and onCompleted() when the client finishes sending.

import io.grpc.stub.StreamObserver;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import client_streaming.LogEntry;
import client_streaming.LogServiceGrpc;
import client_streaming.UploadSummary;

public class LogServer {
    private Server server;

    private void start() throws Exception {
        int port = 50051;
        server = ServerBuilder.forPort(port)
            .addService(new LogServiceImpl())
            .build()
            .start();
        System.out.println("Server started, listening on " + port);
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.err.println("*** shutting down gRPC server since JVM is shutting down");
            LogServer.this.stop();
            System.err.println("*** server shut down");
        }));
    }

    private void stop() {
        if (server != null) {
            server.shutdown();
        }
    }

    private void blockUntilShutdown() throws InterruptedException {
        if (server != null) {
            server.awaitTermination();
        }
    }

    public static void main(String[] args) throws Exception {
        final LogServer logServer = new LogServer();
        logServer.start();
        logServer.blockUntilShutdown();
    }

    static class LogServiceImpl extends LogServiceGrpc.LogServiceImplBase {
        @Override
        public StreamObserver<LogEntry> uploadLogs(StreamObserver<UploadSummary> responseObserver) {
            return new StreamObserver<LogEntry>() {
                private int logCount = 0;

                @Override
                public void onNext(LogEntry log) {
                    // Process each log entry as it arrives
                    System.out.println("Received log: " + log.getMessage() + " at " + log.getTimestamp());
                    logCount++;
                }

                @Override
                public void onError(Throwable t) {
                    System.err.println("UploadLogs cancelled or failed: " + t.getMessage());
                    responseObserver.onError(t);
                }

                @Override
                public void onCompleted() {
                    // After all logs are received, send a single summary response
                    UploadSummary summary = UploadSummary.newBuilder()
                        .setUploadedCount(logCount)
                        .setStatusMessage("Successfully processed " + logCount + " log entries.")
                        .build();
                    responseObserver.onNext(summary);
                    responseObserver.onCompleted();
                    System.out.println("Finished processing client stream. Sent summary.");
                }
            };
        }
    }
}

Client: Sending the Stream

On the client side, you'll get a StreamObserver to send your messages. You call onNext() for each message you want to send and finally onCompleted() to signal the end of the stream.

The server's single response will be handled by a separate StreamObserver you provide.

import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import client_streaming.LogEntry;
import client_streaming.LogServiceGrpc;
import client_streaming.UploadSummary;

import java.util.concurrent.TimeUnit;

public class LogClient {
    private final LogServiceGrpc.LogServiceStub asyncStub;
    private final ManagedChannel channel;

    public LogClient(String host, int port) {
        channel = ManagedChannelBuilder.forAddress(host, port)
            .usePlaintext() // For demonstration, use plaintext
            .build();
        asyncStub = LogServiceGrpc.newStub(channel);
    }

    public void shutdown() throws InterruptedException {
        channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
    }

    public void uploadMultipleLogs() throws InterruptedException {
        StreamObserver<UploadSummary> responseObserver = new StreamObserver<UploadSummary>() {
            @Override
            public void onNext(UploadSummary summary) {
                System.out.println("Server Response: " + summary.getStatusMessage() + " (" + summary.getUploadedCount() + " logs)");
            }

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

            @Override
            public void onCompleted() {
                System.out.println("Server has completed processing.");
            }
        };

        StreamObserver<LogEntry> requestObserver = asyncStub.uploadLogs(responseObserver);

        try {
            // Send multiple log entries
            LogEntry log1 = LogEntry.newBuilder().setMessage("User login attempt").setTimestamp(System.currentTimeMillis()).build();
            LogEntry log2 = LogEntry.newBuilder().setMessage("Database query executed").setTimestamp(System.currentTimeMillis() + 100).build();
            LogEntry log3 = LogEntry.newBuilder().setMessage("API call completed").setTimestamp(System.currentTimeMillis() + 200).build();

            requestObserver.onNext(log1);
            System.out.println("Client sent log 1");
            Thread.sleep(100); // Simulate some delay
            requestObserver.onNext(log2);
            System.out.println("Client sent log 2");
            Thread.sleep(100);
            requestObserver.onNext(log3);
            System.out.println("Client sent log 3");

            // Mark the end of the client stream
            requestObserver.onCompleted();
            System.out.println("Client finished sending logs.");

            // Wait for server response (handled by responseObserver)
            Thread.sleep(1000); // Give time for server to respond
        } catch (RuntimeException e) {
            requestObserver.onError(e);
            throw e;
        }
    }

    public static void main(String[] args) throws Exception {
        LogClient client = new LogClient("localhost", 50051);
        try {
            client.uploadMultipleLogs();
        } finally {
            client.shutdown();
        }
    }
}

Running the Example

To see client streaming in action:

  1. First, compile your .proto file to generate the necessary Java classes.
  2. Run the LogServer application. It will start listening for requests.
  3. Then, run the LogClient application. It will send three log entries and wait for the server's summary.

Observe the console output from both the client and server to understand the flow of messages.

Key StreamObserver Methods

The StreamObserver interface is crucial for handling streaming RPCs. Both the client and server use implementations of this interface.

  • onNext(T value): Called for each message received in the stream. The client uses this to send messages, the server uses it to receive.
  • onError(Throwable t): Called if an RPC fails or is cancelled.
  • onCompleted(): Called when the stream has finished. The client calls this after sending all messages; the server calls it after sending its final response.

When to Use Client Streaming

Client streaming is ideal for scenarios where a client needs to send a large amount of data or a series of related messages to a server, and only cares about a single final result.

  • Large File Uploads: Sending a file chunk by chunk.
  • Log Aggregation: A client sending many log entries to a central logging service.
  • Batch Operations: Sending a list of items to be processed as a single batch, receiving a summary.
  • Sensor Data Collection: Continuously sending readings from a sensor device.

Quick Check

You're designing a gRPC service where a client needs to send a series of sensor readings to a server, and the server will process them and return a single summary report.

Recap: Client Streaming

You've learned about client-side streaming in gRPC!

  • Client streaming allows a client to send a sequence of messages.
  • The server processes these messages and sends a single response back.
  • In Protobuf, you define it by using the stream keyword for the request type.
  • Both client and server use StreamObserver to manage the flow of messages (onNext(), onError(), onCompleted()).
  • It's great for tasks like uploading large data or sending continuous log entries.

Next, we'll explore server-side streaming!

Frequently asked questions

Is the “Client Streaming Explained” lesson free?

Yes — the full text of “Client Streaming Explained” is free to read here on the web, and the gRPC & High Performance APIs course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the gRPC & High Performance APIs course, upgrade to CoddyKit PRO.

What will I learn in “Client Streaming Explained”?

Learn how to implement client-side streaming, allowing a client to send a sequence of messages to the server. You practise gRPC & High Performance APIs with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start gRPC & High Performance APIs?

No prior experience is required. gRPC & High Performance APIs on CoddyKit is structured for beginners through advanced learners; this is — lesson 2 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Client Streaming Explained” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this gRPC & High Performance APIs lesson?

Yes. Every gRPC & High Performance APIs lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Server Streaming Explained
  2. Client Streaming Explained
  3. Bidirectional Streaming
  4. Streaming Flow Control & Backpressure
← Back to gRPC & High Performance APIs