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.