class WatchConnectionManager<T extends HasMetadata, L extends KubernetesResourceList<T>>
    extends AbstractWatchManager<T> {

  private final long connectTimeoutMillis;
  protected WatcherWebSocketListener<T> listener;
  private volatile CompletableFuture<WebSocket> websocketFuture;

  volatile boolean ready;

  public WatchConnectionManager(final HttpClient client, final BaseOperation<T, L, ?> baseOperation,
      final ListOptions listOptions, final Watcher<T> watcher, final int reconnectInterval, final int reconnectLimit,
      long websocketTimeout) throws MalformedURLException {
    super(watcher, baseOperation, listOptions, reconnectLimit, reconnectInterval, client);
    this.connectTimeoutMillis = websocketTimeout;
  }

  @Override
  protected void closeCurrentRequest() {
    Optional.ofNullable(this.websocketFuture).ifPresent(
        theFuture -> theFuture.whenComplete(
            (w, t) -> Optional.ofNullable(w).ifPresent(ws -> ws.sendClose(1000, null))));
  }

  public CompletableFuture<WebSocket> getWebsocketFuture() {
    return websocketFuture;
  }

  @Override
  protected void start(URL url, Map<String, String> headers, WatchRequestState state) {<FILL_FUNCTION_BODY>}

}

    this.listener = new WatcherWebSocketListener<>(this, state);
    Builder builder = client.newWebSocketBuilder();
    headers.forEach(builder::header);
    builder.uri(URI.create(url.toString())).connectTimeout(connectTimeoutMillis, TimeUnit.MILLISECONDS);

    this.websocketFuture = builder.buildAsync(this.listener).handle((w, t) -> {
      if (t != null) {
        try {
          if (t instanceof WebSocketHandshakeException) {
            WebSocketHandshakeException wshe = (WebSocketHandshakeException) t;
            HttpResponse<?> response = wshe.getResponse();
            final int code = response.code();
            // We do not expect a 200 in response to the websocket connection. If it occurs, we throw
            // an exception and try the watch via a persistent HTTP Get.
            // Newer Kubernetes might also return 503 Service Unavailable in case WebSockets are not supported
            Status status = OperationSupport.createStatus(response, this.baseOperation.getKubernetesSerialization());
            t = OperationSupport.requestFailure(client.newHttpRequestBuilder().url(url).build(),
                status, "Received " + code + " on websocket");
          }
          throw KubernetesClientException.launderThrowable(t);
        } finally {
          if (ready) {
            // if we're ready, then invoke the reconnect logic
            watchEnded(t, state);
          }
        }
      }
      this.ready = true;
      return w;
    });
  
<methods>public void close() ,public synchronized void closeRequest() ,public void setWatchEndCheckMs(int) <variables>private static final int INFO_LOG_CONNECTION_ERRORS,protected BaseOperation<T,?,?> baseOperation,protected final non-sealed io.fabric8.kubernetes.client.http.HttpClient client,private final Map<Class<?>,java.lang.Integer> endErrors,final non-sealed java.util.concurrent.atomic.AtomicBoolean forceClosed,volatile io.fabric8.kubernetes.client.dsl.internal.AbstractWatchManager.WatchRequestState latestRequestState,private final non-sealed ListOptions listOptions,private static final Logger logger,private final non-sealed boolean receiveBookmarks,private Future<?> reconnectAttempt,private final non-sealed int reconnectLimit,private final non-sealed java.net.URL requestUrl,final non-sealed AtomicReference<java.lang.String> resourceVersion,private java.util.concurrent.atomic.AtomicInteger retryAfterSeconds,private final non-sealed io.fabric8.kubernetes.client.utils.ExponentialBackoffIntervalCalculator retryIntervalCalculator,private int watchEndCheckMs,final non-sealed Watcher<T> watcher