@Test public void testReceiveFlowFiles() throws Exception { final HttpServletRequest req = createCommonHttpServletRequest(); final DataTransferResource resource = getDataTransferResource(); final HttpFlowFileServerProtocol serverProtocol = resource.getHttpFlowFileServerProtocol(null); final RootGroupPort port = mock(RootGroupPort.class); doReturn(port).when(serverProtocol).getPort(); doAnswer(invocation -> { Peer peer = (Peer) invocation.getArguments()[0]; ((HttpServerCommunicationsSession)peer.getCommunicationsSession()).setChecksum("server-checksum"); return 7; }).when(port).receiveFlowFiles(any(Peer.class), any()); final ServletContext context = null; final InputStream inputStream = null; final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String transactionId = transactionManager.createTransaction(); final Response response = resource.receiveFlowFiles("port-id", transactionId, req, context, inputStream); transactionManager.cancelTransaction(transactionId); final Object entity = response.getEntity(); assertEquals(202, response.getStatus()); assertEquals("server-checksum", entity); }
@POST @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.TEXT_PLAIN) @Path("input-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files to the input port", response = String.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) public Response receiveFlowFiles( @ApiParam( value = "The input port id.", required = true ) @PathParam("portId") String portId, @PathParam("transactionId") String transactionId, @Context HttpServletRequest req, @Context ServletContext context, InputStream inputStream) { serviceFacade.authorizeAccess(lookup -> { authorizeDataTransfer(lookup, ResourceType.InputPort, portId); }); final ValidateRequestResult validationResult = validateResult(req, portId, transactionId); if (validationResult.errResponse != null) { return validationResult.errResponse; } logger.debug("receiveFlowFiles request: portId={}", portId); final ByteArrayOutputStream out = new ByteArrayOutputStream(); final Peer peer = constructPeer(req, inputStream, out, portId, transactionId); final int transportProtocolVersion = validationResult.transportProtocolVersion; try { HttpFlowFileServerProtocol serverProtocol = initiateServerProtocol(req, peer, transportProtocolVersion); int numOfFlowFiles = serverProtocol.getPort().receiveFlowFiles(peer, serverProtocol); logger.debug("finished receiving flow files, numOfFlowFiles={}", numOfFlowFiles); if (numOfFlowFiles < 1) { return Response.status(Response.Status.BAD_REQUEST) .entity("Client should send request when there is data to send. There was no flow file sent.").build(); } } catch (HandshakeException e) { return responseCreator.handshakeExceptionResponse(e); } catch (NotAuthorizedException e) { return responseCreator.unauthorizedResponse(e); } catch (BadRequestException | RequestExpiredException e) { return responseCreator.badRequestResponse(e); } catch (Exception e) { return responseCreator.unexpectedErrorResponse(portId, e); } String serverChecksum = ((HttpServerCommunicationsSession) peer.getCommunicationsSession()).getChecksum(); return responseCreator.acceptedResponse(transactionManager, serverChecksum, transportProtocolVersion); }
DataTransferResource extends ApplicationResource { @POST @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.TEXT_PLAIN) @Path("input-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files to the input port", response = String.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) public Response receiveFlowFiles( @ApiParam( value = "The input port id.", required = true ) @PathParam("portId") String portId, @PathParam("transactionId") String transactionId, @Context HttpServletRequest req, @Context ServletContext context, InputStream inputStream) { serviceFacade.authorizeAccess(lookup -> { authorizeDataTransfer(lookup, ResourceType.InputPort, portId); }); final ValidateRequestResult validationResult = validateResult(req, portId, transactionId); if (validationResult.errResponse != null) { return validationResult.errResponse; } logger.debug("receiveFlowFiles request: portId={}", portId); final ByteArrayOutputStream out = new ByteArrayOutputStream(); final Peer peer = constructPeer(req, inputStream, out, portId, transactionId); final int transportProtocolVersion = validationResult.transportProtocolVersion; try { HttpFlowFileServerProtocol serverProtocol = initiateServerProtocol(req, peer, transportProtocolVersion); int numOfFlowFiles = serverProtocol.getPort().receiveFlowFiles(peer, serverProtocol); logger.debug("finished receiving flow files, numOfFlowFiles={}", numOfFlowFiles); if (numOfFlowFiles < 1) { return Response.status(Response.Status.BAD_REQUEST) .entity("Client should send request when there is data to send. There was no flow file sent.").build(); } } catch (HandshakeException e) { return responseCreator.handshakeExceptionResponse(e); } catch (NotAuthorizedException e) { return responseCreator.unauthorizedResponse(e); } catch (BadRequestException | RequestExpiredException e) { return responseCreator.badRequestResponse(e); } catch (Exception e) { return responseCreator.unexpectedErrorResponse(portId, e); } String serverChecksum = ((HttpServerCommunicationsSession) peer.getCommunicationsSession()).getChecksum(); return responseCreator.acceptedResponse(transactionManager, serverChecksum, transportProtocolVersion); } }
DataTransferResource extends ApplicationResource { @POST @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.TEXT_PLAIN) @Path("input-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files to the input port", response = String.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) public Response receiveFlowFiles( @ApiParam( value = "The input port id.", required = true ) @PathParam("portId") String portId, @PathParam("transactionId") String transactionId, @Context HttpServletRequest req, @Context ServletContext context, InputStream inputStream) { serviceFacade.authorizeAccess(lookup -> { authorizeDataTransfer(lookup, ResourceType.InputPort, portId); }); final ValidateRequestResult validationResult = validateResult(req, portId, transactionId); if (validationResult.errResponse != null) { return validationResult.errResponse; } logger.debug("receiveFlowFiles request: portId={}", portId); final ByteArrayOutputStream out = new ByteArrayOutputStream(); final Peer peer = constructPeer(req, inputStream, out, portId, transactionId); final int transportProtocolVersion = validationResult.transportProtocolVersion; try { HttpFlowFileServerProtocol serverProtocol = initiateServerProtocol(req, peer, transportProtocolVersion); int numOfFlowFiles = serverProtocol.getPort().receiveFlowFiles(peer, serverProtocol); logger.debug("finished receiving flow files, numOfFlowFiles={}", numOfFlowFiles); if (numOfFlowFiles < 1) { return Response.status(Response.Status.BAD_REQUEST) .entity("Client should send request when there is data to send. There was no flow file sent.").build(); } } catch (HandshakeException e) { return responseCreator.handshakeExceptionResponse(e); } catch (NotAuthorizedException e) { return responseCreator.unauthorizedResponse(e); } catch (BadRequestException | RequestExpiredException e) { return responseCreator.badRequestResponse(e); } catch (Exception e) { return responseCreator.unexpectedErrorResponse(portId, e); } String serverChecksum = ((HttpServerCommunicationsSession) peer.getCommunicationsSession()).getChecksum(); return responseCreator.acceptedResponse(transactionManager, serverChecksum, transportProtocolVersion); } DataTransferResource(final NiFiProperties nifiProperties); }
DataTransferResource extends ApplicationResource { @POST @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.TEXT_PLAIN) @Path("input-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files to the input port", response = String.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) public Response receiveFlowFiles( @ApiParam( value = "The input port id.", required = true ) @PathParam("portId") String portId, @PathParam("transactionId") String transactionId, @Context HttpServletRequest req, @Context ServletContext context, InputStream inputStream) { serviceFacade.authorizeAccess(lookup -> { authorizeDataTransfer(lookup, ResourceType.InputPort, portId); }); final ValidateRequestResult validationResult = validateResult(req, portId, transactionId); if (validationResult.errResponse != null) { return validationResult.errResponse; } logger.debug("receiveFlowFiles request: portId={}", portId); final ByteArrayOutputStream out = new ByteArrayOutputStream(); final Peer peer = constructPeer(req, inputStream, out, portId, transactionId); final int transportProtocolVersion = validationResult.transportProtocolVersion; try { HttpFlowFileServerProtocol serverProtocol = initiateServerProtocol(req, peer, transportProtocolVersion); int numOfFlowFiles = serverProtocol.getPort().receiveFlowFiles(peer, serverProtocol); logger.debug("finished receiving flow files, numOfFlowFiles={}", numOfFlowFiles); if (numOfFlowFiles < 1) { return Response.status(Response.Status.BAD_REQUEST) .entity("Client should send request when there is data to send. There was no flow file sent.").build(); } } catch (HandshakeException e) { return responseCreator.handshakeExceptionResponse(e); } catch (NotAuthorizedException e) { return responseCreator.unauthorizedResponse(e); } catch (BadRequestException | RequestExpiredException e) { return responseCreator.badRequestResponse(e); } catch (Exception e) { return responseCreator.unexpectedErrorResponse(portId, e); } String serverChecksum = ((HttpServerCommunicationsSession) peer.getCommunicationsSession()).getChecksum(); return responseCreator.acceptedResponse(transactionManager, serverChecksum, transportProtocolVersion); } DataTransferResource(final NiFiProperties nifiProperties); @POST @Produces(MediaType.APPLICATION_JSON) @Path("{portType}/{portId}/transactions") @ApiOperation( value = "Create a transaction to the specified output port or input port", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/{component-type}/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response createPortTransaction(
            @ApiParam(
                    value = "The port type.",
                    required = true,
                    allowableValues = "input-ports, output-ports"
            )
            @PathParam("portType") String portType,
            @PathParam("portId") String portId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); @POST @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.TEXT_PLAIN) @Path("input-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files to the input port", response = String.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response receiveFlowFiles(
            @ApiParam(
                    value = "The input port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response commitOutputPortTransaction(
            @ApiParam(
                    value = "The response code. Available values are CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).",
                    required = true
            )
            @QueryParam(RESPONSE_CODE) Integer responseCode,
            @ApiParam(
                    value = "A checksum calculated at client side using CRC32 to check flow file content integrity. It must match with the value calculated at server side.",
                    required = true
            )
            @QueryParam(CHECK_SUM) @DefaultValue(StringUtils.EMPTY) String checksum,
            @ApiParam(
                    value = "The output port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @ApiParam(
                    value = "The transaction id.",
                    required = true
            )
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("input-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response commitInputPortTransaction(
            @ApiParam(
                    value = "The response code. Available values are BAD_CHECKSUM(19), CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).",
                    required = true
            )
            @QueryParam(RESPONSE_CODE) Integer responseCode,
            @ApiParam(
                    value = "The input port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @ApiParam(
                    value = "The transaction id.",
                    required = true
            )
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @GET @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_OCTET_STREAM) @Path("output-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files from the output port", response = StreamingOutput.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 200, message = "There is no flow file to return."), @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response transferFlowFiles(
            @ApiParam(
                    value = "The output port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            InputStream inputStream); @PUT @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_JSON) @Path("input-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Extend transaction TTL", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") } ) Response extendInputPortTransactionTTL(
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); @PUT @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Extend transaction TTL", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response extendOutputPortTransactionTTL(
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); Response extendPortTransactionTTL(
            String portType,
            String portId,
            String transactionId,
            HttpServletRequest req,
            HttpServletResponse res,
            ServletContext context,
            UriInfo uriInfo,
            InputStream inputStream); void setServiceFacade(NiFiServiceFacade serviceFacade); }
