PipeSanityCheckTest.java

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.RepeatedTest;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Pipe;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
class PipeSanityCheckTest {

  @RepeatedTest(1000)
  void test() throws IOException, InterruptedException {
    var pipe = Pipe.open();
    var sink = pipe.sink();
    var source = pipe.source();
    ReentrantLock sinkCloseLock = new ReentrantLock();
    var writeThread = new Thread(() -> {
      while (true) {
        try {
          sinkCloseLock.lock();
          if (!sink.isOpen()) {
            return;
          }
          log.info("before write");
          int bytesWritten = sink.write(ByteBuffer.wrap("test".getBytes()));
          log.info("after write {}", bytesWritten);
        } catch (IOException e) {
          log.error("error", e);
          throw new RuntimeException(e);
        } finally {
          sinkCloseLock.unlock();
        }
      }
    }, "Write Thread");
    var readThread = new Thread(() -> {
      int bytesRead;
      do {
        try {
          log.info("before read");
          bytesRead = source.read(ByteBuffer.allocate(1000));
          log.info("after read {}", bytesRead);
        } catch (IOException e) {
          log.error("error", e);
          throw new RuntimeException(e);
        }
      } while (bytesRead != -1);
    }, "Read Thread");

    writeThread.start();
    readThread.start();

    long sleepTime = (long) (Math.random() * 100);
    log.info("sleep time {}", sleepTime);
    Thread.sleep(sleepTime);
    log.info("before close");
    try {
      sinkCloseLock.lock();
      sink.close();
    } finally {
      sinkCloseLock.unlock();
    }
    log.info("after close");

    writeThread.join();
    readThread.join();

    source.close();
  }

}