class VertxWebSocket implements WebSocket {

  private static final Logger LOG = LoggerFactory.getLogger(VertxWebSocket.class);

  private final io.vertx.core.http.WebSocket ws;
  private final AtomicInteger pending = new AtomicInteger();
  private final Listener listener;

  VertxWebSocket(io.vertx.core.http.WebSocket ws, Listener listener) {
    this.ws = ws;
    this.listener = listener;
  }

  void init() {<FILL_FUNCTION_BODY>}

  @Override
  public boolean send(ByteBuffer buffer) {
    Buffer vertxBuffer = Buffer.buffer(Unpooled.copiedBuffer(buffer));
    int len = vertxBuffer.length();
    pending.addAndGet(len);
    Future<Void> res = ws.writeBinaryMessage(vertxBuffer);
    if (res.isComplete()) {
      pending.addAndGet(-len);
      return res.succeeded();
    }
    res.onComplete(result -> {
      if (result.cause() != null) {
        LOG.error("Queued write did not succeed", result.cause());
      }
      pending.addAndGet(-len);
    });
    return true;
  }

  @Override
  public synchronized boolean sendClose(int code, String reason) {
    if (ws.isClosed()) {
      return false;
    }
    Future<Void> res = ws.close((short) code, reason);
    res.onComplete(result -> {
      ws.fetch(1);
      if (result.cause() != null) {
        LOG.error("Queued close did not succeed", result.cause());
      }
    });
    return true;
  }

  @Override
  public long queueSize() {
    return pending.get();
  }

  @Override
  public void request() {
    ws.fetch(1);
  }
}

    ws.binaryMessageHandler(msg -> {
      ws.pause();
      listener.onMessage(this, msg.getByteBuf().nioBuffer());
    });
    ws.textMessageHandler(msg -> {
      ws.pause();
      listener.onMessage(this, msg);
    });
    // use end, not close, because close is processed immediately vs. end is in frame order
    ws.endHandler(v -> listener.onClose(this, ws.closeStatusCode(), ws.closeReason()));
    ws.exceptionHandler(err -> {
      try {
        if (err instanceof CorruptedWebSocketFrameException) {
          err = new ProtocolException().initCause(err);
        } else if (err instanceof HttpClosedException) {
          err = new IOException(err);
        }
        listener.onError(this, err);
      } finally {
        // onError should be terminal
        if (!ws.isClosed()) {
          ws.close();
        }
      }
    });
    listener.onOpen(this);
  
<no_super_class>