@Test public void readEntitySentWithMultipleChunks() throws Exception { when(route.apply(any())).thenReturn(Observable.just(new ResponseBuilderImpl().ok())); ExecutorService service = Executors.newSingleThreadExecutor(); ClientToServerChannelHandler handler = new ClientToServerChannelHandler(service, route, null); DefaultHttpRequest request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "localhost"); handler.channelRead(context, request); DefaultHttpContent content = new DefaultHttpContent(Unpooled.wrappedBuffer("Test Content ".getBytes())); handler.channelRead(context, content); content = new DefaultHttpContent(Unpooled.wrappedBuffer("Additional Content".getBytes())); handler.channelRead(context, content); DefaultLastHttpContent last = new DefaultLastHttpContent(); handler.channelRead(context, last); waitForExecutorToFinishAllTasks(service); verify(route).apply(requestCaptor.capture()); ServerRequest appliedRequest = requestCaptor.getValue(); assertEquals("Test Content Additional Content", appliedRequest.body().asString().toBlocking().last()); }
@Override public void channelRead(ChannelHandlerContext context, Object message) { if (message instanceof HttpRequest) { HttpRequest request = (HttpRequest) message; if (HttpUtil.is100ContinueExpected(request)) { sendContinueResponse(context); } bodyStream = ReplaySubject.create(); serverRequest = ServerRequestBuilder.buildServerRequest(request, bodyStream); processRequest(context, request); if (request.decoderResult().isFailure()) { bodyStream.onError(request.decoderResult().cause()); } } if (message instanceof HttpContent) { HttpContent httpContent = (HttpContent) message; ByteBuf content = httpContent.content(); if (content.isReadable()) { bodyStream.onNext(content.nioBuffer()); if (httpContent.decoderResult().isFailure()) { bodyStream.onError(httpContent.decoderResult().cause()); } } if (message instanceof LastHttpContent) { LastHttpContent trailer = (LastHttpContent) message; if (!trailer.trailingHeaders().isEmpty()) { serverRequest.setTrailingHeaders(ServerRequestBuilder.buildHeadersMap(trailer.trailingHeaders())); } bodyStream.onCompleted(); } } }
ClientToServerChannelHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext context, Object message) { if (message instanceof HttpRequest) { HttpRequest request = (HttpRequest) message; if (HttpUtil.is100ContinueExpected(request)) { sendContinueResponse(context); } bodyStream = ReplaySubject.create(); serverRequest = ServerRequestBuilder.buildServerRequest(request, bodyStream); processRequest(context, request); if (request.decoderResult().isFailure()) { bodyStream.onError(request.decoderResult().cause()); } } if (message instanceof HttpContent) { HttpContent httpContent = (HttpContent) message; ByteBuf content = httpContent.content(); if (content.isReadable()) { bodyStream.onNext(content.nioBuffer()); if (httpContent.decoderResult().isFailure()) { bodyStream.onError(httpContent.decoderResult().cause()); } } if (message instanceof LastHttpContent) { LastHttpContent trailer = (LastHttpContent) message; if (!trailer.trailingHeaders().isEmpty()) { serverRequest.setTrailingHeaders(ServerRequestBuilder.buildHeadersMap(trailer.trailingHeaders())); } bodyStream.onCompleted(); } } } }
ClientToServerChannelHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext context, Object message) { if (message instanceof HttpRequest) { HttpRequest request = (HttpRequest) message; if (HttpUtil.is100ContinueExpected(request)) { sendContinueResponse(context); } bodyStream = ReplaySubject.create(); serverRequest = ServerRequestBuilder.buildServerRequest(request, bodyStream); processRequest(context, request); if (request.decoderResult().isFailure()) { bodyStream.onError(request.decoderResult().cause()); } } if (message instanceof HttpContent) { HttpContent httpContent = (HttpContent) message; ByteBuf content = httpContent.content(); if (content.isReadable()) { bodyStream.onNext(content.nioBuffer()); if (httpContent.decoderResult().isFailure()) { bodyStream.onError(httpContent.decoderResult().cause()); } } if (message instanceof LastHttpContent) { LastHttpContent trailer = (LastHttpContent) message; if (!trailer.trailingHeaders().isEmpty()) { serverRequest.setTrailingHeaders(ServerRequestBuilder.buildHeadersMap(trailer.trailingHeaders())); } bodyStream.onCompleted(); } } } ClientToServerChannelHandler(
            ExecutorService threadPool,
            Route route,
            BiFunction<ServerRequest, Throwable, Observable<Response>> errorResponseConstructor); }
ClientToServerChannelHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext context, Object message) { if (message instanceof HttpRequest) { HttpRequest request = (HttpRequest) message; if (HttpUtil.is100ContinueExpected(request)) { sendContinueResponse(context); } bodyStream = ReplaySubject.create(); serverRequest = ServerRequestBuilder.buildServerRequest(request, bodyStream); processRequest(context, request); if (request.decoderResult().isFailure()) { bodyStream.onError(request.decoderResult().cause()); } } if (message instanceof HttpContent) { HttpContent httpContent = (HttpContent) message; ByteBuf content = httpContent.content(); if (content.isReadable()) { bodyStream.onNext(content.nioBuffer()); if (httpContent.decoderResult().isFailure()) { bodyStream.onError(httpContent.decoderResult().cause()); } } if (message instanceof LastHttpContent) { LastHttpContent trailer = (LastHttpContent) message; if (!trailer.trailingHeaders().isEmpty()) { serverRequest.setTrailingHeaders(ServerRequestBuilder.buildHeadersMap(trailer.trailingHeaders())); } bodyStream.onCompleted(); } } } ClientToServerChannelHandler(
            ExecutorService threadPool,
            Route route,
            BiFunction<ServerRequest, Throwable, Observable<Response>> errorResponseConstructor); @Override void channelInactive(ChannelHandlerContext context); @Override void channelReadComplete(ChannelHandlerContext context); @Override void channelRead(ChannelHandlerContext context, Object message); @Override void exceptionCaught(ChannelHandlerContext context, Throwable cause); }
ClientToServerChannelHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext context, Object message) { if (message instanceof HttpRequest) { HttpRequest request = (HttpRequest) message; if (HttpUtil.is100ContinueExpected(request)) { sendContinueResponse(context); } bodyStream = ReplaySubject.create(); serverRequest = ServerRequestBuilder.buildServerRequest(request, bodyStream); processRequest(context, request); if (request.decoderResult().isFailure()) { bodyStream.onError(request.decoderResult().cause()); } } if (message instanceof HttpContent) { HttpContent httpContent = (HttpContent) message; ByteBuf content = httpContent.content(); if (content.isReadable()) { bodyStream.onNext(content.nioBuffer()); if (httpContent.decoderResult().isFailure()) { bodyStream.onError(httpContent.decoderResult().cause()); } } if (message instanceof LastHttpContent) { LastHttpContent trailer = (LastHttpContent) message; if (!trailer.trailingHeaders().isEmpty()) { serverRequest.setTrailingHeaders(ServerRequestBuilder.buildHeadersMap(trailer.trailingHeaders())); } bodyStream.onCompleted(); } } } ClientToServerChannelHandler(
            ExecutorService threadPool,
            Route route,
            BiFunction<ServerRequest, Throwable, Observable<Response>> errorResponseConstructor); @Override void channelInactive(ChannelHandlerContext context); @Override void channelReadComplete(ChannelHandlerContext context); @Override void channelRead(ChannelHandlerContext context, Object message); @Override void exceptionCaught(ChannelHandlerContext context, Throwable cause);  }