Маршрут Spring Integration Flow на основе имени файла - PullRequest
Новая
       18

Маршрут Spring Integration Flow на основе имени файла

0 голосов
/ 29 августа 2018

У меня есть требование распаковать и сохранить файл и обработать его содержимое. Внутри zip-файла может быть два типа файла: индивидуальный или фирменный. Это можно отличить по имени файла. После обработки всех файлов следует вызвать другой программный модуль, а также архивировать обработанный файл в другом месте. Хотел бы использовать Spring интеграции для того же. Я пытаюсь добиться этого с помощью следующего кода, но это создает проблемы при маршрутизации на основе имени файла. Я использую JDK 8, Spring 5 

.<File, Boolean>route(new Function<File, Boolean>() {

                    @Override
                    public Boolean apply(File f) {
                        return f.getName().contains("individual");
                    }
                }, m -> m
                        .subFlowMapping(true, sf -> sf.gateway(individualProcessor()))
                        .subFlowMapping(false, sf -> sf.gateway(firmProcessor()))
                )

Исключение 

Caused by: java.lang.IllegalArgumentException: Found ambiguous parameter type [interface java.util.function.Function] for method match: [public default <V> java.util.function.Function<V, R> java.util.function.Function.compose(java.util.function.Function<? super V, ? extends T>), public static <T> java.util.function.Function<T, T> java.util.function.Function.identity(), public java.lang.Boolean com.xxx.thirdpatysystem.config.IntegrationConfig$1.apply(java.io.File)]
    at org.springframework.util.Assert.isNull(Assert.java:155) ~[spring-core-5.0.4.RELEASE.jar:5.0.4.RELEASE]
    at org.springframework.integration.util.MessagingMethodInvokerHelper.findHandlerMethodsForTarget(MessagingMethodInvokerHelper.java:843) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
    at org.springframework.integration.util.MessagingMethodInvokerHelper.<init>(MessagingMethodInvokerHelper.java:362) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
    at org.springframework.integration.util.MessagingMethodInvokerHelper.<init>(MessagingMethodInvokerHelper.java:231) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
    at org.springframework.integration.util.MessagingMethodInvokerHelper.<init>(MessagingMethodInvokerHelper.java:225) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
    at org.springframework.integration.handler.MethodInvokingMessageProcessor.<init>(MethodInvokingMessageProcessor.java:60) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
    at org.springframework.integration.router.MethodInvokingRouter.<init>(MethodInvokingRouter.java:46) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
    at org.springframework.integration.dsl.IntegrationFlowDefinition.route(IntegrationFlowDefinition.java:1922) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
    at org.springframework.integration.dsl.IntegrationFlowDefinition.route(IntegrationFlowDefinition.java:1895) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]

даже я попробовал ниже 

.<File, Boolean>route(f -> f.getName().contains("individual"), m -> m
                        .subFlowMapping(true, sf -> sf.gateway(individualProcessor()))
                        .subFlowMapping(false, sf -> sf.gateway(firmProcessor()))
                )

Исключение 

Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.integration.dsl.IntegrationFlow]: Factory method 'thirdpatysystemFlow' threw exception; nested exception is java.lang.UnsupportedOperationException
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:185) ~[spring-beans-5.0.4.RELEASE.jar:5.0.4.RELEASE]
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:579) ~[spring-beans-5.0.4.RELEASE.jar:5.0.4.RELEASE]
    ... 17 common frames omitted
Caused by: java.lang.UnsupportedOperationException: null
    at org.springframework.integration.dsl.StandardIntegrationFlow.configure(StandardIntegrationFlow.java:89) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
    at org.springframework.integration.dsl.IntegrationFlowDefinition.gateway(IntegrationFlowDefinition.java:2172) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]
    at org.springframework.integration.dsl.IntegrationFlowDefinition.gateway(IntegrationFlowDefinition.java:2151) ~[spring-integration-core-5.0.3.RELEASE.jar:5.0.3.RELEASE]

Весь фрагмент кода ниже 

import java.io.File;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.dsl.channel.MessageChannels;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.FileWritingMessageHandler;
import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
import org.springframework.integration.file.filters.ChainFileListFilter;
import org.springframework.integration.file.filters.RegexPatternFileListFilter;
import org.springframework.integration.zip.splitter.UnZipResultSplitter;
import org.springframework.integration.zip.transformer.UnZipTransformer;
import org.springframework.integration.zip.transformer.ZipResultType;
import org.springframework.messaging.MessageHandler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

