Почему мой метод-обработчик не запускается, когда определяется как лямбда?

Question

Я определяю IntegrationFlow для потоковой передачи из SFTP в S3 с синтаксисом DSL следующим образом:

return IntegrationFlows.from(Sftp.inboundStreamingAdapter(remoteFileTemplate)
                        .remoteDirectory("remoteDirectory"),
                e -> e.poller(Pollers.fixedDelay(POLL, TimeUnit.SECONDS)))
                .transform(new StreamTransformer())
                .handle(s3UploadMessageHandler(outputFolderPath, "headers['file_remoteFile']")) // Upload on S3
                .get();

private S3MessageHandler s3UploadMessageHandler(String folderPath, String spelFileName) {
        S3MessageHandler s3MessageHandler = new S3MessageHandler(amazonS3, s3ConfigProperties.getBuckets().getCardManagementData());
        s3MessageHandler.setKeyExpression(new SpelExpressionParser().parseExpression(String.format("'%s/'.concat(%s)", folderPath, spelFileName)));
        s3MessageHandler.setCommand(S3MessageHandler.Command.UPLOAD);
        return s3MessageHandler;
    }

И он работает как задумано: файл хорошо загружен в мое хранилище S3 , Однако я хотел бы избежать синтаксиса SPEL и вставлять заголовки из сообщения в метод s3uploadMessageHandler, чтобы я мог использовать простой ValueExpression для установки keyExpression в методе s3UploadMessageHandler. Для этого я изменил

handle(s3UploadMessageHandler(outputFolderPath, "headers['file_remoteFile']")) // Upload on S3

на

handle(m -> s3UploadMessageHandler(outputFolderPath, (String) m.getHeaders().get("file_remoteFile"))) // Upload on S3

Но теперь этот обработчик, похоже, больше не запускается. В журналах нет ошибок, и я знаю из журналов, что опрос SFTP все еще работает.

Я попытался найти причину этого и увидел, что при вводе метода handle в IntegrationFlowdefinition.java тип класса messageHandler отличается: это S3MessageHandler при вызове без лямбды и MyCallingClass$lambda при вызове с лямбда-выражением.

Что я упустил, чтобы мой сценарий работал?

Answer 1

Существует два способа обработки сообщения. Один из них - реализация MessageHandler - это наиболее эффективный подход, который реализован в рамках реализации адаптера канала, например S3MessageHandler. Другим способом является вызов метода POJO - это наиболее удобный для пользователя подход, когда вам не нужно беспокоиться о каких-либо интерфейсах фреймворка.

Итак, когда вы используете его вот так .handle(s3UploadMessageHandler(...)), вы ссылаетесь на MessageHandler и фреймворк знает, что бин для этого MessageHandler должен быть зарегистрирован, поскольку ваш s3UploadMessageHandler() не является @Bean.

Когда вы используете его как лямбду, фреймворк обрабатывает его как вызов метода POJO, и для MethodInvokingMessageHandler зарегистрирован бин, но не ваш S3MessageHandler.

В любом случае, даже если вы измените s3UploadMessageHandler() на метод @Bean, он не будет работать, потому что вы не позволяете платформе вызывать S3MessageHandler.handleMessage(). Здесь вы просто вызываете этот метод private во время выполнения, чтобы создать экземпляр S3MessageHandler для каждого сообщения запроса: MethodInvokingMessageHandler вызывает вашу лямбду в ее handleMessage(), и это все - с S3 ничего не произойдет.

* * * * * * * * * * * * * * * * * * * * * ValueExpression здесь не может вам помочь, потому что вам нужно сравнить файл назначения с каждым отдельным сообщением запроса. Поэтому вам нужно выражение времени выполнения. В самом деле new SpelExpressionParser().parseExpression() нет ничего плохого. Просто потому, что у нас нет выбора, и мы должны иметь только один S3MessageHandler без состояния и не создавать его заново во время выполнения при каждом запросе, как вы пытаетесь достичь с помощью этой подозрительной лямбды и ValueExpression.