Bugfix/8934 handle estimation sse cancel

[NekoLyn]

Main changes:

1. Keep-alive from inside the read loop. DAS's own heartbeats drive it, so the timer thread is gone. A write to a closed portal socket throws there, which unwinds the read and closes the DAS connection for free.
2. A dedicated WebClient for the streamed DAS endpoints. A RestTemplate cannot cancel mid-response, and the cancel is what closes the socket DAS is watching. The connector is a fork of Spring's JdkClientHttpConnector without the cache(0) on the response body, which swallows the cancel.
3. A dedicated thread pool for SSE work instead of ForkJoinPool.commonPool(), which runs one thread on a 2-vCPU container, so a single slow estimate blocked every other SSE request.

[utas-raymondng]

Hmm... I think there is a SSE client that you can use directly?

https://www.baeldung.com/spring-server-sent-events

[NekoLyn]

Hmm... I think there is a SSE client that you can use directly?

https://www.baeldung.com/spring-server-sent-events

Good call, and I did try it — WebClient.bodyToFlux(ServerSentEvent<String>) is the one Spring client that decodes text/event-stream, and it deletes ~90 lines of hand-rolled frame parsing.

However it doesn't work here:
With the default JdkClientHttpConnector, cancelling the stream doesn't close the connection. Spring's JdkClientHttpResponse#adaptBody ends in .cache(0), which never passes the cancel up to the JDK, so the socket to DAS stays open and DAS keeps computing an estimate nobody will read. That's the exact bug this PR want to fix.

Measured by AI against a real socket — consume two frames, then cancel:

Transport	Upstream sees disconnect
What's on this branch today (RestTemplate, closing the InputStream)	yes
WebClient + JdkClientHttpConnector	never
WebClient + ReactorClientHttpConnector	yes, 762ms

WebClient + ReactorClientHttpConnector would work - the connector that does cancel properly needs reactor-netty-http: 11 new artifacts (reactor-netty-core、reactor-netty-http、netty-codec-dns、netty-resolver-dns、nett
y-resolver-dns-classes-macos、netty-resolver-dns-native-macos、netty-codec-socks... ) plus a second HTTP stack with its own event loop, to serve one endpoint.
It works, so I'm happy to go this way if you'd rather — it just felt like a lot to pay for deleting a parser.

[utas-raymondng]

Looks like you can write a custom connector to overwrite the annoying behavior, then you do not need the reactor

mport java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Flow;
import java.util.function.Function;

import reactor.adapter.JdkFlowAdapter;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatusCode;
import org.springframework.http.client.reactive.AbstractClientHttpRequest;
import org.springframework.http.client.reactive.AbstractClientHttpResponse;
import org.springframework.http.client.reactive.ClientHttpConnector;
import org.springframework.http.client.reactive.ClientHttpRequest;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.util.CollectionUtils;
import org.springframework.util.LinkedCaseInsensitiveMap;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.reactivestreams.Publisher;

/**
 * JDK HttpClient connector that forwards subscriber cancel to the TCP connection.
 * Unlike {@code JdkClientHttpConnector}, the body Flux is not {@code .cache(0)}.
 */
public final class CancelPropagatingJdkConnector implements ClientHttpConnector {

    private final HttpClient httpClient;
    private final DataBufferFactory buffers = DefaultDataBufferFactory.sharedInstance;
    private final Duration timeout;

    public CancelPropagatingJdkConnector(HttpClient httpClient, Duration timeout) {
        this.httpClient = httpClient;
        this.timeout = timeout;
    }

