PipedOutputStreamToChannel.java

try (
    final PipedOutputStream outputStream = new PipedOutputStream();
    final PipedInputStream inputStream = new PipedInputStream(outputStream);
    final ReadableByteChannel channel = Channels.newChannel(inputStream);
) {
    new Thread(() -> {
        try {
            for (int i = 0; i < 10; i++) {
                Thread.sleep(300);
                outputStream.write(("idx:" + i).getBytes());
            }
            outputStream.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }).start();

    final ByteBuffer buffer = ByteBuffer.allocate(5);
    while ((channel.read(buffer)) > -1) {
        buffer.flip();
        logger.info("read {}", new String(buffer.array()));
    }
}

PipedStream.java

final PipedOutputStream outputStream = new PipedOutputStream();
final PipedInputStream inputStream = new PipedInputStream(outputStream);

new Thread(() -> {
    try {
        for (int i = 0; i < 10; i++) {
                Thread.sleep(1_000);
                outputStream.write(i);
        }
        outputStream.close();
    } catch (Exception e) {
        e.printStackTrace();
    }
}).start();

for (int read; (read = inputStream.read()) > -1;) {
    logger.info("read {}", read);
}

StandardFlowFileCodec.java

public void encode(DataPacket dataPacket, OutputStream encodedOut) throws IOException {
  final CountDownLatch dataPacketIsSent = new CountDownLatch(1);
  final Thread sendingThread = new Thread(() -> {
      try {
          super.encode(dataPacket, encodedOut);
          dataPacketIsSent.countDown();
      } catch (Exception e) {
          logger.error("Failed to send packet.", e);
      }
  });
  sendingThread.start();

  try {
      if (!dataPacketIsSent.await(5, TimeUnit.SECONDS)) {
          throw new IOException(new TimeoutException("Sending dataPacket has been timeout."));
      }
  } catch (InterruptedException e) {
      throw new IOException(e);
  }
}