Appels de flux sur les bibliothèques clientes Cloud pour Java

Les appels de streaming permettent des schémas d'interaction plus complexes que les simples requêtes/réponses, ce qui permet d'envoyer ou de recevoir plusieurs messages sur une même connexion.

Les bibliothèques clientes Cloud pour Java sont compatibles avec trois types d'appels de streaming :

  • Streaming côté serveur : le serveur vous renvoie un flux de réponses.
  • Streaming côté client : vous envoyez un flux de requêtes au serveur.
  • Streaming bidirectionnel : vous pouvez envoyer un flux de requêtes au serveur, et le serveur peut vous renvoyer un flux de réponses.

Les implémentations de streaming sont modélisées d'après les implémentations gRPC-Java pour le streaming serveur, client et bidirectionnel.

Compatibilité avec le streaming sur différents transports

Le streaming est entièrement compatible avec gRPC, mais seulement partiellement avec HttpJson. Consultez le tableau ci-dessous pour connaître les services de streaming compatibles.

Type de streaming gRPC HttpJson
Streaming côté serveur Compatible Compatible
Streaming client Compatible Non compatible
Streaming bidirectionnel Compatible Non compatible

Les appels unaires (non en streaming) sont acceptés pour gRPC et HttpJson.

Déterminer le type de streaming

Pour déterminer le type de flux de l'appel, vérifiez le type Callable renvoyé :

  • ServerStreamingCallable : streaming sur le serveur.
  • ClientStreamingCallable : flux client.
  • BidiStreamingCallable : streaming bidirectionnel.

Par exemple, en utilisant Java-Aiplatform et Java-Speech :

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

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

Effectuer des appels de streaming

La procédure pour effectuer un appel de streaming diffère selon que vous utilisez le streaming serveur ou le streaming bidirectionnel.

Streaming côté serveur

Le streaming côté serveur ne nécessite aucune implémentation supplémentaire. La classe ServerStream vous permet d'itérer le flux de réponses. En utilisant Java-Maps-Routing comme exemple, voici comment appeler 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
  }
}

Dans cet exemple, le client envoie un seul ComputeRouteMatrixRequest et reçoit un flux de réponses.

Streaming bidirectionnel

Le streaming bidirectionnel nécessite une implémentation supplémentaire pour effectuer l'appel. En utilisant Java-Speech comme exemple, les étapes suivantes fournissent un exemple d'implémentation pour effectuer un appel de streaming bidirectionnel.

Commencez par implémenter l'interface ResponseObserver en utilisant le code suivant comme guide :

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

Ensuite, procédez comme suit :

  1. Créez une instance de l'observateur : java BidiResponseObserver<StreamingRecognizeResponse> responseObserver = new BidiResponseObserver<>();
  2. Transmettez l'observateur à l'appelable : java ClientStream<EchoRequest> clientStream = speechClient.streamingRecognizeCallable().splitCall(responseObserver);
  3. Envoyez les requêtes au serveur et fermez le flux une fois l'opération terminée : java clientStream.send(StreamingRecognizeRequest.newBuilder().build()); clientStream.send(StreamingRecognizeRequest.newBuilder().build()); // ... other requests ... clientStream.send(StreamingRecognizeRequest.newBuilder().build()); clientStream.closeSend();

  4. Parcourez les réponses : ```java List responses = responseObserver.getFuture().get();

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

Erreurs de streaming non prises en charge

Pour les bibliothèques clientes compatibles avec les transports gRPC et HTTP/JSON, il est possible de configurer accidentellement la bibliothèque cliente pour qu'elle appelle un appel de streaming non compatible. Par exemple, la configuration suivante montre le client HttpJson de Java-Speech effectuant un appel de streaming bidirectionnel :

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

Cela n'entraîne pas d'erreur de compilation, mais une erreur d'exécution : 

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.