Как вы используете WebFlux для анализа потока событий, который не соответствует отправленным событиям сервера? - PullRequest
Новая
       3

Как вы используете WebFlux для анализа потока событий, который не соответствует отправленным событиям сервера?

0 голосов
/ 08 февраля 2019

Я пытаюсь использовать WebClient для работы с конечной точкой Docker /events.Однако он не соответствует договору text/eventstream, поскольку каждое сообщение разделяется 2 LF.Он просто отправляет его как один JSON-документ, за которым следует другой.

Он также устанавливает тип MIME на application/json, а не text/eventstream.

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

1 Ответ

0 голосов
/ 08 февраля 2019

Вместо того, чтобы пытаться обработать ServerSentEvent, просто получите его как String.Затем попытайтесь проанализировать его как JSON (игнорирование тех, которые терпят неудачу, что, как я предполагаю, может произойти, но я сам не попал) 

@PostConstruct
public void setUpStreamer() {
    final Map<String, List<String>> filters = new HashMap<>();
    filters.put("type", Collections.singletonList("service"));

    WebClient.create(daemonEndpoint)
        .get()
        .uri("/events?filters={filters}",
           mapper.writeValueAsString(filters))
        .retrieve()
        .bodyToFlux(String.class)
        .flatMap(Mono::justOrEmpty)
        .map(s -> {
            try {
                return mapper.readValue(s, Map.class);
            } catch (IOException e) {
                log.warn("unable to parse {} as JSON", s);
                return null;
            }
        })
        .flatMap(Mono::justOrEmpty)
        .subscribe(
            event -> {
                log.trace("event={}", event);
                refreshRoutes();
            },
            throwable -> log.error("Error on event stream: {}", throwable.getMessage(), throwable),
            () -> log.warn("event stream completed")
        );
}
...