docs: public static <I extends StreamSourceChannel, O extends StreamSinkChannel> void initiateTransfer(long count,
          final I source, final O sink, final ChannelListener<? super I> sourceListener,
          final ChannelListener<? super O> sinkListener, final ChannelExceptionHandler<? super I> readExceptionHandler,
          final ChannelExceptionHandler<? super O> writeExceptionHandler, Pool<ByteBuffer> pool) {
          if (pool == null) {
              throw msg.nullParameter("pool");
          }
          final Pooled<ByteBuffer> allocated = pool.allocate();
          boolean free = true;
          try {
              final ByteBuffer buffer = allocated.getResource();
              long transferred;
              for (; ; ) {
                  try {
                      transferred = source.transferTo(count, buffer, sink);
                  } catch (IOException e) {
                      ChannelListeners.invokeChannelExceptionHandler(source, readExceptionHandler, e);
                      return;
                  }
                  if (transferred == 0 && !buffer.hasRemaining()) {
                      break;
                  }
                  if (transferred == -1) {
                      if (count == Long.MAX_VALUE) {
                          Channels.setReadListener(source, sourceListener);
                          if (sourceListener == null) {
                              source.suspendReads();
                          } else {
                              source.wakeupReads();
                          }
  
                          Channels.setWriteListener(sink, sinkListener);
                          if (sinkListener == null) {
                              sink.suspendWrites();
                          } else {
                              sink.wakeupWrites();
                          }
                      } else {
                          source.suspendReads();
                          sink.suspendWrites();
                          ChannelListeners.invokeChannelExceptionHandler(source, readExceptionHandler, new EOFException());
                      }
                      return;
                  }
                  if (count != Long.MAX_VALUE) {
                      count -= transferred;
                  }
                  while (buffer.hasRemaining()) {
                      final int res;
                      try {
                          res = sink.write(buffer);
                      } catch (IOException e) {
                          ChannelListeners.invokeChannelExceptionHandler(sink, writeExceptionHandler, e);
                          return;
                      }
                      if (res == 0) {
                          // write first listener
                          final TransferListener<I, O> listener =
                              new TransferListener<I, O>(count, allocated, source, sink, sourceListener, sinkListener,
                                  writeExceptionHandler, readExceptionHandler, 1);
                          source.suspendReads();
                          source.getReadSetter().set(listener);
                          // flush the write channel
                          sink.getWriteSetter()
                              .set(ChannelListeners.flushingChannelListener(listener,
                                  new ChannelExceptionHandler<Channel>() {
                                      @Override
                                      public void handleException(Channel channel, IOException exception) {
                                          listener.writeFailed(exception);
                                      }
                                  }));
                          sink.resumeWrites();
                          free = false;
                          return;
                      } else if (count != Long.MAX_VALUE) {
                          count -= res;
                      }
                  }
                  if (count == 0) {
                      //we are done
                      Channels.setReadListener(source, sourceListener);
                      if (sourceListener == null) {
                          source.suspendReads();
                      } else {
                          source.wakeupReads();
                      }
  
                      Channels.setWriteListener(sink, sinkListener);
                      if (sinkListener == null) {
                          sink.suspendWrites();
                      } else {
                          sink.wakeupWrites();
                      }
                      return;
                  }
              }
  
              // flush the write channel
              try {
                  // TODO: think about checking to see if a flush has already been issued sink.isReadyForFlush()
                  if (!sink.flush()) {
                      final TransferListener<I, O> listener =
                          new TransferListener<I, O>(count, allocated, source, sink, sourceListener, sinkListener,
                              writeExceptionHandler, readExceptionHandler, 1);
                      source.suspendReads();
                      source.getReadSetter().set(listener);
                      // flush the write channel
                      sink.getWriteSetter()
                          .set(ChannelListeners.flushingChannelListener(listener, new ChannelExceptionHandler<Channel>() {
                              @Override
                              public void handleException(Channel channel, IOException exception) {
                                  listener.writeFailed(exception);
                              }
                          }));
                      sink.resumeWrites();
                      free = false;
                      return;
                  }
              } catch (IOException e) {
                  ChannelListeners.invokeChannelExceptionHandler(sink, writeExceptionHandler, e);
                  return;
              }
  
              // read first listener
              final TransferListener<I, O> listener =
                  new TransferListener<I, O>(count, allocated, source, sink, sourceListener, sinkListener,
                      writeExceptionHandler, readExceptionHandler, 0);
              sink.suspendWrites();
              sink.getWriteSetter().set(listener);
              source.getReadSetter().set(listener);
              source.resumeReads();
              free = false;
              return;
          } finally {
              if (free) allocated.free();
          }
      }