Any Help Guys ?

Thanks
Avinash


On Thursday, September 15, 2016 at 12:53:04 PM UTC+5:30, Avinash Dongre 
wrote:
>
> I have client which make a request to server and Server in reply sends a 
> Stream-reply, This works fine for small number of message but when the 
> number of messages are large then My client is shutdown before I could 
> receive all the message.
>
> Following is how I have implemented Server RPC Streaming method.
>
> ScanReply.Builder scanReplyBuilder = ScanReply.newBuilder();
> final ServerCallStreamObserver<ScanReply> scso =
>     (ServerCallStreamObserver<ScanReply>) responseObserver;
>
> Runnable drain = new Runnable() {
>   long remaining = 20_000_000L;
>
>   public void run() {
>     if (remaining == 0L) return;
>     for (; remaining > 0L && scso.isReady(); remaining--) {
>       scanReplyBuilder.clear();
>       scanReplyBuilder.setRow(ByteString.copyFrom(new byte[128]));
>       scso.onNext(scanReplyBuilder.build());
>     }
>     if (remaining == 0) {
>       scso.onCompleted();
>     }
>   }
> };
> scso.setOnReadyHandler(drain);
> drain.run();
>
>
> and Client code is implemented as Follows, this 
>
>
>     final CountDownLatch finishLatch = new CountDownLatch(1);
>     int rowsReceived = 0;
>     final List<Long> rowList = new ArrayList<>();
>     ScanRequest request = ScanRequest.newBuilder()
>         .setTableName(tableName)
>         .setStartKey(ByteString.copyFrom(startRow))
>         .setStopKey(ByteString.copyFrom(stopRow))
>         .build();
>
>     StreamObserver<ScanReply> responseObserver = new 
> StreamObserver<ScanReply>() {
>       @Override
>       public void onNext(ScanReply value) {
>         byte[] row = value.getRow().toByteArray();
>         rowList.add(1L);
>       }
>
>       @Override
>       public void onError(Throwable t) {
>         finishLatch.countDown();
>       }
>
>       @Override
>       public void onCompleted() {
>         finishLatch.countDown();
>         System.out.println("TableServiceClient.onCompleted.SIZE --> " + 
> rowList.size());
>       }
>     };
>     this.valuesList.get(0).tableScan(request, responseObserver);
>     try {
>       finishLatch.await(1, TimeUnit.MINUTES);
>     } catch (InterruptedException e) {
>       e.printStackTrace();
>     }
>
> Thanks
>
> Avinash
>
>

-- 
You received this message because you are subscribed to the Google Groups 
"grpc.io" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to grpc-io+unsubscr...@googlegroups.com.
To post to this group, send email to grpc-io@googlegroups.com.
Visit this group at https://groups.google.com/group/grpc-io.
To view this discussion on the web visit 
https://groups.google.com/d/msgid/grpc-io/5a1e18e9-9c64-4bc0-b3cc-f9e1d7ecc83e%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

Reply via email to