问如果在流式传输远程目录文件时数据处理失败，则停止重命名文件

EN

.handle(Sftp.outboundGateway(sftpSessionFactory(), GET, "payload.remoteDirectory + payload.filename").options(STREAM).temporaryFileSuffix("_reading"))
.handle(readData(),c->c.advice(afterReading()))
.enrichHeaders(h -> h
        .headerExpression(FileHeaders.RENAME_TO, "headers[file_remoteDirectory] + 'archive/' + headers[file_remoteFile]")
        .headerExpression(FileHeaders.REMOTE_FILE, "headers[file_remoteFile]")
        .header(FileHeaders.REMOTE_DIRECTORY, "headers[file_remoteDirectory]"))
.handle(Sftp.outboundGateway(sftpSessionFactory(), MV, "headers[file_remoteDirectory]+headers[file_remoteFile]").renameExpression("headers['file_renameTo']"))
.get();

@Bean
    public ExpressionEvaluatingRequestHandlerAdvice afterReading() {
        ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
        advice.setSuccessChannelName("successReading.input");
        advice.setOnSuccessExpressionString("payload + ' was successful streamed'");
        advice.setFailureChannelName("failureReading.input");
        advice.setOnFailureExpressionString("payload + ' was bad, with reason: ' + #exception.cause.message");
        advice.setTrapException(true);
        advice.setPropagateEvaluationFailures(true);
        return advice;
    }

    @Bean
    public IntegrationFlow successReading() {
        return f -> f.log();
    }

    @Bean
    public IntegrationFlow failureReading() {
        return f -> f.log(ERROR);
    }


public GenericHandler readData() {
return new GenericHandler() {
    @Override
    public Object handle(Object o, Map map) {
        InputStream file = (InputStream) o;
        String fileName = (String) map.get(REMOTE_FILE);
        try {
            // processing data
        } catch (Exception e) {
            return new SftpException(500, String.format("Error while processing the file %s because of  Error: %s and reason %s", fileName, e.getMessage(), e.getCause()));
        }
        Closeable closeable = (Closeable) map.get(CLOSEABLE_RESOURCE);
        if (closeable != null) {
            try {
                closeable.close();
                file.close();
            } catch (Exception e) {
                logger.error(String.format("Session didn`t get closed after reading the stream data for file %s and error %s"), fileName, e.getMessage());
            }
        }
        return map;
    }
    };
}