@Test public void testFileCleanup() throws Exception { final Path dir = temporaryFolder.newFolder().toPath(); final Path file = dir.resolve("file"); Files.createFile(file); RestfulGateway mockRestfulGateway = TestingRestfulGateway.newBuilder() .build(); final GatewayRetriever<RestfulGateway> mockGatewayRetriever = () -> CompletableFuture.completedFuture(mockRestfulGateway); CompletableFuture<Void> requestProcessingCompleteFuture = new CompletableFuture<>(); TestHandler handler = new TestHandler(requestProcessingCompleteFuture, mockGatewayRetriever); RouteResult<?> routeResult = new RouteResult<>("", "", Collections.emptyMap(), Collections.emptyMap(), ""); HttpRequest request = new DefaultFullHttpRequest( HttpVersion.HTTP_1_1, HttpMethod.GET, TestHandler.TestHeaders.INSTANCE.getTargetRestEndpointURL(), Unpooled.wrappedBuffer(new byte[0])); RoutedRequest<?> routerRequest = new RoutedRequest<>(routeResult, request); Attribute<FileUploads> attribute = new SimpleAttribute(); attribute.set(new FileUploads(dir)); Channel channel = mock(Channel.class); when(channel.attr(any(AttributeKey.class))).thenReturn(attribute); ChannelHandlerContext context = mock(ChannelHandlerContext.class); when(context.channel()).thenReturn(channel); handler.respondAsLeader(context, routerRequest, mockRestfulGateway); Assert.assertTrue(Files.exists(file)); requestProcessingCompleteFuture.complete(null); Assert.assertFalse(Files.exists(file)); }
@Override protected void respondAsLeader(ChannelHandlerContext ctx, RoutedRequest routedRequest, T gateway) { HttpRequest httpRequest = routedRequest.getRequest(); if (log.isTraceEnabled()) { log.trace("Received request " + httpRequest.uri() + '.'); } FileUploads uploadedFiles = null; try { inFlightRequestTracker.registerRequest(); if (!(httpRequest instanceof FullHttpRequest)) { log.error("Implementation error: Received a request that wasn't a FullHttpRequest."); throw new RestHandlerException("Bad request received.", HttpResponseStatus.BAD_REQUEST); } final ByteBuf msgContent = ((FullHttpRequest) httpRequest).content(); uploadedFiles = FileUploadHandler.getMultipartFileUploads(ctx); if (!untypedResponseMessageHeaders.acceptsFileUploads() && !uploadedFiles.getUploadedFiles().isEmpty()) { throw new RestHandlerException("File uploads not allowed.", HttpResponseStatus.BAD_REQUEST); } R request; if (msgContent.capacity() == 0) { try { request = MAPPER.readValue("{}", untypedResponseMessageHeaders.getRequestClass()); } catch (JsonParseException | JsonMappingException je) { throw new RestHandlerException("Bad request received. Request did not conform to expected format.", HttpResponseStatus.BAD_REQUEST, je); } } else { try { InputStream in = new ByteBufInputStream(msgContent); request = MAPPER.readValue(in, untypedResponseMessageHeaders.getRequestClass()); } catch (JsonParseException | JsonMappingException je) { throw new RestHandlerException( String.format("Request did not match expected format %s.", untypedResponseMessageHeaders.getRequestClass().getSimpleName()), HttpResponseStatus.BAD_REQUEST, je); } } final HandlerRequest<R, M> handlerRequest; try { handlerRequest = new HandlerRequest<R, M>( request, untypedResponseMessageHeaders.getUnresolvedMessageParameters(), routedRequest.getRouteResult().pathParams(), routedRequest.getRouteResult().queryParams(), uploadedFiles.getUploadedFiles()); } catch (HandlerRequestException hre) { log.error("Could not create the handler request.", hre); throw new RestHandlerException( String.format("Bad request, could not parse parameters: %s", hre.getMessage()), HttpResponseStatus.BAD_REQUEST, hre); } log.trace("Starting request processing."); CompletableFuture<Void> requestProcessingFuture = respondToRequest( ctx, httpRequest, handlerRequest, gateway); final FileUploads finalUploadedFiles = uploadedFiles; requestProcessingFuture .whenComplete((Void ignored, Throwable throwable) -> { if (throwable != null) { handleException(ExceptionUtils.stripCompletionException(throwable), ctx, httpRequest) .whenComplete((Void ignored2, Throwable throwable2) -> finalizeRequestProcessing(finalUploadedFiles)); } else { finalizeRequestProcessing(finalUploadedFiles); } }); } catch (Throwable e) { final FileUploads finalUploadedFiles = uploadedFiles; handleException(e, ctx, httpRequest) .whenComplete((Void ignored, Throwable throwable) -> finalizeRequestProcessing(finalUploadedFiles)); } }
AbstractHandler extends LeaderRetrievalHandler<T> implements AutoCloseableAsync { @Override protected void respondAsLeader(ChannelHandlerContext ctx, RoutedRequest routedRequest, T gateway) { HttpRequest httpRequest = routedRequest.getRequest(); if (log.isTraceEnabled()) { log.trace("Received request " + httpRequest.uri() + '.'); } FileUploads uploadedFiles = null; try { inFlightRequestTracker.registerRequest(); if (!(httpRequest instanceof FullHttpRequest)) { log.error("Implementation error: Received a request that wasn't a FullHttpRequest."); throw new RestHandlerException("Bad request received.", HttpResponseStatus.BAD_REQUEST); } final ByteBuf msgContent = ((FullHttpRequest) httpRequest).content(); uploadedFiles = FileUploadHandler.getMultipartFileUploads(ctx); if (!untypedResponseMessageHeaders.acceptsFileUploads() && !uploadedFiles.getUploadedFiles().isEmpty()) { throw new RestHandlerException("File uploads not allowed.", HttpResponseStatus.BAD_REQUEST); } R request; if (msgContent.capacity() == 0) { try { request = MAPPER.readValue("{}", untypedResponseMessageHeaders.getRequestClass()); } catch (JsonParseException | JsonMappingException je) { throw new RestHandlerException("Bad request received. Request did not conform to expected format.", HttpResponseStatus.BAD_REQUEST, je); } } else { try { InputStream in = new ByteBufInputStream(msgContent); request = MAPPER.readValue(in, untypedResponseMessageHeaders.getRequestClass()); } catch (JsonParseException | JsonMappingException je) { throw new RestHandlerException( String.format("Request did not match expected format %s.", untypedResponseMessageHeaders.getRequestClass().getSimpleName()), HttpResponseStatus.BAD_REQUEST, je); } } final HandlerRequest<R, M> handlerRequest; try { handlerRequest = new HandlerRequest<R, M>( request, untypedResponseMessageHeaders.getUnresolvedMessageParameters(), routedRequest.getRouteResult().pathParams(), routedRequest.getRouteResult().queryParams(), uploadedFiles.getUploadedFiles()); } catch (HandlerRequestException hre) { log.error("Could not create the handler request.", hre); throw new RestHandlerException( String.format("Bad request, could not parse parameters: %s", hre.getMessage()), HttpResponseStatus.BAD_REQUEST, hre); } log.trace("Starting request processing."); CompletableFuture<Void> requestProcessingFuture = respondToRequest( ctx, httpRequest, handlerRequest, gateway); final FileUploads finalUploadedFiles = uploadedFiles; requestProcessingFuture .whenComplete((Void ignored, Throwable throwable) -> { if (throwable != null) { handleException(ExceptionUtils.stripCompletionException(throwable), ctx, httpRequest) .whenComplete((Void ignored2, Throwable throwable2) -> finalizeRequestProcessing(finalUploadedFiles)); } else { finalizeRequestProcessing(finalUploadedFiles); } }); } catch (Throwable e) { final FileUploads finalUploadedFiles = uploadedFiles; handleException(e, ctx, httpRequest) .whenComplete((Void ignored, Throwable throwable) -> finalizeRequestProcessing(finalUploadedFiles)); } } }
AbstractHandler extends LeaderRetrievalHandler<T> implements AutoCloseableAsync { @Override protected void respondAsLeader(ChannelHandlerContext ctx, RoutedRequest routedRequest, T gateway) { HttpRequest httpRequest = routedRequest.getRequest(); if (log.isTraceEnabled()) { log.trace("Received request " + httpRequest.uri() + '.'); } FileUploads uploadedFiles = null; try { inFlightRequestTracker.registerRequest(); if (!(httpRequest instanceof FullHttpRequest)) { log.error("Implementation error: Received a request that wasn't a FullHttpRequest."); throw new RestHandlerException("Bad request received.", HttpResponseStatus.BAD_REQUEST); } final ByteBuf msgContent = ((FullHttpRequest) httpRequest).content(); uploadedFiles = FileUploadHandler.getMultipartFileUploads(ctx); if (!untypedResponseMessageHeaders.acceptsFileUploads() && !uploadedFiles.getUploadedFiles().isEmpty()) { throw new RestHandlerException("File uploads not allowed.", HttpResponseStatus.BAD_REQUEST); } R request; if (msgContent.capacity() == 0) { try { request = MAPPER.readValue("{}", untypedResponseMessageHeaders.getRequestClass()); } catch (JsonParseException | JsonMappingException je) { throw new RestHandlerException("Bad request received. Request did not conform to expected format.", HttpResponseStatus.BAD_REQUEST, je); } } else { try { InputStream in = new ByteBufInputStream(msgContent); request = MAPPER.readValue(in, untypedResponseMessageHeaders.getRequestClass()); } catch (JsonParseException | JsonMappingException je) { throw new RestHandlerException( String.format("Request did not match expected format %s.", untypedResponseMessageHeaders.getRequestClass().getSimpleName()), HttpResponseStatus.BAD_REQUEST, je); } } final HandlerRequest<R, M> handlerRequest; try { handlerRequest = new HandlerRequest<R, M>( request, untypedResponseMessageHeaders.getUnresolvedMessageParameters(), routedRequest.getRouteResult().pathParams(), routedRequest.getRouteResult().queryParams(), uploadedFiles.getUploadedFiles()); } catch (HandlerRequestException hre) { log.error("Could not create the handler request.", hre); throw new RestHandlerException( String.format("Bad request, could not parse parameters: %s", hre.getMessage()), HttpResponseStatus.BAD_REQUEST, hre); } log.trace("Starting request processing."); CompletableFuture<Void> requestProcessingFuture = respondToRequest( ctx, httpRequest, handlerRequest, gateway); final FileUploads finalUploadedFiles = uploadedFiles; requestProcessingFuture .whenComplete((Void ignored, Throwable throwable) -> { if (throwable != null) { handleException(ExceptionUtils.stripCompletionException(throwable), ctx, httpRequest) .whenComplete((Void ignored2, Throwable throwable2) -> finalizeRequestProcessing(finalUploadedFiles)); } else { finalizeRequestProcessing(finalUploadedFiles); } }); } catch (Throwable e) { final FileUploads finalUploadedFiles = uploadedFiles; handleException(e, ctx, httpRequest) .whenComplete((Void ignored, Throwable throwable) -> finalizeRequestProcessing(finalUploadedFiles)); } } protected  AbstractHandler(
			@Nonnull GatewayRetriever<? extends T> leaderRetriever,
			@Nonnull Time timeout,
			@Nonnull Map<String, String> responseHeaders,
			@Nonnull UntypedResponseMessageHeaders<R, M> untypedResponseMessageHeaders); }