/**
 * @author dpoddar
 *
 */
@Configuration
@EnableIntegration
public class IntegrationConfig {

    @Value("${input.directory}")
    private String inputDir;

    @Value("${outputDir.directory}")
    private String outputDir;

    @Value("${input.scan.frequency: 100}")
    private long scanFrequency;

    @Bean
    public MessageSource<File> inputFileSource() {
        FileReadingMessageSource src = new FileReadingMessageSource();

        src.setDirectory(new File(inputDir));
        src.setAutoCreateDirectory(true);

        ChainFileListFilter<File> chainFileListFilter = new ChainFileListFilter<>();
        chainFileListFilter.addFilter(new AcceptOnceFileListFilter<>() );
        chainFileListFilter.addFilter(new RegexPatternFileListFilter("(?i)^.+\\.zip$"));
        src.setFilter(chainFileListFilter);
        return src;
    }

    @Bean
    public UnZipTransformer unZipTransformer() {
        UnZipTransformer unZipTransformer = new UnZipTransformer();
        unZipTransformer.setExpectSingleResult(false);
        unZipTransformer.setZipResultType(ZipResultType.FILE);
        //unZipTransformer.setWorkDirectory(new File("/usr/tmp/uncompress"));
        unZipTransformer.setDeleteFiles(true);
        return unZipTransformer;
    }

    @Bean
    public UnZipResultSplitter splitter() {
        UnZipResultSplitter splitter = new UnZipResultSplitter();

        return splitter;
    }

    @Bean
    public DirectChannel outputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageHandler fileOutboundChannelAdapter() {
        FileWritingMessageHandler adapter = new FileWritingMessageHandler(new File(outputDir));
        adapter.setDeleteSourceFiles(true);
        adapter.setAutoCreateDirectory(true);
        adapter.setExpectReply(false);
        return adapter;
    }

    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(25);
        return executor;
    }

    @Autowired
    DirectChannel outputChannel;

    @Autowired
    MessageHandler fileOutboundChannelAdapter;

    @Bean
    public IntegrationFlow individualProcessor() {
        return flow -> flow.handle("thirdpatysystemprocessor","processfile").channel(outputChannel).handle(fileOutboundChannelAdapter);
    }

    @Bean
    public IntegrationFlow firmProcessor() {
        return flow -> flow.handle("thirdpatysystemprocessor","processfile").channel(outputChannel).handle(fileOutboundChannelAdapter);
    }

    @Bean
    public IntegrationFlow thirdpatysystemAgentDemographicFlow() {
        return IntegrationFlows
                .from(inputFileSource(), spec -> spec.poller(Pollers.fixedDelay(scanFrequency,TimeUnit.SECONDS)))
                .transform(unZipTransformer())
                .split(splitter())
                .channel(MessageChannels.executor(taskExecutor()))
                .<File, Boolean>route(new Function<File, Boolean>() {

                    @Override
                    public Boolean apply(File f) {
                        return f.getName().contains("individual");
                    }
                }, m -> m
                        .subFlowMapping(true, sf -> sf.gateway(individualProcessor()))
                        .subFlowMapping(false, sf -> sf.gateway(firmProcessor()))
                )
                .aggregate()
                /*.handle("thirdpatysystemprocessor","processfile")
                .channel(outputChannel())
                .handle(fileOutboundChannelAdapter())*/
                .get()
                ;
    }
}

1 Ответ

0 голосов
/ 29 августа 2018

* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * java.lang.IllegalArgumentException: Found ambiguous parameter type [interface java.util.function.Function] * * * * * * * * * * * * * * * * * * * * * * * 100 * * * * * * * * * * * * 100 * 

MethodInvokingRouter methodInvokingRouter = isLambda(router)
            ? new MethodInvokingRouter(new LambdaMessageProcessor(router, payloadType))
            : new MethodInvokingRouter(router, ClassUtils.FUNCTION_APPLY_METHOD);

Мы явно указываем на метод apply().

Повторное использование существующих компонентов IntegrationFlow в подпотоках (gateway()) было исправлено в версии 5.0.4: https://jira.spring.io/browse/INT-4434

Итак, вам нужно просто обновить ваш проект до последних зависимостей. В частности Spring Integration 5.0.7: https://spring.io/projects/spring-integration#learn

Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.

Похожие темы

...