Stream calls on Cloud Client Libraries for Java

Streaming calls enable more complex interaction patterns than simple request/response, allowing for multiple messages to be sent or received over a single connection.

Cloud Client Libraries for Java support three types of streaming calls:

• Server streaming: The server sends a stream of responses back to you.
• Client streaming: You send a stream of requests to the server.
• Bidirectional streaming: You can send a stream of requests to the server, and the server can send back a stream of responses to you.

The streaming implementations are modeled after gRPC-Java implementations for server, client, and bidirectional streaming.

Streaming support across transports

Streaming is fully supported when using gRPC, but only partially supported for HttpJson. See the following table for streaming support.

Unary calls (non-streaming) are supported for both gRPC and HttpJson.

Determining the type of streaming

To determine the streaming type of the call, check the returned Callable type:

• ServerStreamingCallable: Server streaming.
• ClientStreamingCallable: Client streaming.
• BidiStreamingCallable: Bidirectional streaming.

For example, using Java-Aiplatform and Java-Speech:

// Server Streaming
ServerStreamingCallable<ReadTensorboardBlobDataRequest, ReadTensorboardBlobDataResponse> callable = aiplatformClient.readTensorboardBlobDataCallable();

// Bidirectional Streaming
BidiStreamingCallable<StreamingRecognizeRequest, StreamingRecognizeResponse> callable = speechClient.streamingRecognizeCallable();

Making streaming calls

Making a streaming call will differ depending on whether you are using server streaming or bidirectional streaming.

Server streaming

Server streaming doesn't require additional implementation. The ServerStream class lets you iterate the stream of responses. Using Java-Maps-Routing as an example, the following shows how to call the Server Streaming API:

try (RoutesClient routesClient = RoutesClient.create()) {
  ServerStreamingCallable<ComputeRouteMatrixRequest, RouteMatrixElement> computeRouteMatrix =
    routesClient.computeRouteMatrixCallable();  
  ServerStream<RouteMatrixElement> stream = computeRouteMatrix.call(
    ComputeRouteMatrixRequest.newBuilder().build());
  for (RouteMatrixElement element : stream) {
    // Do something with response
  }
}

In this example, the client sends a single ComputeRouteMatrixRequest and receives a stream of responses.

Bidirectional streaming

Bidirectional streaming requires additional implementation to make the call. Using Java-Speech as an example, the following steps provide an example implementation to make a bidirectional streaming call.

First, implement the ResponseObserver interface using the following code as a guideline:

class BidiResponseObserver<T> implements ResponseObserver<T> {
  private final List<T> responses = new ArrayList<>();
  private final SettableApiFuture<List<T>> future = SettableApiFuture.create();

  @Override
  public void onStart(StreamController controller) {
    // no-op
  }

  @Override
  public void onResponse(T response) {
    responses.add(response);
  }

  @Override
  public void onError(Throwable t) {
    future.setException(t);
  }

  @Override
  public void onComplete() {
    future.set(responses);
  }

  public SettableApiFuture<List<T>> getFuture() {
    return future;
  }
}

Then, follow these steps:

1. Create an instance of the observer: java BidiResponseObserver<StreamingRecognizeResponse> responseObserver = new BidiResponseObserver<>();
2. Pass the observer into the callable: java ClientStream<EchoRequest> clientStream = speechClient.streamingRecognizeCallable().splitCall(responseObserver);
3. Send the requests to the server and close the stream when complete: java clientStream.send(StreamingRecognizeRequest.newBuilder().build()); clientStream.send(StreamingRecognizeRequest.newBuilder().build()); // ... other requests ... clientStream.send(StreamingRecognizeRequest.newBuilder().build()); clientStream.closeSend();

4. Iterate through the responses: ```java List responses = responseObserver.getFuture().get();

   for (StreamingRecognizeResponse response : responses) { // Do something with response } ```

Non-supported streaming errors

For client libraries that support both gRPC and HTTP/JSON transports, it's possible to accidentally configure the client library to invoke a non-supported streaming call. For example, the following configuration shows the Java-Speech's HttpJson client making a bidirectional streaming call:

// SpeechClient is configured to use HttpJson
try (SpeechClient speechClient = SpeechClient.create(SpeechSettings.newHttpJsonBuilder().build())) {
  // Bidi Callable is not supported in HttpJson
  BidiStreamingCallable<StreamingRecognizeRequest, StreamingRecognizeResponse> callable = speechClient.streamingRecognizeCallable();
  ...
}

This doesn't result in a compilation error, but shows up as a runtime error:

Not implemented: streamingRecognizeCallable(). REST transport is not implemented for this method yet.
Important: The client library MUST be configured with gRPC to use client or bidirectional streaming.