AbstractHandler extends LeaderRetrievalHandler<T> implements AutoCloseableAsync { @Override protected void respondAsLeader(ChannelHandlerContext ctx, RoutedRequest routedRequest, T gateway) { HttpRequest httpRequest = routedRequest.getRequest(); if (log.isTraceEnabled()) { log.trace("Received request " + httpRequest.uri() + '.'); } FileUploads uploadedFiles = null; try { inFlightRequestTracker.registerRequest(); if (!(httpRequest instanceof FullHttpRequest)) { log.error("Implementation error: Received a request that wasn't a FullHttpRequest."); throw new RestHandlerException("Bad request received.", HttpResponseStatus.BAD_REQUEST); } final ByteBuf msgContent = ((FullHttpRequest) httpRequest).content(); uploadedFiles = FileUploadHandler.getMultipartFileUploads(ctx); if (!untypedResponseMessageHeaders.acceptsFileUploads() && !uploadedFiles.getUploadedFiles().isEmpty()) { throw new RestHandlerException("File uploads not allowed.", HttpResponseStatus.BAD_REQUEST); } R request; if (msgContent.capacity() == 0) { try { request = MAPPER.readValue("{}", untypedResponseMessageHeaders.getRequestClass()); } catch (JsonParseException | JsonMappingException je) { throw new RestHandlerException("Bad request received. Request did not conform to expected format.", HttpResponseStatus.BAD_REQUEST, je); } } else { try { InputStream in = new ByteBufInputStream(msgContent); request = MAPPER.readValue(in, untypedResponseMessageHeaders.getRequestClass()); } catch (JsonParseException | JsonMappingException je) { throw new RestHandlerException( String.format("Request did not match expected format %s.", untypedResponseMessageHeaders.getRequestClass().getSimpleName()), HttpResponseStatus.BAD_REQUEST, je); } } final HandlerRequest<R, M> handlerRequest; try { handlerRequest = new HandlerRequest<R, M>( request, untypedResponseMessageHeaders.getUnresolvedMessageParameters(), routedRequest.getRouteResult().pathParams(), routedRequest.getRouteResult().queryParams(), uploadedFiles.getUploadedFiles()); } catch (HandlerRequestException hre) { log.error("Could not create the handler request.", hre); throw new RestHandlerException( String.format("Bad request, could not parse parameters: %s", hre.getMessage()), HttpResponseStatus.BAD_REQUEST, hre); } log.trace("Starting request processing."); CompletableFuture<Void> requestProcessingFuture = respondToRequest( ctx, httpRequest, handlerRequest, gateway); final FileUploads finalUploadedFiles = uploadedFiles; requestProcessingFuture .whenComplete((Void ignored, Throwable throwable) -> { if (throwable != null) { handleException(ExceptionUtils.stripCompletionException(throwable), ctx, httpRequest) .whenComplete((Void ignored2, Throwable throwable2) -> finalizeRequestProcessing(finalUploadedFiles)); } else { finalizeRequestProcessing(finalUploadedFiles); } }); } catch (Throwable e) { final FileUploads finalUploadedFiles = uploadedFiles; handleException(e, ctx, httpRequest) .whenComplete((Void ignored, Throwable throwable) -> finalizeRequestProcessing(finalUploadedFiles)); } } protected  AbstractHandler(
			@Nonnull GatewayRetriever<? extends T> leaderRetriever,
			@Nonnull Time timeout,
			@Nonnull Map<String, String> responseHeaders,
			@Nonnull UntypedResponseMessageHeaders<R, M> untypedResponseMessageHeaders); @Override final CompletableFuture<Void> closeAsync(); }
