Panggilan streaming di Library Klien Cloud untuk Java

Panggilan streaming memungkinkan pola interaksi yang lebih kompleks daripada permintaan/respons sederhana, sehingga memungkinkan beberapa pesan dikirim atau diterima melalui satu koneksi.

Library Klien Cloud untuk Java mendukung tiga jenis panggilan streaming:

• Streaming server: Server mengirimkan aliran respons kembali kepada Anda.
• Streaming klien: Anda mengirimkan aliran permintaan ke server.
• Streaming dua arah: Anda dapat mengirimkan aliran permintaan ke server, dan server dapat mengirimkan kembali aliran respons kepada Anda.

Implementasi streaming dimodelkan setelah implementasi gRPC-Java untuk streaming server, klien, dan dua arah.

Dukungan streaming di seluruh transport

Streaming didukung sepenuhnya saat menggunakan gRPC, tetapi hanya didukung sebagian untuk HttpJson. Lihat tabel berikut untuk dukungan streaming.

Panggilan unary (non-streaming) didukung untuk gRPC dan HttpJson.

Menentukan jenis streaming

Untuk menentukan jenis streaming panggilan, periksa jenis Callable yang ditampilkan:

• ServerStreamingCallable: Streaming server.
• ClientStreamingCallable: Streaming klien.
• BidiStreamingCallable: Streaming dua arah.

Misalnya, menggunakan Java-Aiplatform dan Java-Speech:

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

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

Melakukan panggilan streaming

Melakukan panggilan streaming akan berbeda, bergantung pada apakah Anda menggunakan streaming server atau streaming dua arah.

Streaming server

Streaming server tidak memerlukan implementasi tambahan. Class ServerStream memungkinkan Anda melakukan iterasi aliran respons. Dengan menggunakan Java-Maps-Routing sebagai contoh, berikut ini cara memanggil 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
  }
}

Dalam contoh ini, klien mengirimkan satu ComputeRouteMatrixRequest dan menerima aliran respons.

Streaming dua arah

Streaming dua arah memerlukan implementasi tambahan untuk melakukan panggilan. Dengan menggunakan Java-Speech sebagai contoh, langkah-langkah berikut memberikan contoh implementasi untuk melakukan panggilan streaming dua arah.

Pertama, terapkan antarmuka ResponseObserver menggunakan kode berikut sebagai panduan:

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

Lalu, ikuti langkah-langkah berikut:

1. Buat instance observer:

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

2. Teruskan observer ke callable:

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

3. Kirim permintaan ke server dan tutup aliran setelah selesai:

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

4. Lakukan iterasi melalui respons:

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

Error streaming yang tidak didukung

Untuk library klien yang mendukung transport gRPC dan HTTP/JSON, Anda dapat secara tidak sengaja mengonfigurasi library klien untuk memanggil panggilan streaming yang tidak didukung. Misalnya, konfigurasi berikut menunjukkan klien HttpJson Java-Speech melakukan panggilan streaming dua arah:

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

Hal ini tidak akan menghasilkan error kompilasi, tetapi akan muncul sebagai error 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.