无限通量无限通量或者：如何在不完成焊剂的情况下重新连接到websocket蒸汽

如何解决无限通量无限通量或者：如何在不完成焊剂的情况下重新连接到websocket蒸汽

我不确定我想做的事是否接近最佳实践。如果没有，我想知道！

我正在构建一个对websocket上发布的外部事件做出反应的系统。身份验证之后，我可以请求发出事件并以单个通量侦听它们。我可以将执行器连接到该助焊剂上来做事（想想家庭自动化，按一下按钮就可以开灯）。

当websocket连接断开时会出现问题：磁通完成，我必须在新连接上将所有监听执行器重新连接到新的磁通上。我认为，在与新事件建立新连接时，无限通量再次开始发射可能会更容易（当然，在启动新连接时必须再次进行身份验证和请求事件）。

我想出了一种解决方案，可以解决我想要的事情，但这感觉像是个大麻烦：

    private Flux<String> finite() {
        return Flux.interval(Duration.ofMillis(10)).handle((i,sink) -> {
            if (i == 10) {
                sink.complete();
            } else {
                sink.next(i);
            }
        }).map(it -> "hi :" + it).take(100);
    }

    public static class Holder {
        @Setter
        FluxSink<Object> sink;

        public void next(EmitterProcessor<String> output) {
            sink.next(output);
        }
    }


    EmitterProcessor<String> output;
    Holder holder;

    @DisplayName("Should create infinite flux of finite fluxes")
    @Test
    public void infiniteFlux() throws InterruptedException {

        AtomicInteger count1 = new AtomicInteger(0);
        AtomicInteger count2 = new AtomicInteger(0);
        holder = new Holder();
        ConnectableFlux flux = Flux.create(fluxSink -> holder.setSink(fluxSink)).map(it -> (Flux) it).flatMap(it -> it).publish();

        flux.connect();
        flux.subscribe((it) -> count1.incrementAndGet());
        Thread.sleep(150);
        output = EmitterProcessor.create();
        finite().subscribeWith(output);
        holder.next(output);

        Thread.sleep(150);
        flux.subscribe((it) -> count2.incrementAndGet());
        output = EmitterProcessor.create();
        finite().subscribeWith(output);
        holder.next(output);
        Thread.sleep(150);
        assertEquals(20,count1.get());
        assertEquals(10,count2.get());

    }

这是否可以代替处理器来完成？还是我应该忘记它，并在完成websocket连接后重新连接一切？

这是我当前的websocket实现。我不太喜欢在开始收听时需要“知道”所有类型的事件，但这是因为我仍然不确定100％如何使用websockets处理发送和接收

    public void startListening(Flux<HaEventRequest> eventRequests) {
        Mono<String> login = Mono.just(loginPayload());
        Flux<String> subscribe = eventRequests
                .doOnNext(req -> log.info("Registering: " + req.getEventType()))
                .map(Json::write);
        Flux<String> input = login.concatWith(subscribe);

        EmitterProcessor<HaEventResponse> output = EmitterProcessor.create();

        Mono<Void> sessionMono = client.execute(URI.create(wsUrl),session -> session.send(input.map(session::textMessage))
                                                                                     .thenMany(session.receive()
                                                                                                      .map(WebSocketMessage::getPayloadAsText)
                                                                                                      .map(message -> Json.read(message,HaEventResponse.class))
                                                                                                      .filter(eventResponse -> eventResponse.getEvent() != null)
                                                                                                      .subscribeWith(output)
                                                                                                      .then())
                                                                                     .then());

        eventFlux = output.doOnSubscribe(s -> sessionMono.subscribe())
                          .onErrorContinue((throwable,o) -> log.error("Error occured during web request. Dumping it: ",throwable));
    }


    public Flux<HaEventResponse> streamEvents(String eventType) {
        return eventFlux.filter(eventResponse -> eventResponse.getEvent()
                                                              .getEventType()
                                                              .equals(eventType));
    }