0Pricing
gRPC & High Performance APIs · Leçon

Le flux client expliqué

Apprenez à mettre en œuvre le flux côté client, qui permet à un client d’envoyer une suite de messages au serveur.

Le flux client expliqué est une leçon gRPC & High Performance APIs gratuite sur CoddyKit. Ceci est la leçon 2 sur 4. Tu peux lire la leçon complète ci-dessous gratuitement — puis la pratiquer en direct dans le navigateur avec un éditeur de code intégré et un tuteur IA 24/7. Elle fait partie du parcours d'apprentissage gRPC & High Performance APIs, et ta progression se synchronise sur le web et l'application CoddyKit. Le cours gRPC & High Performance APIs comprend 4 leçons au total.

Certaines parties de cette leçon n'ont pas encore été traduites et s'affichent en anglais.

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!

Questions Fréquemment Posées

La leçon « Le flux client expliqué » est-elle gratuite ?

Oui — le texte complet de « Le flux client expliqué » est gratuit à lire ici sur le web. Pour la pratiquer de manière interactive (un éditeur de code intégré et un tuteur IA 24/7) et déverrouiller le reste du cours gRPC & High Performance APIs, passe à CoddyKit PRO. Le cours gRPC & High Performance APIs comprend 4 leçons au total.

Qu'est-ce que j'apprendrai dans « Le flux client expliqué » ?

Apprenez à mettre en œuvre le flux côté client, qui permet à un client d’envoyer une suite de messages au serveur. Tu pratiques gRPC & High Performance APIs avec du code pratique que tu exécutes directement dans le navigateur, et un tuteur IA 24/7 répond à tes questions au fur et à mesure que tu avances dans la leçon.

Dois-je avoir de l'expérience pour commencer gRPC & High Performance APIs ?

Aucune expérience préalable n'est requise. gRPC & High Performance APIs sur CoddyKit est structuré pour les débutants jusqu'aux apprenants avancés, donc tu peux commencer ici ou depuis le début et avancer à ton rythme. Ceci est la leçon 2 sur 4.

Combien de temps prend la leçon « Le flux client expliqué » ?

La plupart des leçons CoddyKit prennent environ 5–10 minutes. Chacune est courte et interactive, tu progresses régulièrement et tu repiques exactement où tu t'es arrêté sur le web et l'app.

Peux-tu écrire et exécuter du code dans cette leçon gRPC & High Performance APIs ?

Oui. Chaque leçon gRPC & High Performance APIs inclut un éditeur de code intégré, tu écris et exécutes du vrai code directement dans ton navigateur et tu reçois des retours IA instantanés — aucune configuration locale requise.

Toutes les leçons de ce cours

  1. Le flux serveur expliqué
  2. Le flux client expliqué
  3. Flux bidirectionnel
  4. Contrôle du flux et rétropression des flux
← Retour à gRPC & High Performance APIs