class NioReactor {

  private final Selector selector;
  private final Dispatcher dispatcher;
  /**
   * All the work of altering the SelectionKey operations and Selector operations are performed in
   * the context of main event loop of reactor. So when any channel needs to change its readability
   * or writability, a new command is added in the command queue and then the event loop picks up
   * the command and executes it in next iteration.
   */
  private final Queue<Runnable> pendingCommands = new ConcurrentLinkedQueue<>();
  private final ExecutorService reactorMain = Executors.newSingleThreadExecutor();

  /**
   * Creates a reactor which will use provided {@code dispatcher} to dispatch events. The
   * application can provide various implementations of dispatcher which suits its needs.
   *
   * @param dispatcher a non-null dispatcher used to dispatch events on registered channels.
   * @throws IOException if any I/O error occurs.
   */
  public NioReactor(Dispatcher dispatcher) throws IOException {
    this.dispatcher = dispatcher;
    this.selector = Selector.open();
  }

  /**
   * Starts the reactor event loop in a new thread.
   */
  public void start() {<FILL_FUNCTION_BODY>}

  /**
   * Stops the reactor and related resources such as dispatcher.
   *
   * @throws InterruptedException if interrupted while stopping the reactor.
   * @throws IOException          if any I/O error occurs.
   */
  public void stop() throws InterruptedException, IOException {
    reactorMain.shutdown();
    selector.wakeup();
    if (!reactorMain.awaitTermination(4, TimeUnit.SECONDS)) {
      reactorMain.shutdownNow();
    }
    selector.close();
    LOGGER.info("Reactor stopped");
  }

  /**
   * Registers a new channel (handle) with this reactor. Reactor will start waiting for events on
   * this channel and notify of any events. While registering the channel the reactor uses {@link
   * AbstractNioChannel#getInterestedOps()} to know about the interested operation of this channel.
   *
   * @param channel a new channel on which reactor will wait for events. The channel must be bound
   *                prior to being registered.
   * @return this
   * @throws IOException if any I/O error occurs.
   */
  public NioReactor registerChannel(AbstractNioChannel channel) throws IOException {
    var key = channel.getJavaChannel().register(selector, channel.getInterestedOps());
    key.attach(channel);
    channel.setReactor(this);
    return this;
  }

  private void eventLoop() throws IOException {
    // honor interrupt request
    while (!Thread.interrupted()) {
      // honor any pending commands first
      processPendingCommands();

      /*
       * Synchronous event de-multiplexing happens here, this is blocking call which returns when it
       * is possible to initiate non-blocking operation on any of the registered channels.
       */
      selector.select();

      /*
       * Represents the events that have occurred on registered handles.
       */
      var keys = selector.selectedKeys();
      var iterator = keys.iterator();

      while (iterator.hasNext()) {
        var key = iterator.next();
        if (!key.isValid()) {
          iterator.remove();
          continue;
        }
        processKey(key);
      }
      keys.clear();
    }
  }

  private void processPendingCommands() {
    var iterator = pendingCommands.iterator();
    while (iterator.hasNext()) {
      var command = iterator.next();
      command.run();
      iterator.remove();
    }
  }

  /*
   * Initiation dispatcher logic, it checks the type of event and notifier application specific
   * event handler to handle the event.
   */
  private void processKey(SelectionKey key) throws IOException {
    if (key.isAcceptable()) {
      onChannelAcceptable(key);
    } else if (key.isReadable()) {
      onChannelReadable(key);
    } else if (key.isWritable()) {
      onChannelWritable(key);
    }
  }

  private static void onChannelWritable(SelectionKey key) throws IOException {
    var channel = (AbstractNioChannel) key.attachment();
    channel.flush(key);
  }

  private void onChannelReadable(SelectionKey key) {
    try {
      // reads the incoming data in context of reactor main loop. Can this be improved?
      var readObject = ((AbstractNioChannel) key.attachment()).read(key);
      dispatchReadEvent(key, readObject);
    } catch (IOException e) {
      try {
        key.channel().close();
      } catch (IOException e1) {
        LOGGER.error("error closing channel", e1);
      }
    }
  }

  /*
   * Uses the application provided dispatcher to dispatch events to application handler.
   */
  private void dispatchReadEvent(SelectionKey key, Object readObject) {
    dispatcher.onChannelReadEvent((AbstractNioChannel) key.attachment(), readObject, key);
  }

  private void onChannelAcceptable(SelectionKey key) throws IOException {
    var serverSocketChannel = (ServerSocketChannel) key.channel();
    var socketChannel = serverSocketChannel.accept();
    socketChannel.configureBlocking(false);
    var readKey = socketChannel.register(selector, SelectionKey.OP_READ);
    readKey.attach(key.attachment());
  }

  /**
   * Queues the change of operations request of a channel, which will change the interested
   * operations of the channel sometime in the future.
   *
   * <p>This is a non-blocking method and does not guarantee that the operations have changed when
   * this method returns.
   *
   * @param key           the key for which operations have to be changed.
   * @param interestedOps the new interest operations.
   */
  public void changeOps(SelectionKey key, int interestedOps) {
    pendingCommands.add(new ChangeKeyOpsCommand(key, interestedOps));
    selector.wakeup();
  }

  /**
   * A command that changes the interested operations of the key provided.
   */
  static class ChangeKeyOpsCommand implements Runnable {
    private final SelectionKey key;
    private final int interestedOps;

    public ChangeKeyOpsCommand(SelectionKey key, int interestedOps) {
      this.key = key;
      this.interestedOps = interestedOps;
    }

    public void run() {
      key.interestOps(interestedOps);
    }

    @Override
    public String toString() {
      return "Change of ops to: " + interestedOps;
    }
  }
}

    reactorMain.execute(() -> {
      try {
        LOGGER.info("Reactor started, waiting for events...");
        eventLoop();
      } catch (IOException e) {
        LOGGER.error("exception in event loop", e);
      }
    });
  
<no_super_class>