class WatchEventsListener extends WebSocketListener {
  private final AtomicReference<WebSocket> webSocketRef = new AtomicReference<>();
  private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
  private final AttributeSet attributeSet;
  private final Context context;
  private final Set<WatchEventsListener> watchEventListenerList;
  private final Logger logger;
  private final Consumer<WatchEventsListener> onOpenAction;

  public WatchEventsListener(Context context, AttributeSet attributeSet, final Set<WatchEventsListener> watchEventListenerList,
      Logger logger, Consumer<WatchEventsListener> onOpenAction) {
    this.logger = logger;
    this.context = context;
    this.attributeSet = attributeSet;
    this.watchEventListenerList = watchEventListenerList;
    this.onOpenAction = onOpenAction;
  }

  public boolean attributeMatches(AttributeSet set) {
    return set.matches(attributeSet);
  }

  @Override
  public void onOpen(WebSocket webSocket, Response response) {
    webSocketRef.set(webSocket);
    onOpenAction.accept(this);
  }

  @Override
  public void onClosing(WebSocket webSocket, int code, String reason) {
    webSocketRef.get().close(code, reason);
  }

  @Override
  public void onClosed(WebSocket webSocket, int code, String reason) {
    if (webSocketRef.get() != null) {
      webSocketRef.get().close(code, reason);
    }
    executor.shutdown();
    try {
      executor.awaitTermination(1, TimeUnit.MINUTES);
    } catch (InterruptedException e) {
      logger.debug("Interrupted waiting for the executor service to shutdown: {}", e.getMessage());
      Thread.currentThread().interrupt();
    }
    watchEventListenerList.remove(this);
  }

  @Override
  public void onFailure(WebSocket webSocket, Throwable t, Response response) {
    webSocket.close(1000, t.getMessage());
    executor.shutdown();
    try {
      executor.awaitTermination(1, TimeUnit.MINUTES);
    } catch (InterruptedException e) {
      logger.debug("Interrupted waiting for the executor service to shutdown: {}", e.getMessage());
      Thread.currentThread().interrupt();
    }
    watchEventListenerList.remove(this);
  }

  public void sendWebSocketResponse(String object, Watcher.Action action) {
    WebSocketMessage message = toWebSocketMessage(context,
        new WatchEvent(Serialization.unmarshal(object, GenericKubernetesResource.class), action.name()));
    executor.schedule(() -> webSocketRef.get().send(message.getBody()), message.getDelay(), TimeUnit.SECONDS);
  }

  private static WebSocketMessage toWebSocketMessage(Context context, Object content) {
    return toWebSocketMessage(context, content, true);
  }

  private static WebSocketMessage toWebSocketMessage(Context context, Object content, Boolean toBeRemoved) {
    return toWebSocketMessage(context, 0L, content, toBeRemoved);
  }

  private static WebSocketMessage toWebSocketMessage(Context context, Long delay, Object content, Boolean toBeRemoved) {<FILL_FUNCTION_BODY>}
}

    if (content instanceof String) {
      return new WebSocketMessage(delay, (String) content, toBeRemoved);
    } else if (content instanceof WebSocketMessage) {
      return (WebSocketMessage) content;
    } else {
      try {
        return toWebSocketMessage(context, delay, context.getMapper().writeValueAsString(content), toBeRemoved);
      } catch (JsonProcessingException e) {
        throw new IllegalArgumentException(e);
      }
    }
  
<no_super_class>