@Test void returnsNewConfigsWhenChunked() throws Exception { CoreContext ctx = new CoreContext(mock(Core.class), 1, ENV, PasswordAuthenticator.create(USER, PASS)); BaseEndpoint endpoint = mock(BaseEndpoint.class); EndpointContext endpointContext = mock(EndpointContext.class); when(endpointContext.environment()).thenReturn(ENV); when(endpoint.context()).thenReturn(endpointContext); EmbeddedChannel channel = new EmbeddedChannel(new ManagerMessageHandler(endpoint, ctx)); BucketConfigStreamingRequest request = new BucketConfigStreamingRequest(Duration.ofSeconds(1), ctx, BestEffortRetryStrategy.INSTANCE, "bucket", ctx.authenticator()); channel.write(request); HttpRequest outboundHeader = channel.readOutbound(); assertEquals(HttpMethod.GET, outboundHeader.method()); assertEquals("/pools/default/bs/bucket", outboundHeader.uri()); assertEquals(HttpVersion.HTTP_1_1, outboundHeader.protocolVersion()); CompletableFuture<BucketConfigStreamingResponse> response = request.response(); assertFalse(response.isDone()); HttpResponse inboundResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK); channel.writeInbound(inboundResponse); BucketConfigStreamingResponse completedResponse = request.response().get(); final List<String> configsPushed = Collections.synchronizedList(new ArrayList<>()); final AtomicBoolean terminated = new AtomicBoolean(false); Thread listener = new Thread(() -> completedResponse.configs().subscribe( configsPushed::add, (e) -> {}, () -> terminated.set(true)) ); listener.setDaemon(true); listener.start(); ByteBuf fullContent = Unpooled.copiedBuffer( readResource("terse_stream_two_configs.json", ManagerMessageHandlerTest.class), StandardCharsets.UTF_8 ); while (fullContent.readableBytes() > 0) { int len = new Random().nextInt(fullContent.readableBytes() + 1); if (len == 0) { continue; } channel.writeInbound(new DefaultHttpContent(fullContent.readBytes(len))); } waitUntilCondition(() -> configsPushed.size() >= 1); for (String config : configsPushed) { assertTrue(config.startsWith("{")); assertTrue(config.endsWith("}")); } assertFalse(terminated.get()); channel.writeInbound(new DefaultLastHttpContent()); waitUntilCondition(terminated::get); ReferenceCountUtil.release(fullContent); channel.close().awaitUninterruptibly(); }
@Override @SuppressWarnings({ "unchecked" }) public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) { if (msg instanceof ManagerRequest) { if (currentRequest != null) { RetryOrchestrator.maybeRetry(coreContext, (ManagerRequest<Response>) msg, RetryReason.NOT_PIPELINED_REQUEST_IN_FLIGHT); if (endpoint != null) { endpoint.decrementOutstandingRequests(); } return; } try { currentRequest = (ManagerRequest<Response>) msg; FullHttpRequest encoded = currentRequest.encode(); encoded.headers().set(HttpHeaderNames.HOST, remoteHost); encoded.headers().set(HttpHeaderNames.USER_AGENT, endpoint.context().environment().userAgent().formattedLong()); ctx.writeAndFlush(encoded); } catch (Throwable t) { currentRequest.response().completeExceptionally(t); if (endpoint != null) { endpoint.decrementOutstandingRequests(); } } currentContent.clear(); } else { if (endpoint != null) { endpoint.decrementOutstandingRequests(); } eventBus.publish(new InvalidRequestDetectedEvent(ioContext, ServiceType.MANAGER, msg)); ctx.channel().close().addListener(f -> eventBus.publish(new ChannelClosedProactivelyEvent( ioContext, ChannelClosedProactivelyEvent.Reason.INVALID_REQUEST_DETECTED) )); } }
ManagerMessageHandler extends ChannelDuplexHandler { @Override @SuppressWarnings({ "unchecked" }) public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) { if (msg instanceof ManagerRequest) { if (currentRequest != null) { RetryOrchestrator.maybeRetry(coreContext, (ManagerRequest<Response>) msg, RetryReason.NOT_PIPELINED_REQUEST_IN_FLIGHT); if (endpoint != null) { endpoint.decrementOutstandingRequests(); } return; } try { currentRequest = (ManagerRequest<Response>) msg; FullHttpRequest encoded = currentRequest.encode(); encoded.headers().set(HttpHeaderNames.HOST, remoteHost); encoded.headers().set(HttpHeaderNames.USER_AGENT, endpoint.context().environment().userAgent().formattedLong()); ctx.writeAndFlush(encoded); } catch (Throwable t) { currentRequest.response().completeExceptionally(t); if (endpoint != null) { endpoint.decrementOutstandingRequests(); } } currentContent.clear(); } else { if (endpoint != null) { endpoint.decrementOutstandingRequests(); } eventBus.publish(new InvalidRequestDetectedEvent(ioContext, ServiceType.MANAGER, msg)); ctx.channel().close().addListener(f -> eventBus.publish(new ChannelClosedProactivelyEvent( ioContext, ChannelClosedProactivelyEvent.Reason.INVALID_REQUEST_DETECTED) )); } } }
ManagerMessageHandler extends ChannelDuplexHandler { @Override @SuppressWarnings({ "unchecked" }) public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) { if (msg instanceof ManagerRequest) { if (currentRequest != null) { RetryOrchestrator.maybeRetry(coreContext, (ManagerRequest<Response>) msg, RetryReason.NOT_PIPELINED_REQUEST_IN_FLIGHT); if (endpoint != null) { endpoint.decrementOutstandingRequests(); } return; } try { currentRequest = (ManagerRequest<Response>) msg; FullHttpRequest encoded = currentRequest.encode(); encoded.headers().set(HttpHeaderNames.HOST, remoteHost); encoded.headers().set(HttpHeaderNames.USER_AGENT, endpoint.context().environment().userAgent().formattedLong()); ctx.writeAndFlush(encoded); } catch (Throwable t) { currentRequest.response().completeExceptionally(t); if (endpoint != null) { endpoint.decrementOutstandingRequests(); } } currentContent.clear(); } else { if (endpoint != null) { endpoint.decrementOutstandingRequests(); } eventBus.publish(new InvalidRequestDetectedEvent(ioContext, ServiceType.MANAGER, msg)); ctx.channel().close().addListener(f -> eventBus.publish(new ChannelClosedProactivelyEvent( ioContext, ChannelClosedProactivelyEvent.Reason.INVALID_REQUEST_DETECTED) )); } } ManagerMessageHandler(final BaseEndpoint endpoint, final CoreContext coreContext); }
ManagerMessageHandler extends ChannelDuplexHandler { @Override @SuppressWarnings({ "unchecked" }) public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) { if (msg instanceof ManagerRequest) { if (currentRequest != null) { RetryOrchestrator.maybeRetry(coreContext, (ManagerRequest<Response>) msg, RetryReason.NOT_PIPELINED_REQUEST_IN_FLIGHT); if (endpoint != null) { endpoint.decrementOutstandingRequests(); } return; } try { currentRequest = (ManagerRequest<Response>) msg; FullHttpRequest encoded = currentRequest.encode(); encoded.headers().set(HttpHeaderNames.HOST, remoteHost); encoded.headers().set(HttpHeaderNames.USER_AGENT, endpoint.context().environment().userAgent().formattedLong()); ctx.writeAndFlush(encoded); } catch (Throwable t) { currentRequest.response().completeExceptionally(t); if (endpoint != null) { endpoint.decrementOutstandingRequests(); } } currentContent.clear(); } else { if (endpoint != null) { endpoint.decrementOutstandingRequests(); } eventBus.publish(new InvalidRequestDetectedEvent(ioContext, ServiceType.MANAGER, msg)); ctx.channel().close().addListener(f -> eventBus.publish(new ChannelClosedProactivelyEvent( ioContext, ChannelClosedProactivelyEvent.Reason.INVALID_REQUEST_DETECTED) )); } } ManagerMessageHandler(final BaseEndpoint endpoint, final CoreContext coreContext); @Override void channelActive(final ChannelHandlerContext ctx); @Override @SuppressWarnings({ "unchecked" }) void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise); @Override void channelRead(ChannelHandlerContext ctx, Object msg); @Override void channelInactive(final ChannelHandlerContext ctx); @Override void userEventTriggered(final ChannelHandlerContext ctx, final Object evt); }
ManagerMessageHandler extends ChannelDuplexHandler { @Override @SuppressWarnings({ "unchecked" }) public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) { if (msg instanceof ManagerRequest) { if (currentRequest != null) { RetryOrchestrator.maybeRetry(coreContext, (ManagerRequest<Response>) msg, RetryReason.NOT_PIPELINED_REQUEST_IN_FLIGHT); if (endpoint != null) { endpoint.decrementOutstandingRequests(); } return; } try { currentRequest = (ManagerRequest<Response>) msg; FullHttpRequest encoded = currentRequest.encode(); encoded.headers().set(HttpHeaderNames.HOST, remoteHost); encoded.headers().set(HttpHeaderNames.USER_AGENT, endpoint.context().environment().userAgent().formattedLong()); ctx.writeAndFlush(encoded); } catch (Throwable t) { currentRequest.response().completeExceptionally(t); if (endpoint != null) { endpoint.decrementOutstandingRequests(); } } currentContent.clear(); } else { if (endpoint != null) { endpoint.decrementOutstandingRequests(); } eventBus.publish(new InvalidRequestDetectedEvent(ioContext, ServiceType.MANAGER, msg)); ctx.channel().close().addListener(f -> eventBus.publish(new ChannelClosedProactivelyEvent( ioContext, ChannelClosedProactivelyEvent.Reason.INVALID_REQUEST_DETECTED) )); } } ManagerMessageHandler(final BaseEndpoint endpoint, final CoreContext coreContext); @Override void channelActive(final ChannelHandlerContext ctx); @Override @SuppressWarnings({ "unchecked" }) void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise); @Override void channelRead(ChannelHandlerContext ctx, Object msg); @Override void channelInactive(final ChannelHandlerContext ctx); @Override void userEventTriggered(final ChannelHandlerContext ctx, final Object evt);  }