How to close client sse connection in spring boot application

I have a spring boot application and I have to connect to some outside service using SSE. WebClient establishes the connection and then I'm using Flux for reading responses. Everything works ok, but the problem is that the connection stays open, because the process is not designed to reach the finish point every time in that 3rd party service. I would like to close the connection manually as a client since I know when this connection should finish. How can I do that?

Establishing connection:

private Flux<ServerSentEvent<String>> connect(String accessToken) {
    TcpClient timeoutClient = createTimeoutClient();
    ReactorClientHttpConnector reactorClientHttpConnector = new ReactorClientHttpConnector(HttpClient.from(timeoutClient));
    String url = npzServerBaseUrl+uniqueCodePath;
    WebClient client = WebClient
            .builder()
            .clientConnector(reactorClientHttpConnector)
            .defaultHeader(HttpHeaders.AUTHORIZATION, Naming.TOKEN_PREFIX + accessToken)
            .baseUrl(url)
            .build();

    ParameterizedTypeReference<ServerSentEvent<String>> type
            = new ParameterizedTypeReference<ServerSentEvent<String>>() {};
    return client.get()
            .retrieve()
            .onStatus(HttpStatus::is4xxClientError, clientResponse -> {
                String msg = "Error from server: "+clientResponse.statusCode().toString();
                        //invalidate access token
                        if (clientResponse.statusCode().value()==401) {
                            //remove invalid token and connect again
                            loginContext.invalidToken(accessToken);
                            return Mono.error(new InvalidNpzToken(msg));
                        }
                        return Mono.error(new IllegalStateException(msg));
                    }
            )
            .onStatus(HttpStatus::is5xxServerError, clientResponse ->
                    Mono.error(new IllegalStateException("Error from server: "+clientResponse.statusCode().toString()))
            )
            .bodyToFlux(type);
}

private TcpClient createTimeoutClient() {
    return TcpClient.create()
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, SECONDS*1000)
            .option(EpollChannelOption.TCP_USER_TIMEOUT, SECONDS*1000)
            .doOnConnected(
                    c -> c.addHandlerLast(new ReadTimeoutHandler(SECONDS))
                            .addHandlerLast(new WriteTimeoutHandler(SECONDS)));
}

Handling content:

Flux<ServerSentEvent<String>> eventStream = connect(accessToken);

    eventStream.subscribe(
            content -> {
                log.info("Time: {} - event: name[{}], id [{}], content[{}] ",
                    LocalTime.now(), content.event(), content.id(), content.data());
                if ("uuid".equals(content.event().trim())) {
                    listener.receivedUniqueCode(content.data().trim());
                } else if ("code".equals(content.event().trim())) {
                    listener.receivedCode(content.data().trim());
                }
            },
            (Throwable error) -> {
                if (error instanceof InvalidToken) {
                    log.error("Error receiving SSE", error);
                    //let's retry connection as token has expired
                    getCode(request, listener);
                }
            },
            () -> log.info("Connection closed!"));

What I expect is that I can call connection.close() or something like that and connection will be closed.

Thanks

Some more information if needed.

Imports:

import io.netty.channel.ChannelOption;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;
import reactor.netty.tcp.TcpClient;

Spring boot:

<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.1.RELEASE</version>