DataTransferResource extends ApplicationResource { @POST @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.TEXT_PLAIN) @Path("input-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files to the input port", response = String.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) public Response receiveFlowFiles( @ApiParam( value = "The input port id.", required = true ) @PathParam("portId") String portId, @PathParam("transactionId") String transactionId, @Context HttpServletRequest req, @Context ServletContext context, InputStream inputStream) { serviceFacade.authorizeAccess(lookup -> { authorizeDataTransfer(lookup, ResourceType.InputPort, portId); }); final ValidateRequestResult validationResult = validateResult(req, portId, transactionId); if (validationResult.errResponse != null) { return validationResult.errResponse; } logger.debug("receiveFlowFiles request: portId={}", portId); final ByteArrayOutputStream out = new ByteArrayOutputStream(); final Peer peer = constructPeer(req, inputStream, out, portId, transactionId); final int transportProtocolVersion = validationResult.transportProtocolVersion; try { HttpFlowFileServerProtocol serverProtocol = initiateServerProtocol(req, peer, transportProtocolVersion); int numOfFlowFiles = serverProtocol.getPort().receiveFlowFiles(peer, serverProtocol); logger.debug("finished receiving flow files, numOfFlowFiles={}", numOfFlowFiles); if (numOfFlowFiles < 1) { return Response.status(Response.Status.BAD_REQUEST) .entity("Client should send request when there is data to send. There was no flow file sent.").build(); } } catch (HandshakeException e) { return responseCreator.handshakeExceptionResponse(e); } catch (NotAuthorizedException e) { return responseCreator.unauthorizedResponse(e); } catch (BadRequestException | RequestExpiredException e) { return responseCreator.badRequestResponse(e); } catch (Exception e) { return responseCreator.unexpectedErrorResponse(portId, e); } String serverChecksum = ((HttpServerCommunicationsSession) peer.getCommunicationsSession()).getChecksum(); return responseCreator.acceptedResponse(transactionManager, serverChecksum, transportProtocolVersion); } DataTransferResource(final NiFiProperties nifiProperties); @POST @Produces(MediaType.APPLICATION_JSON) @Path("{portType}/{portId}/transactions") @ApiOperation( value = "Create a transaction to the specified output port or input port", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/{component-type}/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response createPortTransaction(
            @ApiParam(
                    value = "The port type.",
                    required = true,
                    allowableValues = "input-ports, output-ports"
            )
            @PathParam("portType") String portType,
            @PathParam("portId") String portId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); @POST @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.TEXT_PLAIN) @Path("input-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files to the input port", response = String.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response receiveFlowFiles(
            @ApiParam(
                    value = "The input port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response commitOutputPortTransaction(
            @ApiParam(
                    value = "The response code. Available values are CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).",
                    required = true
            )
            @QueryParam(RESPONSE_CODE) Integer responseCode,
            @ApiParam(
                    value = "A checksum calculated at client side using CRC32 to check flow file content integrity. It must match with the value calculated at server side.",
                    required = true
            )
            @QueryParam(CHECK_SUM) @DefaultValue(StringUtils.EMPTY) String checksum,
            @ApiParam(
                    value = "The output port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @ApiParam(
                    value = "The transaction id.",
                    required = true
            )
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("input-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response commitInputPortTransaction(
            @ApiParam(
                    value = "The response code. Available values are BAD_CHECKSUM(19), CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).",
                    required = true
            )
            @QueryParam(RESPONSE_CODE) Integer responseCode,
            @ApiParam(
                    value = "The input port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @ApiParam(
                    value = "The transaction id.",
                    required = true
            )
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @GET @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_OCTET_STREAM) @Path("output-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files from the output port", response = StreamingOutput.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 200, message = "There is no flow file to return."), @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response transferFlowFiles(
            @ApiParam(
                    value = "The output port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            InputStream inputStream); @PUT @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_JSON) @Path("input-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Extend transaction TTL", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") } ) Response extendInputPortTransactionTTL(
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); @PUT @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Extend transaction TTL", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), @ApiResponse(code = 401, message = "Client could not be authenticated."), @ApiResponse(code = 403, message = "Client is not authorized to make this request."), @ApiResponse(code = 404, message = "The specified resource could not be found."), @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response extendOutputPortTransactionTTL(
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); Response extendPortTransactionTTL(
            String portType,
            String portId,
            String transactionId,
            HttpServletRequest req,
            HttpServletResponse res,
            ServletContext context,
            UriInfo uriInfo,
            InputStream inputStream); void setServiceFacade(NiFiServiceFacade serviceFacade); static final String CHECK_SUM; static final String RESPONSE_CODE; }