@Test public void streamSucceededWithGrpcError() { ArgumentCaptor<BidirectionalStream.Callback> callbackCaptor = ArgumentCaptor.forClass(BidirectionalStream.Callback.class); verify(factory) .newBidirectionalStreamBuilder( isA(String.class), callbackCaptor.capture(), isA(Executor.class)); BidirectionalStream.Callback callback = callbackCaptor.getValue(); callback.onStreamReady(cronetStream); verify(cronetStream, times(0)).write(isA(ByteBuffer.class), isA(Boolean.class)); clientStream.abstractClientStreamSink().writeFrame(null, true, true, 1); verify(cronetStream, times(1)).write(isA(ByteBuffer.class), isA(Boolean.class)); verify(cronetStream, times(1)).flush(); clientStream.request(2); UrlResponseInfo info = new UrlResponseInfoImpl( new ArrayList<String>(), 200, "", responseHeader("200"), false, "", ""); callback.onResponseHeadersReceived(cronetStream, info); verify(cronetStream, times(1)).read(isA(ByteBuffer.class)); callback.onReadCompleted(cronetStream, null, ByteBuffer.allocate(0), true); ((CronetClientStream.BidirectionalStreamCallback) callback) .processTrailers(trailers(Status.PERMISSION_DENIED.getCode().value())); callback.onSucceeded(cronetStream, info); ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class); verify(clientListener) .closed(statusCaptor.capture(), isA(RpcProgress.class), isA(Metadata.class)); Status status = statusCaptor.getValue(); assertFalse(status.isOk()); assertEquals(Status.PERMISSION_DENIED.getCode(), status.getCode()); }
@Override protected Sink abstractClientStreamSink() { return sink; }
CronetClientStream extends AbstractClientStream { @Override protected Sink abstractClientStreamSink() { return sink; } }
CronetClientStream extends AbstractClientStream { @Override protected Sink abstractClientStreamSink() { return sink; } CronetClientStream(
      final String url,
      @Nullable String userAgent,
      Executor executor,
      final Metadata headers,
      CronetClientTransport transport,
      Runnable startCallback,
      Object lock,
      int maxMessageSize,
      boolean alwaysUsePut,
      MethodDescriptor<?, ?> method,
      StatsTraceContext statsTraceCtx,
      CallOptions callOptions,
      TransportTracer transportTracer); }
CronetClientStream extends AbstractClientStream { @Override protected Sink abstractClientStreamSink() { return sink; } CronetClientStream(
      final String url,
      @Nullable String userAgent,
      Executor executor,
      final Metadata headers,
      CronetClientTransport transport,
      Runnable startCallback,
      Object lock,
      int maxMessageSize,
      boolean alwaysUsePut,
      MethodDescriptor<?, ?> method,
      StatsTraceContext statsTraceCtx,
      CallOptions callOptions,
      TransportTracer transportTracer); @Override void setAuthority(String authority); @Override Attributes getAttributes(); }
CronetClientStream extends AbstractClientStream { @Override protected Sink abstractClientStreamSink() { return sink; } CronetClientStream(
      final String url,
      @Nullable String userAgent,
      Executor executor,
      final Metadata headers,
      CronetClientTransport transport,
      Runnable startCallback,
      Object lock,
      int maxMessageSize,
      boolean alwaysUsePut,
      MethodDescriptor<?, ?> method,
      StatsTraceContext statsTraceCtx,
      CallOptions callOptions,
      TransportTracer transportTracer); @Override void setAuthority(String authority); @Override Attributes getAttributes();  }