class DataTransferResource extends ApplicationResource {

    @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}")
            }
    )
    @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"),
            }
    )
    public 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) {

        // authorize access
        serviceFacade.authorizeAccess(lookup -> {
            authorizeDataTransfer(lookup, ResourceType.OutputPort, portId);
        });

        final ValidateRequestResult validationResult = validateResult(req, portId, transactionId);
        if (validationResult.errResponse != null) {
            return validationResult.errResponse;
        }

        logger.debug("transferFlowFiles request: portId={}", portId);

        // Before opening the real output stream for HTTP response,
        // use this temporary output stream to buffer handshake result.
        final ByteArrayOutputStream tempBos = new ByteArrayOutputStream();
        final Peer peer = constructPeer(req, inputStream, tempBos, portId, transactionId);
        final int transportProtocolVersion = validationResult.transportProtocolVersion;
        try {
            final HttpFlowFileServerProtocol serverProtocol = initiateServerProtocol(req, peer, transportProtocolVersion);

            StreamingOutput flowFileContent = new StreamingOutput() {
                @Override
                public void write(OutputStream outputStream) throws IOException, WebApplicationException {

                    HttpOutput output = (HttpOutput) peer.getCommunicationsSession().getOutput();
                    output.setOutputStream(outputStream);

                    try {
                        int numOfFlowFiles = serverProtocol.getPort().transferFlowFiles(peer, serverProtocol);
                        logger.debug("finished transferring flow files, numOfFlowFiles={}", numOfFlowFiles);
                        if (numOfFlowFiles < 1) {
                            // There was no flow file to transfer. Throw this exception to stop responding with SEE OTHER.
                            throw new WebApplicationException(Response.Status.OK);
                        }
                    } catch (NotAuthorizedException | BadRequestException | RequestExpiredException e) {
                        // Handshake is done outside of write() method, so these exception wouldn't be thrown.
                        throw new IOException("Failed to process the request.", e);
                    }
                }

            };

            return responseCreator.acceptedResponse(transactionManager, flowFileContent, transportProtocolVersion);

        } catch (HandshakeException e) {
            return responseCreator.handshakeExceptionResponse(e);

        } catch (Exception e) {
            return responseCreator.unexpectedErrorResponse(portId, e);
        }
    }

    public  DataTransferResource(final NiFiProperties nifiProperties);

}