class AbstractNioChannel {

  private final SelectableChannel channel;
  private final ChannelHandler handler;
  private final Map<SelectableChannel, Queue<Object>> channelToPendingWrites;
  private NioReactor reactor;

  /**
   * Creates a new channel.
   *
   * @param handler which will handle events occurring on this channel.
   * @param channel a NIO channel to be wrapped.
   */
  public AbstractNioChannel(ChannelHandler handler, SelectableChannel channel) {
    this.handler = handler;
    this.channel = channel;
    this.channelToPendingWrites = new ConcurrentHashMap<>();
  }

  /**
   * Injects the reactor in this channel.
   */
  void setReactor(NioReactor reactor) {
    this.reactor = reactor;
  }

  /**
   * Get channel.
   *
   * @return the wrapped NIO channel.
   */
  public SelectableChannel getJavaChannel() {
    return channel;
  }

  /**
   * The operation in which the channel is interested, this operation is provided to {@link
   * Selector}.
   *
   * @return interested operation.
   * @see SelectionKey
   */
  public abstract int getInterestedOps();

  /**
   * Binds the channel on provided port.
   *
   * @throws IOException if any I/O error occurs.
   */
  public abstract void bind() throws IOException;

  /**
   * Reads the data using the key and returns the read data. The underlying channel should be
   * fetched using {@link SelectionKey#channel()}.
   *
   * @param key the key on which read event occurred.
   * @return data read.
   * @throws IOException if any I/O error occurs.
   */
  public abstract Object read(SelectionKey key) throws IOException;

  /**
   * Get handler.
   *
   * @return the handler associated with this channel.
   */
  public ChannelHandler getHandler() {
    return handler;
  }

  /*
   * Called from the context of reactor thread when the key becomes writable. The channel writes the
   * whole pending block of data at once.
   */
  void flush(SelectionKey key) throws IOException {<FILL_FUNCTION_BODY>}

  /**
   * Writes the data to the channel.
   *
   * @param pendingWrite the data to be written on channel.
   * @param key          the key which is writable.
   * @throws IOException if any I/O error occurs.
   */
  protected abstract void doWrite(Object pendingWrite, SelectionKey key) throws IOException;

  /**
   * Queues the data for writing. The data is not guaranteed to be written on underlying channel
   * when this method returns. It will be written when the channel is flushed.
   *
   * <p>This method is used by the {@link ChannelHandler} to send reply back to the client. <br>
   * Example:
   *
   * <pre>
   * <code>
   * {@literal @}Override
   * public void handleChannelRead(AbstractNioChannel channel, Object readObj, SelectionKey key) {
   *   byte[] data = ((ByteBuffer)readObj).array();
   *   ByteBuffer buffer = ByteBuffer.wrap("Server reply".getBytes());
   *   channel.write(buffer, key);
   * }
   * </code>
   * </pre>
   *
   * @param data the data to be written on underlying channel.
   * @param key  the key which is writable.
   */
  public void write(Object data, SelectionKey key) {
    var pendingWrites = this.channelToPendingWrites.get(key.channel());
    if (pendingWrites == null) {
      synchronized (this.channelToPendingWrites) {
        pendingWrites = this.channelToPendingWrites.computeIfAbsent(key.channel(), k -> new ConcurrentLinkedQueue<>());
      }
    }
    pendingWrites.add(data);
    reactor.changeOps(key, SelectionKey.OP_WRITE);
  }
}

    var pendingWrites = channelToPendingWrites.get(key.channel());
    Object pendingWrite;
    while ((pendingWrite = pendingWrites.poll()) != null) {
      // ask the concrete channel to make sense of data and write it to java channel
      doWrite(pendingWrite, key);
    }
    // We don't have anything more to write so channel is interested in reading more data
    reactor.changeOps(key, SelectionKey.OP_READ);
  
<no_super_class>