AbstractHandler extends LeaderRetrievalHandler<T> implements AutoCloseableAsync { @Override protected void respondAsLeader(ChannelHandlerContext ctx, RoutedRequest routedRequest, T gateway) { HttpRequest httpRequest = routedRequest.getRequest(); if (log.isTraceEnabled()) { log.trace("Received request " + httpRequest.uri() + '.'); } FileUploads uploadedFiles = null; try { inFlightRequestTracker.registerRequest(); if (!(httpRequest instanceof FullHttpRequest)) { log.error("Implementation error: Received a request that wasn't a FullHttpRequest."); throw new RestHandlerException("Bad request received.", HttpResponseStatus.BAD_REQUEST); } final ByteBuf msgContent = ((FullHttpRequest) httpRequest).content(); uploadedFiles = FileUploadHandler.getMultipartFileUploads(ctx); if (!untypedResponseMessageHeaders.acceptsFileUploads() && !uploadedFiles.getUploadedFiles().isEmpty()) { throw new RestHandlerException("File uploads not allowed.", HttpResponseStatus.BAD_REQUEST); } R request; if (msgContent.capacity() == 0) { try { request = MAPPER.readValue("{}", untypedResponseMessageHeaders.getRequestClass()); } catch (JsonParseException | JsonMappingException je) { throw new RestHandlerException("Bad request received. Request did not conform to expected format.", HttpResponseStatus.BAD_REQUEST, je); } } else { try { InputStream in = new ByteBufInputStream(msgContent); request = MAPPER.readValue(in, untypedResponseMessageHeaders.getRequestClass()); } catch (JsonParseException | JsonMappingException je) { throw new RestHandlerException( String.format("Request did not match expected format %s.", untypedResponseMessageHeaders.getRequestClass().getSimpleName()), HttpResponseStatus.BAD_REQUEST, je); } } final HandlerRequest<R, M> handlerRequest; try { handlerRequest = new HandlerRequest<R, M>( request, untypedResponseMessageHeaders.getUnresolvedMessageParameters(), routedRequest.getRouteResult().pathParams(), routedRequest.getRouteResult().queryParams(), uploadedFiles.getUploadedFiles()); } catch (HandlerRequestException hre) { log.error("Could not create the handler request.", hre); throw new RestHandlerException( String.format("Bad request, could not parse parameters: %s", hre.getMessage()), HttpResponseStatus.BAD_REQUEST, hre); } log.trace("Starting request processing."); CompletableFuture<Void> requestProcessingFuture = respondToRequest( ctx, httpRequest, handlerRequest, gateway); final FileUploads finalUploadedFiles = uploadedFiles; requestProcessingFuture .whenComplete((Void ignored, Throwable throwable) -> { if (throwable != null) { handleException(ExceptionUtils.stripCompletionException(throwable), ctx, httpRequest) .whenComplete((Void ignored2, Throwable throwable2) -> finalizeRequestProcessing(finalUploadedFiles)); } else { finalizeRequestProcessing(finalUploadedFiles); } }); } catch (Throwable e) { final FileUploads finalUploadedFiles = uploadedFiles; handleException(e, ctx, httpRequest) .whenComplete((Void ignored, Throwable throwable) -> finalizeRequestProcessing(finalUploadedFiles)); } } protected  AbstractHandler(
			@Nonnull GatewayRetriever<? extends T> leaderRetriever,
			@Nonnull Time timeout,
			@Nonnull Map<String, String> responseHeaders,
			@Nonnull UntypedResponseMessageHeaders<R, M> untypedResponseMessageHeaders); @Override final CompletableFuture<Void> closeAsync();  }