如何在春季创建反应堆热流？

Question

我有以下代码。我以为使用发布可以创建一个热流，但是每个连接上的uniqueId是不同的。我希望create方法仅执行一次，然后对任意数量的订阅者作为热流运行。

    private Date lastChecked = Calendar.getInstance().getTime();

    @GetMapping(path = "/news/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<Object> comments(@RequestParam(required=false) String date) throws ParseException {
        if(date != null) {
            lastChecked=new SimpleDateFormat("dd-MM-yyyy").parse(date);
        }
        return Flux.create(fluxSink -> {
            String uniqueID = UUID.randomUUID().toString();
            Flux.interval(Duration.ofSeconds(1))
                .map(s -> {
                    System.out.println(uniqueID);
                    this.newsRepository.findByTimestampGreaterThanOrderByTimestampDesc(lastChecked).take(5).doOnEach(new Consumer<Signal<News>>() {
                        @Override
                        public void accept(Signal<News> commentSignal) {
                            fluxSink.next(commentSignal.get());
                        }
                    }).subscribe();
                    lastChecked = Calendar.getInstance().getTime();
                    lastChecked.setTime(lastChecked.getTime() - 2000);
                    return s;
                })
                .subscribe();
        })
        .publishOn(Schedulers.parallel())
        .onBackpressureBuffer()
        .publish() //executing publish here 
        .autoConnect();
    }

Answer 1

我找到了解决方案，仅使用PostConstruct批注创建一次ConnectableFlux：

    private Date lastChecked = Calendar.getInstance().getTime();
    private ConnectableFlux<Object> flux;

    @PostConstruct
    public void load()
    {
        System.out.println("Loaded");
        this.flux = Flux.create(fluxSink -> {
            String uniqueID = UUID.randomUUID().toString();
            Flux.interval(Duration.ofSeconds(1))
                .map(s -> {
                    System.out.println(uniqueID);
                    this.newsRepository.findByTimestampGreaterThanOrderByTimestampDesc(lastChecked).take(5).doOnEach(new Consumer<Signal<News>>() {
                        @Override
                        public void accept(Signal<News> commentSignal) {
                            fluxSink.next(commentSignal.get());
                        }
                    }).subscribe();
                    lastChecked = Calendar.getInstance().getTime();
                    lastChecked.setTime(lastChecked.getTime() - 2000);
                    return s;
                })
                .subscribe();
        }).publishOn(Schedulers.parallel())
            .onBackpressureBuffer()
            .publish();
    }

    @GetMapping(path = "/news/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<Object> comments(@RequestParam(required=false) String date) throws ParseException {

        return flux.autoConnect();
    }