I'm trying to setup a dynamic IntegrationFlow to read files from a local directory. Therefore I've created the following parameterizable class:
@Configuration
public class FileConsoleFlow {
@Bean("flowBuilderFactory")
Function<Properties, IntegrationFlow> flowBuilderFactory() {
return (properties) -> new SimpleFlowBuilder(properties).build();
}
static class SimpleFlowBuilder {
private Properties properties;
public FlowBuilder(Properties properties) {
this.properties = properties;
}
public IntegrationFlow build() {
return IntegrationFlows.from(localSource(), c -> c.poller(pollerMetadata()))
.handle(message -> System.out.println(message))
.get();
}
public FileReadingMessageSource localSource() {
CompositeFileListFilter<File> filters = new CompositeFileListFilter<>();
filters.addFilter(new SimplePatternFileListFilter(properties.getProperty("local-file-pattern")));
filters.addFilter(new AcceptOnceFileListFilter<>(10000));
RecursiveDirectoryScanner scanner = new RecursiveDirectoryScanner();
scanner.setFilter(filters);
FileReadingMessageSource fileSource = new FileReadingMessageSource();
fileSource.setDirectory(new File(properties.getProperty("local-directory")));
fileSource.setScanner(scanner);
return fileSource;
}
public PollerMetadata pollerMetadata() {
return Pollers.fixedDelay(1000)
.maxMessagesPerPoll(100)
.advice(transactionInterceptor())
.transactionSynchronizationFactory(transactionSynchronizationFactory())
.get();
}
public TransactionInterceptor transactionInterceptor() {
return new TransactionInterceptorBuilder()
.transactionManager(txManager())
.build();
}
public TransactionManager txManager() {
return new PseudoTransactionManager();
}
public TransactionSynchronizationFactory transactionSynchronizationFactory() {
ExpressionEvaluatingTransactionSynchronizationProcessor processor =
new ExpressionEvaluatingTransactionSynchronizationProcessor();
SpelExpressionParser spelParser = new SpelExpressionParser();
processor.setAfterCommitExpression(spelParser.parseExpression("payload.delete()"));
processor.setAfterRollbackExpression(spelParser.parseExpression("payload.renameTo(new java.io.File(payload.absolutePath + '.FAILED'))"));
return new DefaultTransactionSynchronizationFactory(processor);
}
}
}
The parameterized flows are registered with the help of the IntegrationFlowContext as follows:
...
integrationFlowContext.registration(flow)
.autoStartup(false)
.id("flow-" + id)
.register();
...
integrationFlowContext.getRegistry().forEach((name, flow) -> flow.start());
After starting up the application I get the following warning:
2022-02-08 14:11:41.951 WARN [,1e3c4776ef54251e,1e3c4776ef54251e] 1 --- [ scheduling-1] o.s.i.expression.ExpressionUtils : Creating EvaluationContext with no beanFactory
java.lang.RuntimeException: No beanFactory
at org.springframework.integration.expression.ExpressionUtils.createStandardEvaluationContext(ExpressionUtils.java:90) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor.createEvaluationContext(ExpressionEvaluatingTransactionSynchronizationProcessor.java:218) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor.prepareEvaluationContextToUse(ExpressionEvaluatingTransactionSynchronizationProcessor.java:202) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor.doProcess(ExpressionEvaluatingTransactionSynchronizationProcessor.java:141) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor.processAfterCommit(ExpressionEvaluatingTransactionSynchronizationProcessor.java:125) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory$DefaultTransactionalResourceSynchronization.processResourceAfterCommit(DefaultTransactionSynchronizationFactory.java:79) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory$DefaultTransactionalResourceSynchronization.processResourceAfterCommit(DefaultTransactionSynchronizationFactory.java:53) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.transaction.support.ResourceHolderSynchronization.afterCommit(ResourceHolderSynchronization.java:87) ~[spring-tx-5.3.15.jar:5.3.15]
at org.springframework.transaction.support.TransactionSynchronizationUtils.invokeAfterCommit(TransactionSynchronizationUtils.java:135) ~[spring-tx-5.3.15.jar:5.3.15]
at org.springframework.transaction.support.TransactionSynchronizationUtils.triggerAfterCommit(TransactionSynchronizationUtils.java:123) ~[spring-tx-5.3.15.jar:5.3.15]
at org.springframework.transaction.support.AbstractPlatformTransactionManager.triggerAfterCommit(AbstractPlatformTransactionManager.java:936) ~[spring-tx-5.3.15.jar:5.3.15]
at org.springframework.transaction.support.AbstractPlatformTransactionManager.processCommit(AbstractPlatformTransactionManager.java:782) ~[spring-tx-5.3.15.jar:5.3.15]
at org.springframework.transaction.support.AbstractPlatformTransactionManager.commit(AbstractPlatformTransactionManager.java:711) ~[spring-tx-5.3.15.jar:5.3.15]
at org.springframework.transaction.interceptor.TransactionAspectSupport.commitTransactionAfterReturning(TransactionAspectSupport.java:654) ~[spring-tx-5.3.15.jar:5.3.15]
at org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:407) ~[spring-tx-5.3.15.jar:5.3.15]
at org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:119) ~[spring-tx-5.3.15.jar:5.3.15]
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.3.15.jar:5.3.15]
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:215) ~[spring-aop-5.3.15.jar:5.3.15]
at jdk.proxy2/jdk.proxy2.$Proxy101.call(Unknown Source) ~[na:na]
at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:413) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$4(AbstractPollingEndpoint.java:348) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50) ~[spring-core-5.3.15.jar:5.3.15]
at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$5(AbstractPollingEndpoint.java:341) ~[spring-integration-core-5.5.8.jar:5.5.8]
at org.springframework.cloud.sleuth.instrument.async.TraceRunnable.run(TraceRunnable.java:64) ~[spring-cloud-sleuth-instrumentation-3.1.0.jar:3.1.0]
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54) ~[spring-context-5.3.15.jar:5.3.15]
at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:95) ~[spring-context-5.3.15.jar:5.3.15]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[na:na]
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[na:na]
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[na:na]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[na:na]
at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]
If I understood things correctly, the IntegrationFlowContext is responsible for the creation of the integration-beans (and thus the setting of the beanFactory). But then why do I get this error message?
The error means that TransactionSynchronizationFactory must be declared as a bean and therefore its BeanFactory requirements would be fulfilled.
Please, revise all your volatile objects and consider to make some of them as global singleton beans.
I believe all of these could go to the main configuration, not dynamic: PollerMetadata, TransactionInterceptor, TransactionManager and TransactionSynchronizationFactory. The dynamic part in your solution is really just that FileReadingMessageSource.
It is better to use as less dynamic as possible for better performance and proper reusability. You also would avoid duplications whenever they are not needed.
I would even move that .handle(message -> System.out.println(message)) to the separate global singleton IntegrationFlow and use only a channel reference from this one to avoid extra duplication.