class InputStreamPumper {

  private static final int DEFAULT_BUFFER_SIZE = 8192;

  private InputStreamPumper() {
  }

  private static final Logger LOGGER = LoggerFactory.getLogger(InputStreamPumper.class);

  public interface Writable {

    void write(byte[] b, int off, int len) throws IOException;

  }

  /**
   * Relies on {@link InputStream#available()} and a Thread sleep to ensure that the reads are interruptible.
   */
  public static InputStream asInterruptible(InputStream is) {<FILL_FUNCTION_BODY>}

  /**
   * See InputStream.transferTo(java.io.OutputStream) in Java 9 or later
   */
  public static void transferTo(InputStream in, Writable out) throws IOException {
    byte[] buffer = new byte[DEFAULT_BUFFER_SIZE];
    int length;
    while ((length = in.read(buffer, 0, buffer.length)) != -1) {
      out.write(buffer, 0, length);
    }
  }

  /**
   * Pumps the given {@link InputStream} into the {@link Writable} target via a task started in the given {@link Executor}.
   * <br>
   * The input is not closed by this call.
   * <br>
   * If the {@link InputStream} is not interruptible, such as System.in, use {@link #asInterruptible(InputStream)} to decorate
   * the stream for this call.
   */
  public static CompletableFuture<?> pump(InputStream in, Writable out, Executor executor) {
    return CompletableFuture.runAsync(() -> {
      try {
        InputStreamPumper.transferTo(in, out);
      } catch (InterruptedIOException e) {
        LOGGER.debug("Interrupted while pumping stream.", e);
      } catch (Exception e) {
        if (!Thread.currentThread().isInterrupted()) {
          LOGGER.error("Error while pumping stream.", e);
        } else {
          LOGGER.debug("Interrupted while pumping stream.");
        }
      }
    }, executor);
  }

  static class WritableOutputStream extends OutputStream {

    Writable writer;

    WritableOutputStream(Writable writer) {
      this.writer = writer;
    }

    @Override
    public void write(byte[] b, int off, int len) throws IOException {
      writer.write(b, off, len);
    }

    @Override
    public void write(int b) throws IOException {
      throw new UnsupportedOperationException();
    }

  }

  public static OutputStream writableOutputStream(Writable writer, Integer bufferSize) {
    return new BufferedOutputStream(new WritableOutputStream(writer), Utils.getNonNullOrElse(bufferSize, DEFAULT_BUFFER_SIZE));
  }

}

    return new InputStream() {

      @Override
      public int read() {
        throw new UnsupportedOperationException();
      }

      @Override
      public int read(byte[] b, int off, int len) throws IOException {
        while (!Thread.currentThread().isInterrupted()) {
          if (is.available() > 0) {
            return is.read(b, off, len);
          }
          try {
            // The default sleep interval of 50 milliseconds came from the previous
            // incarnation of this code.  If this needs tweaked, it can be changed to a parameter.
            Thread.sleep(50);
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IOException(e);
          }
        }
        throw new IOException();
      }
    };
  
<no_super_class>