Chiamate di streaming nelle librerie client di Cloud per Java

Le chiamate di streaming consentono pattern di interazione più complessi rispetto alla semplice richiesta/risposta, consentendo l'invio o la ricezione di più messaggi su una singola connessione.

Le librerie client di Cloud per Java supportano tre tipi di chiamate di streaming:

• Streaming del server: il server ti invia un flusso di risposte.
• Streaming del client: invii un flusso di richieste al server.
• Streaming bidirezionale: puoi inviare un flusso di richieste al server e il server può inviarti un flusso di risposte.

Le implementazioni di streaming sono modellate in base alle implementazioni gRPC-Java per lo streaming del server, del client e bidirezionale.

Supporto dello streaming tra i trasporti

Lo streaming è completamente supportato quando si utilizza gRPC, ma solo parzialmente supportato per HttpJson. Consulta la seguente tabella per il supporto dello streaming.

Le chiamate unarie (non di streaming) sono supportate sia per gRPC sia per HttpJson.

Determinare il tipo di streaming

Per determinare il tipo di streaming della chiamata, controlla il tipo Callable restituito:

• ServerStreamingCallable: streaming del server.
• ClientStreamingCallable: streaming del client.
• BidiStreamingCallable: streaming bidirezionale.

Ad esempio, utilizzando Java-Aiplatform e Java-Speech:

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

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

Effettuare chiamate di streaming

L'esecuzione di una chiamata di streaming varia a seconda che tu stia utilizzando lo streaming del server o lo streaming bidirezionale.

Streaming del server

Lo streaming del server non richiede un'implementazione aggiuntiva. La classe ServerStream consente di scorrere il flusso di risposte. Utilizzando Java-Maps-Routing come esempio, di seguito viene illustrato come chiamare l'API Server Streaming:

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 questo esempio, il client invia una singola ComputeRouteMatrixRequest e riceve un flusso di risposte.

Streaming bidirezionale

Lo streaming bidirezionale richiede un'implementazione aggiuntiva per effettuare la chiamata. Utilizzando Java-Speech come esempio, i seguenti passaggi forniscono un'implementazione di esempio per effettuare una chiamata di streaming bidirezionale.

Innanzitutto, implementa l'interfaccia ResponseObserver utilizzando il seguente codice come linea guida:

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;
  }
}

Quindi, segui questi passaggi:

1. Crea un'istanza dell'osservatore:

   BidiResponseObserver<StreamingRecognizeResponse> responseObserver = new BidiResponseObserver<>();
   

2. Passa l'osservatore al callable:

   ClientStream<EchoRequest> clientStream = speechClient.streamingRecognizeCallable().splitCall(responseObserver);
   

3. Invia le richieste al server e chiudi lo stream al termine:

   clientStream.send(StreamingRecognizeRequest.newBuilder().build());
   clientStream.send(StreamingRecognizeRequest.newBuilder().build());
   // ... other requests ...
   clientStream.send(StreamingRecognizeRequest.newBuilder().build());
   clientStream.closeSend();
   

4. Scorri le risposte:

   List<StreamingRecognizeResponse> responses = responseObserver.getFuture().get();
   
   for (StreamingRecognizeResponse response : responses) {
     // Do something with response
   }
   

Errori di streaming non supportati

Per le librerie client che supportano sia i trasporti gRPC sia HTTP/JSON, è possibile configurare accidentalmente la libreria client per richiamare una chiamata di streaming non supportata. Ad esempio, la seguente configurazione mostra il client HttpJson di Java-Speech che effettua una chiamata di streaming bidirezionale:

// 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();
  ...
}

Non viene generato un errore di compilazione, ma viene visualizzato un errore di runtime:

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.