    @Override
    public Mono<ClientHttpResponse> connect(
            HttpMethod method,
            URI uri,
            Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {

        Request request = new Request(method, uri, buffers, timeout);

        return requestCallback.apply(request).then(Mono.defer(() -> {
            CompletableFuture<HttpResponse<Flow.Publisher<List<ByteBuffer>>>> future =
                    httpClient.sendAsync(request.build(), HttpResponse.BodyHandlers.ofPublisher());

            return Mono.fromCompletionStage(future)
                    .doOnCancel(() -> future.cancel(true))
                    .map(Response::new);
        }));
    }

    private static final class Request extends AbstractClientHttpRequest {
        private final HttpMethod method;
        private final URI uri;
        private final DataBufferFactory buffers;
        private final HttpRequest.Builder builder;

        Request(HttpMethod method, URI uri, DataBufferFactory buffers, Duration timeout) {
            this.method = method;
            this.uri = uri;
            this.buffers = buffers;
            this.builder = HttpRequest.newBuilder(uri).timeout(timeout);
        }

        @Override public HttpMethod getMethod() { return method; }
        @Override public URI getURI() { return uri; }
        @Override public DataBufferFactory bufferFactory() { return buffers; }
        @Override public <T> T getNativeRequest() { return (T) builder.build(); }

        HttpRequest build() { return builder.build(); }

        @Override
        public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
            // SSE is GET; no request body. Commit headers only.
            return doCommit(() -> {
                builder.method(method.name(), HttpRequest.BodyPublishers.noBody());
                return Mono.empty();
            });
        }

        @Override
        public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
            return writeWith(Flux.from(body).flatMap(Function.identity()));
        }

        @Override
        public Mono<Void> setComplete() {
            return doCommit(() -> {
                builder.method(method.name(), HttpRequest.BodyPublishers.noBody());
                return Mono.empty();
            });
        }

        @Override
        protected void applyHeaders() {
            getHeaders().forEach((name, values) -> {
                if (HttpHeaders.CONTENT_LENGTH.equalsIgnoreCase(name)) {
                    return;
                }
                values.forEach(v -> builder.header(name, v));
            });
            if (!getHeaders().containsKey(HttpHeaders.ACCEPT)) {
                builder.header(HttpHeaders.ACCEPT, "*/*");
            }
        }

        @Override
        protected void applyCookies() {
            // not needed for DAS
        }
    }

    private static final class Response extends AbstractClientHttpResponse {
        Response(HttpResponse<Flow.Publisher<List<ByteBuffer>>> nativeResponse) {
            super(
                    HttpStatusCode.valueOf(nativeResponse.statusCode()),
                    headersOf(nativeResponse),
                    new LinkedMultiValueMap<>(),
                    bodyOf(nativeResponse));
        }

        private static HttpHeaders headersOf(HttpResponse<?> response) {
            Map<String, List<String>> map =
                    new LinkedCaseInsensitiveMap<>(response.headers().map().size(), Locale.ROOT);
            MultiValueMap<String, String> headers = CollectionUtils.toMultiValueMap(map);
            headers.putAll(response.headers().map());
            return HttpHeaders.readOnlyHttpHeaders(headers);
        }

        private static Flux<DataBuffer> bodyOf(
                HttpResponse<Flow.Publisher<List<ByteBuffer>>> response) {

            Flow.Publisher<List<ByteBuffer>> body = response.body();
            if (body == null) {
                return Flux.empty();
            }
            DataBufferFactory factory = DefaultDataBufferFactory.sharedInstance;
            return JdkFlowAdapter.flowPublisherToFlux(body)
                    .flatMapIterable(Function.identity())
                    .map(factory::wrap)
                    .doOnDiscard(DataBuffer.class, DataBufferUtils::release);
                    // no .cache(0) — cancel reaches Flow.Subscription.cancel()
        }
    }
}

HttpClient jdk = HttpClient.newBuilder()
        .connectTimeout(Duration.ofSeconds(20))
        .build();

WebClient dasSse = WebClient.builder()
        .clientConnector(new CancelPropagatingJdkConnector(jdk, Duration.ofMinutes(10)))
        .baseUrl(dasBaseUrl)
        .build();

Flux<ServerSentEvent<String>> events = dasSse.get()
        .uri(estimatePath)
        .accept(MediaType.TEXT_EVENT_STREAM)
        .retrieve()
        .bodyToFlux(new ParameterizedTypeReference<ServerSentEvent<String>>() {});

[utas-raymondng]

LGTM