InputStreamHttpBody implements HttpBody { @Override public ChannelFuture writeContent(ChannelHandlerContext ctx, boolean isSecure) throws IOException { final InputStream is = inputStream; Channel channel = ctx.channel(); ChannelFuture channelFuture = channel.write(new ChunkedStream(inputStream), channel.newProgressivePromise()); channelFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { is.close(); } }); return channelFuture; } InputStreamHttpBody(InputStream inputStream, String contentType, Charset characterEncoding); @Override long getContentLength(); @Override String getContentType(); @Override Charset getCharacterEncoding(); @Override ChannelFuture writeContent(ChannelHandlerContext ctx, boolean isSecure);  }
@Test public void operationCompletedShouldCloseInputStream() throws Exception { ChannelHandlerContext ctx = mock(ChannelHandlerContext.class); Channel channel = mock(Channel.class); when(ctx.channel()).thenReturn(channel); ChannelFuture channelFuture = mock(ChannelFuture.class); when(channel.write(any(), any())).thenReturn(channelFuture); when(channelFuture.addListener(any(ChannelFutureListener.class))).thenAnswer((Answer<ChannelFuture>) invocation -> { ChannelFutureListener channelFutureListener = invocation.getArgument(0); ChannelFuture future = mock(ChannelFuture.class); channelFutureListener.operationComplete(future); return future; }); inputStreamHttpBody.writeContent(ctx, false); verify(inputStream).close(); }