Spring响应式文件集成

Question

我正在尝试使用 spring-integration-file 来轮询目录并从放置在该目录中的文件创建反应流。这在大多数情况下都有效，但是当我放置文件但没有订户时，我会遇到异常。为了演示这个问题，我编写了一个小型演示应用程序：

import org.reactivestreams.Publisher;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.http.MediaType;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.file.dsl.Files;
import org.springframework.messaging.Message;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

import java.io.File;

@SpringBootApplication
@RestController
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }

    @Bean
    public Publisher<Message<File>> reactiveSource() {
        return IntegrationFlows
                .from(Files.inboundAdapter(new File("."))
                                .patternFilter("*.csv"),
                        e -> e.poller(Pollers.fixedDelay(1000)))
                .channel("processFileChannel")
                .toReactivePublisher();
    }

    @GetMapping(value = "/files", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> files() {
        return Flux.from(reactiveSource())
                .map(message -> message.getPayload().getAbsolutePath());
    }
}

因此，如果我现在对 localhost:8080/files 进行 curl，然后将 csv 文件放在项目的根目录中，一切都很好，我会看到文件的路径作为对我的 curl 的响应。但是当我不做卷曲然后将文件放在根目录中时，我得到以下异常：

java.lang.IllegalStateException: The [bean 'reactiveSource.channel#0'; defined in: 'com.example.demo.DemoApplication'; 
from source: 'bean method reactiveSource'] doesn't have subscribers to accept messages
        at org.springframework.util.Assert.state(Assert.java:97)
        at org.springframework.integration.channel.FluxMessageChannel.doSend(FluxMessageChannel.java:61)
        at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:570)
        ... 38 more

我认为响应式流的属性之一是，当没有订阅者时，由于流是惰性的，因此流不会启动。但显然情况并非如此。如果没有订阅者，有人可以解释我需要做什么才能让流无法启动吗？

Answer 1

如果您使用最新版本之一，那么您可以使用FluxMessageChannel 频道而不是DirectChannel 来代替"processFileChannel"。这样，SourcePollingChannel 适配器将变为反应式，并且确实不会轮询源，直到该 FluxMessageChannel 发生订阅。

然后，您可以从此FluxMessageChannel 在您的files() API 中创建一个Flux - 在.toReactivePublisher() 中不需要。

在文档中查看更多信息：https://docs.spring.io/spring-integration/docs/current/reference/html/reactive-streams.html#source-polling-channel-adapter

关键是.toReactivePublisher() 恰好在此时将集成流作为Publisher。在此之前的一切都是常规的、命令式的，并且独立于下游逻辑工作。

更新

类似这样的：

@Bean
FluxMessageChannel filesChannel() {
    return new FluxMessageChannel();
}

@Bean
public IntegrationFlow reactiveSource() {
        return IntegrationFlows
                .from(Files.inboundAdapter(new File("."))
                                .patternFilter("*.csv"),
                        e -> e.poller(Pollers.fixedDelay(1000)))
                .channel(filesChannel())
                .get();
    }

@GetMapping(value = "/files", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> files() {
        return Flux.from(filesChannel())
                .map(message -> ((File) message.getPayload()).getAbsolutePath());
    }