何时在服务器上产生事件，并将mediatype.text_event_stream传递给客户端上的订阅服务器

Question

我创建了一个示例客户机/服务器应用程序来熟悉Spring WebFlux/Reactor Netty。现在，当响应包含Flux并且媒体类型是“text/event-stream”时，我对客户端的行为有点困惑。我看到的是，服务器上产生的每个元素都被立即发送到客户机，但还没有交付给订户。在服务器端的生产者完成流量之后，第一次交付给订阅者。对我来说，这意味着所有元素首先在客户端的reactor-netty中收集，直到它得到一个完整/错误事件。

我的结论是正确的还是我会做错什么？如果是真的，这在不久的将来会改变吗？根据我目前观察到的行为，使用Spring Webflux的大多数好处都被否定了，因为与Spring Mvc一样，消费者必须等到创建并传输了整个元素集合之后才能开始处理元素。

我的服务器应用程序是:`

@SpringBootApplication
public class ServerApp {
    public static void main(String[] args) {
        new SpringApplicationBuilder().sources(ServerApp.class).run(args);
    }

    @RestController
    public static class TestController {
        @GetMapping(value = "/test", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
        public Flux<String> testFlux() {
            class AsyncSink implements Consumer<SynchronousSink<String>> {
                private List<String> allStrings = List.of(
                        "Hello Flux1!",
                        "Hello Flux2!",
                        "Hello Flux3!",
                        "Hello Flux4!",
                        "Hello Flux5!");
                private int index = 0;

                @Override
                public void accept(SynchronousSink<String> sink) {
                    if (index == allStrings.size()) {
                        sink.complete();
                    }
                    else {
                        sink.next(allStrings.get(index++));
                    }
                }

            }

            return Flux.generate(new AsyncSink());
        }
    }
}

@SpringBootApplication
public class ClientApp {
    public static void main(String[] args) throws IOException {
        ConfigurableApplicationContext aContext = new SpringApplicationBuilder().web(WebApplicationType.NONE).sources(ClientApp.class).run(args);

        Flux<String> aTestFlux = aContext.getBean(TestProxy.class).getFlux();
        aTestFlux.subscribe(new TestSubscriber());

        System.out.println("Press ENTER to exit.");
        System.in.read();
    }

    @Bean
    public WebClient webClient() {
        return WebClient.builder().baseUrl("http://localhost:8080").build();
    }

    @Component
    public static class TestProxy {
        @Autowired
        private WebClient webClient;

        public Flux<String> getFlux() {
            return webClient.get().uri("/test").accept(MediaType.TEXT_EVENT_STREAM).exchange().flatMapMany(theResponse -> theResponse.bodyToFlux(String.class));
        }
    }

    private static class TestSubscriber extends BaseSubscriber<String> {
        @Override
        public void hookOnSubscribe(Subscription subscription) {
            System.out.println("Subscribed");
            request(Long.MAX_VALUE);
        }

        @Override
        public void hookOnNext(String theValue) {
            System.out.println(" - " + theValue);
            request(1);
        }

        @Override
        protected void hookOnComplete() {
            System.out.println("   done");
        }

        @Override
        protected void hookOnCancel() {
            System.out.println("   cancelled");
        }

        @Override
        protected void hookOnError(Throwable theThrowable) {
            theThrowable.printStackTrace(System.err);
        }
    }
}

data:Hello Flux1!

data:Hello Flux2!

data:Hello Flux3!

data:Hello Flux4!

data:Hello Flux5!

Answer 1

取自反应性文档并改写以满足您的需要。

我的猜测是，在您的示例中，您给了generate函数传递一个使用者，当完成时将发出这个使用者。

通过使用flux#generate(callable stateSupplier、bifunction 、s>generater) 方法，您提供了一个包含您想要发出的项的状态，然后在提供的 bifunction中逐个发出每个项。

 
  Flux<String> flux = Flux.generate(
    () -> List.of("1!", "2!", "3!", "4!", "5!"), 
    (state, sink) -> {
        if (index == allStrings.size()) {
            sink.complete();
        } else {
          sink.next(state.get(index++));
        } 
    });