Mon go ChangeStream с Spring вызывает ошибку: состояние должно быть: ошибка открытия после закрытия источника событий пользовательского интерфейса

Question

Я пытаюсь устранить ошибку mtbodb changestream. Я наблюдаю за изменениями в коллекции с ...

public Flux<Clone> watchForCloneCollectionChanges() {

            // set changestream options to watch for any changes to the clone collection
            Criteria criteria = new Criteria();
            criteria.orOperator(where("operationType").is("insert"), where("operationType").is("replace"));
            ChangeStreamOptions options = ChangeStreamOptions.builder()
                    .filter(Aggregation.newAggregation(Clone.class,
                            Aggregation.match(criteria)
                    )).returnFullDocumentOnUpdate().build();

            // return a flux that watches the changestream and returns the full document
            return reactiveMongoTemplate.changeStream("clone", options, Clone.class)
                    .map(ChangeStreamEvent::getBody)
                    .doOnError(throwable -> logger.error("Error with the teams changestream event: " + throwable.getMessage(), throwable));
    }

, используя обработчик

public Mono<ServerResponse> watchClones(ServerRequest request) {
        Flux<ServerSentEvent<Clone>> sse = this.cloneWatcher.watchForCloneCollectionChanges()
                .map(clone -> ServerSentEvent.<Clone>builder()
                        .data(clone)
                        .build());
        return ServerResponse.ok().body(BodyInserters.fromServerSentEvents(sse));

и функцию маршрутизатора

    @Bean
    public RouterFunction<ServerResponse> route(CloneHandler cloneHandler) {
        return RouterFunctions
                .route(RequestPredicates.GET("/clones"), cloneHandler::getClones)
                .andRoute(RequestPredicates.GET("/clones/watch"), cloneHandler::watchClones);

    }

Для моего интерфейса Я использую Angular, используя темы поведения и источник событий для подключения. Проблема, с которой я сталкиваюсь, заключается в том, что после закрытия страницы changeStream все еще следит за изменениями, даже если внешний интерфейс закрыл источник события. В результате при обновлении документа пружина выдает следующую ошибку:

2020-01-06 15: 22: 34.312 ОШИБКА 27772 --- [andler-executor] org.mongodb.driver.operation : Обратный вызов при вызове onResult выдал ошибку

com.mongodb.MongoException: состояние должно быть: open at com.mongodb.MongoException.fromThrowableNonNull (MongoException. java: 79) ~ [mongodb-driver-core-3.11 .1.jar: na] at com.mongodb.asyn c .client.AbstractSubscription.onError (AbstractSubscription. java: 135) ~ [mongodb-driver-asyn c -3.11.1.jar: na] в com.mongodb.asyn c .client.MongoIterableSubscription $ 2.onResult (MongoIterableSubscription. java: 93) ~ [mongodb-driver-asyn c -3.11.1.jar: na] в com.mongodb.asyn c .client. MongoIterableSubscription AsyncChangeStreamBatchCursor. java: 133) ~ [mongodb-driver-core-3.11.1.jar: na] на com.mongodb.operatio n.AsyncChangeStreamBatchCursor $ 3.onResult (AsyncChangeStreamBatchCursor. java: 129) ~ [mongodb-driver-core-3.11.1.jar: na] по адресу com.mongodb.internal.asyn c. 1033 *: 49) [mongodb-driver-core-3.11.1.jar: na] на com.mongodb.operation.AsyncChangeStreamBatchCursor $ 4.onResult (AsyncChangeStreamBatchCursor. java: 168) [mongodb-driver-core-3.11.1 .jar: na] на com.mongodb.operation.AsyncChangeStreamBatchCursor $ 4.onResult (AsyncChangeStreamBatchCursor. java: 159) [mongodb-driver-core-3.11.1.jar: na] на com.mongodb.internal.asyn c .ErrorHandlingResultCallback.onResult (ErrorHandlingResultCallback. java: 49). [mongodb-driver-core-3.11.1.jar: na] на com.mongodb.operation.AsyncQueryBatchCursor $ CommandResultSingleResultCallback.onResult (AsyncQueryBatchCursor. java: 310) [mongodb-driver-co re-3.11.1.jar: na] в com.mongodb.internal.asyn c .ErrorHandlingResultCallback.onResult (ErrorHandlingResultCallback. java: 49) [mongodb-driver-core-3.11.1.jar: na] в com.mongodb.internal.connection.DefaultServer $ DefaultServerProtocolExecutor $ 2.onResult (DefaultServer. java: 242) [mongodb-driver-core-3.11.1.jar: na] в com.mongodb.internal.asyn c. ErrorHandlingResultCallback.onResult (ErrorHandlingResultCallback. java: 49) [mongodb-driver-core-3.11.1.jar: na] в com.mongodb.internal.connection.CommandProtocolImpl $ 1.onResult (CommandProtocolImpl. java: 83) mongodb-driver-core-3.11.1.jar: na] на com.mongodb.internal.connection.DefaultConnectionPool $ PooledConnection $ 1.onResult (DefaultConnectionPool. java: 467) [mongodb-driver-core-3.11.1.jar : na] at com.mongodb.internal.connection.UsageTrackingInternalConnection $ 2.onResult (UsageTrackingInternalConnection. java: 111) [mongodb-driver-core-3.11.1.jar: na] в com.mongodb.internal.asyn c .ErrorHandlingResultCallback.onResult (ErrorHandlingResul tCallback. java: 49) [mongodb-driver-core-3.11.1.jar: na] по адресу com.mongodb.internal.connection.InternalStreamConnection $ 2 $ 1.onResult (InternalStreamConnection. java: 401) [mongodb-driver -core-3.11.1.jar: na] на com.mongodb.internal.connection.InternalStreamConnection $ 2 $ 1.onResult (InternalStreamConnection. java: 376) [mongodb-driver-core-3.11.1.jar: na] в ком. mongodb.internal.connection.InternalStreamConnection $ MessageHeaderCallback $ MessageCallback.onResult (InternalStreamConnection. java: 677) [mongodb-driver-core-3.11.1.jar: na] в сообщении com.mongodb.internal.connection.InternalStreamConneback $ $ MessageCallback.onResult (InternalStreamConnection. java: 644) [mongodb-driver-core-3.11.1.jar: na] в com.mongodb.internal.connection.InternalStreamConnection $ 5.completed (InternalStreamConnection. java: 514) [ mongodb-driver-core-3.11.1.jar: na] на com.mongodb.internal.connection.InternalStreamConnection $ 5.completed (InternalStreamConnection. java: 511) [mongodb-driver-core-3.11.1.jar: na ] at com.mongodb.internal.connection.AsynchronousChannelStream $ BasicCompletionHandler.completed (AsynchronousChannelStream. java: 220) [mongodb-driver-core-3.11.1.jar: na] в com.mongodb.internal.connection.AsynchronousChannelream $ BasicCompletionHandler.completed (AsynchronousChannelStream. java: 203) [mongodb-driver-core-3.11.1.jar: na] на com.mongodb.intern al.connection.tlschannel.asyn c .AsynchronousTlsChannel $ 3 $ 1.run (AsynchronousTlsChannel. java: 151) [mongodb-driver-core-3.11.1.jar: na] в java .util.concurrent.Executors $ RunnableAdapter.call (Исполнители. java: 511) [na: 1.8.0_172] в java .util.concurrent.FutureTask.run $$$ capture (FutureTask. java: 266) [na: 1.8. 0_172] в java .util.concurrent.FutureTask.run (FutureTask. java) [na: 1.8.0_172] в java .util.concurrent.ThreadPoolExecutor.runWorker (ThreadPoolExecutor. java: 1149) [ na: 1.8.0_172] в java .util.concurrent.ThreadPoolExecutor $ Worker.run (ThreadPoolExecutor. java: 624) [na: 1.8.0_172] в java .lang.Thread.run (Thread. java: 748) [na: 1.8.0_172] Причина: java .lang.IllegalStateException: состояние должно быть: open at com.mongodb.assertions.Assertions.isTrue (Утверждения. java: 70) ~ [mongodb -driver-core-3.11.1.jar: na] на com.mongodb.internal.session.BaseClientSessionImpl.advanceOperationTime (BaseClientSessionImpl. java: 107) ~ [mongodb-driver-core-3.11.1.jar: na] на com.mongodb.internal .session.ClientSessionContext.advanceOperationTime (ClientSessionContext. java: 70) ~ [mongodb-driver-core-3.11.1.jar: na] в com.mongodb.internal.connection.ClusterClockAdvancingSessionContext.advanceOperationTime (ClusterClock: 76) ~ [mongodb-driver-core-3.11.1.jar: na] на com.mongodb.internal.connection.InternalStreamConnection.updateSessionContext (InternalStreamConnection. java: 537) ~ [mongodb-driver-core-3.11.1 .jar: na] at >> com.mongodb.internal.connection.InternalStreamConnection.access $ 800 (InternalStreamConnection. java: 76) ~ [mongodb-driver-core-3.11.1.jar: na] at >> com. mongodb.internal.connection.InternalStreamConnection $ 2 $ 1.onResult (InternalStreamConnection. java: 385) [mongodb-driver-core-3.11.1.jar: na] ... 14 общих кадров опущено

Я искал высоко и низко, но не могу найти решение. Любая помощь будет оценена! Спасибо!