Использование контекста реактора для получения результатов от цепочки downstram

Получил вот такой боб:

@Bean("reactiveNeo4jDatabaseNameProvider")
public ReactiveDatabaseSelectionProvider reactiveDatabaseSelectionProvider() {

    log.info("Database selection step");

    return () -> getRealmFromContext()
            .flatMap(realmId -> {
                Flux<String> stringFlux = translateRealmToDbName(realmId);
                Mono<DatabaseSelection> next = stringFlux.map(DatabaseSelection::byName).next();
                        return next;
                    }
            )
            .switchIfEmpty(Mono.just(DatabaseSelection.undecided()));
}

чья задача - выбрать имя базы данных во время выполнения, читая некоторые данные, заключенные в контексте Reactor, заполненные функцией фильтра webfilter:

public Mono<Void> filter(ServerWebExchange serverWebExchange, WebFilterChain webFilterChain) {

    ....


    return webFilterChain.filter(serverWebExchange)
            .subscriberContext(e -> e.put("realmId", "some string"));

Все работает, как ожидалось: я могу читать из контекста в bean-компоненте селектора db, но мне нужно передать выбор обратно в веб-фильтр, который должен добавить определенный заголовок в окончательный ответ. Я думал что-то вроде этого

serverWebExchange.getResponse().beforeCommit(() -> {
    Mono.subscriberContext().doOnNext(ctx, {
       // fetch the selected db value here and add the header
    });
});

Проблема в том, как я могу добавить что-то новое в контекст и уловить это функцией, описанной выше?

РЕДАКТИРОВАТЬ: функция getRealmFromContext () подписывается на контекст и сопоставляет его, чтобы получить значение области.

Лучший,

FB


spring project-reactor spring-reactive spring-data-neo4j
person FrankBr    schedule 22.10.2020    source источник

Ответы (1)


arrow_upward
0
arrow_downward

Ну, покопавшись, не получилось так, как я ожидал; Я просто решил проблему

Mono.zip(Mono.just(serverWebExchange.getResponse()), selectionDbService.translateRealmToDbName(realmId))
        .subscribe(objects -> objects.getT1().beforeCommit(() -> {
            serverWebExchange.getResponse().getHeaders().add(COMPANY_HEADER_NAME_RESPONSE, objects.getT2());
            return Mono.empty();
        }) );

Чтобы не вызывать службу дважды, я использовал систему кеширования.

person FrankBr    schedule 28.10.2020