@Test public void receiveMoreThanBufferSize() throws Exception { WriteResponse[] responses = Stream.generate(() -> WriteResponse.newBuilder().build()) .limit(BUFFER_SIZE * 2).toArray(WriteResponse[]::new); EXECUTOR.submit(() -> { for (WriteResponse response : responses) { mResponseObserver.onNext(response); } }); Thread.sleep(SHORT_TIMEOUT); for (WriteResponse response : responses) { WriteResponse actualResponse = mStream.receive(TIMEOUT); assertEquals(response, actualResponse); } }
public ResT receive(long timeoutMs) throws IOException { if (mCompleted) { return null; } if (mCanceled) { throw new CancelledException(formatErrorMessage("Stream is already canceled.")); } try { Object response = mResponses.poll(timeoutMs, TimeUnit.MILLISECONDS); if (response == null) { checkError(); DeadlineExceededException e = new DeadlineExceededException( formatErrorMessage("Timeout waiting for response after %dms.", timeoutMs)); throw e; } if (response == mResponseObserver) { mCompleted = true; return null; } checkError(); return (ResT) response; } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new CancelledException( formatErrorMessage("Interrupted while waiting for response."), e); } }
GrpcBlockingStream { public ResT receive(long timeoutMs) throws IOException { if (mCompleted) { return null; } if (mCanceled) { throw new CancelledException(formatErrorMessage("Stream is already canceled.")); } try { Object response = mResponses.poll(timeoutMs, TimeUnit.MILLISECONDS); if (response == null) { checkError(); DeadlineExceededException e = new DeadlineExceededException( formatErrorMessage("Timeout waiting for response after %dms.", timeoutMs)); throw e; } if (response == mResponseObserver) { mCompleted = true; return null; } checkError(); return (ResT) response; } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new CancelledException( formatErrorMessage("Interrupted while waiting for response."), e); } } }
GrpcBlockingStream { public ResT receive(long timeoutMs) throws IOException { if (mCompleted) { return null; } if (mCanceled) { throw new CancelledException(formatErrorMessage("Stream is already canceled.")); } try { Object response = mResponses.poll(timeoutMs, TimeUnit.MILLISECONDS); if (response == null) { checkError(); DeadlineExceededException e = new DeadlineExceededException( formatErrorMessage("Timeout waiting for response after %dms.", timeoutMs)); throw e; } if (response == mResponseObserver) { mCompleted = true; return null; } checkError(); return (ResT) response; } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new CancelledException( formatErrorMessage("Interrupted while waiting for response."), e); } } GrpcBlockingStream(Function<StreamObserver<ResT>, StreamObserver<ReqT>> rpcFunc,
      int bufferSize, String description); }
GrpcBlockingStream { public ResT receive(long timeoutMs) throws IOException { if (mCompleted) { return null; } if (mCanceled) { throw new CancelledException(formatErrorMessage("Stream is already canceled.")); } try { Object response = mResponses.poll(timeoutMs, TimeUnit.MILLISECONDS); if (response == null) { checkError(); DeadlineExceededException e = new DeadlineExceededException( formatErrorMessage("Timeout waiting for response after %dms.", timeoutMs)); throw e; } if (response == mResponseObserver) { mCompleted = true; return null; } checkError(); return (ResT) response; } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new CancelledException( formatErrorMessage("Interrupted while waiting for response."), e); } } GrpcBlockingStream(Function<StreamObserver<ResT>, StreamObserver<ReqT>> rpcFunc,
      int bufferSize, String description); void send(ReqT request, long timeoutMs); void send(ReqT request); ResT receive(long timeoutMs); void close(); void cancel(); void waitForComplete(long timeoutMs); boolean isClosedFromRemote(); boolean isOpen(); boolean isClosed(); boolean isCanceled(); }
GrpcBlockingStream { public ResT receive(long timeoutMs) throws IOException { if (mCompleted) { return null; } if (mCanceled) { throw new CancelledException(formatErrorMessage("Stream is already canceled.")); } try { Object response = mResponses.poll(timeoutMs, TimeUnit.MILLISECONDS); if (response == null) { checkError(); DeadlineExceededException e = new DeadlineExceededException( formatErrorMessage("Timeout waiting for response after %dms.", timeoutMs)); throw e; } if (response == mResponseObserver) { mCompleted = true; return null; } checkError(); return (ResT) response; } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new CancelledException( formatErrorMessage("Interrupted while waiting for response."), e); } } GrpcBlockingStream(Function<StreamObserver<ResT>, StreamObserver<ReqT>> rpcFunc,
      int bufferSize, String description); void send(ReqT request, long timeoutMs); void send(ReqT request); ResT receive(long timeoutMs); void close(); void cancel(); void waitForComplete(long timeoutMs); boolean isClosedFromRemote(); boolean isOpen(); boolean isClosed(); boolean isCanceled();  }