Запрограммируйте SSE-клиент на основе WebFlux с помощью функций connect - PullRequest
Новая
       46

Запрограммируйте SSE-клиент на основе WebFlux с помощью функций connect

1 голос
/ 14 марта 2019

Я разрабатываю SSE-клиент с использованием WebFlux.Чего я хочу добиться, так это настроить клиента таким образом, чтобы

  1. он автоматически переподключался к SSE-серверу, если SSE-сервер был временно недоступен
  2. , он могтакже подключаться во время выполнения, если SSE-сервер не присутствовал с самого начала.

Однако, насколько я понимаю клиенты на основе JavaScript, они в основном способны выполнять (1)и (2).

Мой вопрос: есть ли способ настроить клиент WebFlux для отображения желаемого поведения (1) и (2) (в идеале «декларативно» с помощью компоновщика и свободного стиля кодирования)?- Я не нашел никакой информации о том, как это сделать, однако я предполагаю, что это стандартные требования для SSE-клиента.

Вот основной код, который я использую для своих тестов: 

        final SslContext sslContext
                = SslContextBuilder
                .forClient()
                .trustManager(InsecureTrustManagerFactory.INSTANCE)
                .build();
        final HttpClient httpClient
                = HttpClient
                .create()
                .secure(t -> t.sslContext(sslContext));
        final WebClient client
                = WebClient
                .builder()
                .clientConnector(new ReactorClientHttpConnector(httpClient))
                .build();
        final ParameterizedTypeReference<ServerSentEvent<String>> type  = new ParameterizedTypeReference<ServerSentEvent<String>>() {};

        final Flux<ServerSentEvent<String>> eventStream
                = client.get()
                .uri(endpointURL)
                .exchange()
                .flatMapMany(it -> it.bodyToFlux(type))
                .repeat();

        eventStream
            .subscribe(
                this::updateInformation
                , error -> LOGGER.error("Error - {}", error.getMessage())
                , () -> LOGGER.info("Got event stream completion signal."));
...