I am using Spring WebFlux to publish/receive Server Sent Event. On publish side, the code is as below. Then, in a test case, I loop 10 times to call "publish" to send out messages.

```
public ValidationEventPublisher() {
    this.processor = DirectProcessor.<ExternalEvent>create().serialize();
    this.sink = processor.sink();
}

@RequestMapping(value = "/ssp/common/v1/eventstream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<ExternalEvent>> sse() {
    return processor.map(e -> ServerSentEvent.builder(e).id(String.valueOf(msgId.incrementAndGet())).build())
            .onBackpressureBuffer();
}

public void publish(ValidationEvent validationEvent)
{
    log.info("Published Validation event...");
    sink.next(validationEvent);
}

```

On receiving side, the code is:

```
final Flux<Object> stream = WebClient.builder()
            .baseUrl(sspUrl + "/ssp/common/v1/eventstream")
            .defaultHeader(HttpHeaders.AUTHORIZATION, "Basic " + Base64Utils.encodeToString((username + ":" + password).getBytes(UTF_8)))
            .build()
            .get()
            .retrieve()
            .bodyToFlux(ServerSentEvent.class)
            .filter(e -> e.id() != null)
            .flatMap(e -> Mono.just(e.data()))
            .doOnError(e -> handleServerDown(e))
            .repeat();

    stream.subscribe(e -> processEvent(e));

```

So, the problem is that I can only receive the first message and after that I always got an exception.

```
- Published Validation event...
- Received validation event: {"messageId":"867b5ced-...}
- Published Validation event...
- Published Validation event...
2018-09-05 11:07:43,596 ERROR [reactor-http-nio-6] [] [] 
[org.springframework.web.server.adapter.HttpWebHandlerAdapter:213] - 
Unhandled failure: An established connection was aborted by the software in 
your host machine, response already set (status=null)
2018-09-05 11:07:43,597  WARN [reactor-http-nio-6] [] [] 
[org.springframework.http.server.reactive.ReactorHttpHandlerAdapter:76] - 
Handling completed with error: An established connection was aborted by the 
software in your host machine
...

```

I spent quite sometime doing some research but cannot find relevant information.
Note that I am running both sides in one test case, which means that they are in the same process and JVM. I don't think it matters but just in case.

I found the reason is a "cancellation" is sent out. But why? For example, if I add "doOnCancel" on my receiver side, I can see it is called.