DefaultHttp2ConnectionEncoder implements Http2ConnectionEncoder { @Override public ChannelFuture writeData(final ChannelHandlerContext ctx, final int streamId, ByteBuf data, int padding, final boolean endOfStream, ChannelPromise promise) { final Http2Stream stream; try { stream = requireStream(streamId); switch (stream.state()) { case OPEN: case HALF_CLOSED_REMOTE: break; default: throw new IllegalStateException("Stream " + stream.id() + " in unexpected state " + stream.state()); } } catch (Throwable e) { data.release(); return promise.setFailure(e); } flowController().addFlowControlled(stream, new FlowControlledData(stream, data, padding, endOfStream, promise)); return promise; } DefaultHttp2ConnectionEncoder(Http2Connection connection, Http2FrameWriter frameWriter); @Override void lifecycleManager(Http2LifecycleManager lifecycleManager); @Override Http2FrameWriter frameWriter(); @Override Http2Connection connection(); @Override final Http2RemoteFlowController flowController(); @Override void remoteSettings(Http2Settings settings); @Override ChannelFuture writeData(final ChannelHandlerContext ctx, final int streamId, ByteBuf data, int padding,
            final boolean endOfStream, ChannelPromise promise); @Override ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
            boolean endStream, ChannelPromise promise); @Override ChannelFuture writeHeaders(final ChannelHandlerContext ctx, final int streamId,
            final Http2Headers headers, final int streamDependency, final short weight,
            final boolean exclusive, final int padding, final boolean endOfStream, ChannelPromise promise); @Override ChannelFuture writePriority(ChannelHandlerContext ctx, int streamId, int streamDependency, short weight,
            boolean exclusive, ChannelPromise promise); @Override ChannelFuture writeRstStream(ChannelHandlerContext ctx, int streamId, long errorCode,
            ChannelPromise promise); @Override ChannelFuture writeSettings(ChannelHandlerContext ctx, Http2Settings settings,
            ChannelPromise promise); @Override ChannelFuture writeSettingsAck(ChannelHandlerContext ctx, ChannelPromise promise); @Override ChannelFuture writePing(ChannelHandlerContext ctx, boolean ack, ByteBuf data, ChannelPromise promise); @Override ChannelFuture writePushPromise(ChannelHandlerContext ctx, int streamId, int promisedStreamId,
            Http2Headers headers, int padding, ChannelPromise promise); @Override ChannelFuture writeGoAway(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData,
            ChannelPromise promise); @Override ChannelFuture writeWindowUpdate(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement,
            ChannelPromise promise); @Override ChannelFuture writeFrame(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags,
            ByteBuf payload, ChannelPromise promise); @Override void close(); @Override Http2Settings pollSentSettings(); @Override Configuration configuration();  }
@Test public void dataWriteShouldSucceed() throws Exception { createStream(STREAM_ID, false); final ByteBuf data = dummyData(); ChannelPromise p = newPromise(); encoder.writeData(ctx, STREAM_ID, data, 0, true, p); assertEquals(8, payloadCaptor.getValue().size()); payloadCaptor.getValue().write(ctx, 8); assertEquals(0, payloadCaptor.getValue().size()); assertEquals("abcdefgh", writtenData.get(0)); assertEquals(0, data.refCnt()); assertTrue(p.isSuccess()); }
@Test public void dataFramesShouldMerge() throws Exception { createStream(STREAM_ID, false); final ByteBuf data = dummyData().retain(); ChannelPromise promise1 = newPromise(); encoder.writeData(ctx, STREAM_ID, data, 0, true, promise1); ChannelPromise promise2 = newPromise(); encoder.writeData(ctx, STREAM_ID, data, 0, true, promise2); List<FlowControlled> capturedWrites = payloadCaptor.getAllValues(); FlowControlled mergedPayload = capturedWrites.get(0); mergedPayload.merge(ctx, capturedWrites.get(1)); assertEquals(16, mergedPayload.size()); assertFalse(promise1.isDone()); assertFalse(promise2.isDone()); mergedPayload.write(ctx, 16); assertEquals(0, mergedPayload.size()); assertEquals("abcdefghabcdefgh", writtenData.get(0)); assertEquals(0, data.refCnt()); assertTrue(promise1.isSuccess()); assertTrue(promise2.isSuccess()); }
@Test public void dataFramesShouldMergeUseVoidPromise() throws Exception { createStream(STREAM_ID, false); final ByteBuf data = dummyData().retain(); ChannelPromise promise1 = newVoidPromise(channel); encoder.writeData(ctx, STREAM_ID, data, 0, true, promise1); ChannelPromise promise2 = newVoidPromise(channel); encoder.writeData(ctx, STREAM_ID, data, 0, true, promise2); List<FlowControlled> capturedWrites = payloadCaptor.getAllValues(); FlowControlled mergedPayload = capturedWrites.get(0); mergedPayload.merge(ctx, capturedWrites.get(1)); assertEquals(16, mergedPayload.size()); assertFalse(promise1.isSuccess()); assertFalse(promise2.isSuccess()); mergedPayload.write(ctx, 16); assertEquals(0, mergedPayload.size()); assertEquals("abcdefghabcdefgh", writtenData.get(0)); assertEquals(0, data.refCnt()); assertFalse(promise1.isSuccess()); assertFalse(promise2.isSuccess()); }
@Test public void dataWriteShouldCreateHalfClosedStream() throws Exception { writeAllFlowControlledFrames(); Http2Stream stream = createStream(STREAM_ID, false); ByteBuf data = dummyData(); ChannelPromise promise = newPromise(); encoder.writeData(ctx, STREAM_ID, data.retain(), 0, true, promise); assertTrue(promise.isSuccess()); verify(remoteFlow).addFlowControlled(eq(stream), any(FlowControlled.class)); verify(lifecycleManager).closeStreamLocal(stream, promise); assertEquals(data.toString(UTF_8), writtenData.get(0)); data.release(); }
@Test public void dataWriteToHalfClosedLocalStreamShouldFail() throws Exception { createStream(STREAM_ID, true); ByteBuf data = mock(ByteBuf.class); ChannelPromise promise = newPromise(); encoder.writeData(ctx, STREAM_ID, data, 0, false, promise); assertTrue(promise.isDone()); assertFalse(promise.isSuccess()); assertThat(promise.cause(), instanceOf(IllegalStateException.class)); verify(data).release(); }
@Test public void canWriteDataFrameAfterGoAwaySent() throws Exception { Http2Stream stream = createStream(STREAM_ID, false); connection.goAwaySent(0, 0, EMPTY_BUFFER); ByteBuf data = mock(ByteBuf.class); encoder.writeData(ctx, STREAM_ID, data, 0, false, newPromise()); verify(remoteFlow).addFlowControlled(eq(stream), any(FlowControlled.class)); }
@Test public void canWriteDataFrameAfterGoAwayReceived() throws Exception { Http2Stream stream = createStream(STREAM_ID, false); goAwayReceived(STREAM_ID); ByteBuf data = mock(ByteBuf.class); encoder.writeData(ctx, STREAM_ID, data, 0, false, newPromise()); verify(remoteFlow).addFlowControlled(eq(stream), any(FlowControlled.class)); }