NettyClientHandler extends AbstractNettyHandler { private void createStream(CreateStreamCommand command, final ChannelPromise promise) throws Exception { if (lifecycleManager.getShutdownThrowable() != null) { promise.setFailure(lifecycleManager.getShutdownThrowable()); return; } final int streamId; try { streamId = incrementAndGetNextStreamId(); } catch (StatusException e) { promise.setFailure(e); if (!connection().goAwaySent()) { logger.fine("Stream IDs have been exhausted for this connection. " + "Initiating graceful shutdown of the connection."); super.close(ctx(), ctx().newPromise()); } return; } final NettyClientStream stream = command.stream(); final Http2Headers headers = command.headers(); stream.id(streamId); ChannelPromise tempPromise = ctx().newPromise(); encoder().writeHeaders(ctx(), streamId, headers, 0, false, tempPromise) .addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { if (future.isSuccess()) { Http2Stream http2Stream = connection().stream(streamId); if (http2Stream != null) { http2Stream.setProperty(streamKey, stream); stream.setHttp2Stream(http2Stream); } promise.setSuccess(); } else { final Throwable cause = future.cause(); if (cause instanceof StreamBufferingEncoder.Http2GoAwayException) { StreamBufferingEncoder.Http2GoAwayException e = (StreamBufferingEncoder.Http2GoAwayException) cause; lifecycleManager.notifyShutdown(statusFromGoAway(e.errorCode(), e.debugData())); promise.setFailure(lifecycleManager.getShutdownThrowable()); } else { promise.setFailure(cause); } } } }); } private  NettyClientHandler(Http2ConnectionDecoder decoder,
                             StreamBufferingEncoder encoder, Http2Settings settings,
                             ClientTransportLifecycleManager lifecycleManager,
                             Ticker ticker); @Override void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise); @Override void close(ChannelHandlerContext ctx, ChannelPromise promise); @Override void channelInactive(ChannelHandlerContext ctx);  }
@Test public void createStreamShouldSucceed() throws Exception { createStream(); verifyWrite().writeHeaders(eq(ctx()), eq(3), eq(grpcHeaders), eq(0), eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(false), any(ChannelPromise.class)); }
@Test public void sendFrameShouldSucceed() throws Exception { createStream(); ChannelFuture future = enqueue(new SendGrpcFrameCommand(stream, content(), true)); assertTrue(future.isSuccess()); verifyWrite().writeData(eq(ctx()), eq(3), eq(content()), eq(0), eq(true), any(ChannelPromise.class)); }
@Test public void inboundHeadersShouldForwardToStream() throws Exception { createStream(); Http2Headers headers = new DefaultHttp2Headers().status(STATUS_OK) .set(CONTENT_TYPE_HEADER, CONTENT_TYPE_GRPC); ByteBuf headersFrame = headersFrame(3, headers); channelRead(headersFrame); verify(stream).transportHeadersReceived(headers, false); }
@Test public void inboundDataShouldForwardToStream() throws Exception { ByteBuf data = content().copy(); createStream(); ByteBuf frame = dataFrame(3, false).retain(); channelRead(frame); ArgumentCaptor<ByteBuf> captor = ArgumentCaptor.forClass(ByteBuf.class); verify(stream).transportDataReceived(captor.capture(), eq(false)); assertTrue(ByteBufUtil.equals(data, captor.getValue())); }
@Test public void exhaustedStreamsShouldFail() throws Exception { streamId = Integer.MAX_VALUE; setUp(); ChannelFuture future = createStream(); assertTrue(future.isSuccess()); future = createStream(); assertTrue(future.isDone()); assertFalse(future.isSuccess()); }