Java 类java.nio.channels.Pipe 实例源码

项目：dhus-core 文件：DownloadableProduct.java

@Override
public <T> T getImpl(Class<? extends T> cl)
{
   if (InputStream.class.isAssignableFrom(cl))
   {
      try
      {
         Pipe pipe = Pipe.open();
         DownloadTask dltask = new DownloadTask(pipe);
         Thread download_thread = new Thread(dltask, "Product Download");
         download_thread.start();

         InputStream is = Channels.newInputStream(pipe.source());
         return cl.cast(is);
      }
      catch (IOException ex)
      {
         LOGGER.error("could not create pipe", ex);
      }
   }
   return null;
}

项目：openjdk-jdk10 文件：Open.java

public static void main(String[] args) throws Exception {

        // Load necessary classes ahead of time
        DatagramChannel dc = DatagramChannel.open();
        Exception se = new SocketException();
        SelectorProvider sp = SelectorProvider.provider();
        Pipe p = sp.openPipe();
        ServerSocketChannel ssc = ServerSocketChannel.open();

        test1();
        test2();
        test3();
        test4();
    }

项目：HeliosStreams 文件：InvocationChannel.java

/**
 * {@inheritDoc}
 * @see io.netty.channel.ChannelOutboundInvoker#write(java.lang.Object)
 */
@Override
public ChannelFuture write(Object message) {
    if(message!=null) {
        if(message instanceof FileRegion) {
            try {
                Pipe pipe = Pipe.open();
                FileRegion fr = (FileRegion)message;

                long bytesToRead = fr.count();
                fr.transferTo(pipe.sink(), 0L);
                byte[] content = new byte[(int)bytesToRead];
                pipe.source().read(ByteBuffer.wrap(content));
                channelWrites.add(content);
            } catch (Exception ex) {
                log.error("Failed to read content from pipe", ex);
                channelWrites.add(ex);
            }
        } else {
            channelWrites.add(message);
        }
        log.info("Received Channel Write [{}]  type:[{}]", message, message.getClass().getName());
    }

    return null;
}

项目：HeliosStreams 文件：InvocationChannel.java

/**
 * {@inheritDoc}
 * @see org.jboss.netty.channel.Channel#write(java.lang.Object)
 */
@Override
public ChannelFuture write(Object message) {
    if(message!=null) {
        if(message instanceof FileRegion) {
            try {
                Pipe pipe = Pipe.open();
                FileRegion fr = (FileRegion)message;

                long bytesToRead = fr.getCount();
                fr.transferTo(pipe.sink(), 0L);
                byte[] content = new byte[(int)bytesToRead];
                pipe.source().read(ByteBuffer.wrap(content));
                channelWrites.add(content);
            } catch (Exception ex) {
                log.error("Failed to read content from pipe", ex);
                channelWrites.add(ex);
            }
        } else {
            channelWrites.add(message);
        }
        log.info("Received Channel Write [{}]  type:[{}]", message, message.getClass().getName());
    }

    return Channels.succeededFuture(this);
}

项目：beam 文件：PackageUtilTest.java

@Test
public void testPackageUploadWithFileSucceeds() throws Exception {
  Pipe pipe = Pipe.open();
  String contents = "This is a test!";
  File tmpFile = makeFileWithContents("file.txt", contents);
  when(mockGcsUtil.getObjects(anyListOf(GcsPath.class)))
      .thenReturn(ImmutableList.of(StorageObjectOrIOException.create(
          new FileNotFoundException("some/path"))));

  when(mockGcsUtil.create(any(GcsPath.class), anyString())).thenReturn(pipe.sink());

  List<DataflowPackage> targets =
      defaultPackageUtil.stageClasspathElements(
          ImmutableList.of(tmpFile.getAbsolutePath()), STAGING_PATH, createOptions);
  DataflowPackage target = Iterables.getOnlyElement(targets);

  verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class));
  verify(mockGcsUtil).create(any(GcsPath.class), anyString());
  verifyNoMoreInteractions(mockGcsUtil);

  assertThat(target.getName(), RegexMatcher.matches("file-" + HASH_PATTERN + ".txt"));
  assertThat(target.getLocation(), equalTo(STAGING_PATH + target.getName()));
  assertThat(new LineReader(Channels.newReader(pipe.source(), "UTF-8")).readLine(),
      equalTo(contents));
}

项目：beam 文件：PackageUtilTest.java

@Test
public void testStagingPreservesClasspath() throws Exception {
  File smallFile = makeFileWithContents("small.txt", "small");
  File largeFile = makeFileWithContents("large.txt", "large contents");
  when(mockGcsUtil.getObjects(anyListOf(GcsPath.class)))
      .thenReturn(ImmutableList.of(StorageObjectOrIOException.create(
          new FileNotFoundException("some/path"))));

  when(mockGcsUtil.create(any(GcsPath.class), anyString()))
      .thenAnswer(new Answer<SinkChannel>() {
        @Override
        public SinkChannel answer(InvocationOnMock invocation) throws Throwable {
          return Pipe.open().sink();
        }
      });

  List<DataflowPackage> targets =
      defaultPackageUtil.stageClasspathElements(
          ImmutableList.of(smallFile.getAbsolutePath(), largeFile.getAbsolutePath()),
          STAGING_PATH,
          createOptions);
  // Verify that the packages are returned small, then large, matching input order even though
  // the large file would be uploaded first.
  assertThat(targets.get(0).getName(), startsWith("small"));
  assertThat(targets.get(1).getName(), startsWith("large"));
}

项目：beam 文件：PackageUtilTest.java

@Test
public void testPackageUploadWithEmptyDirectorySucceeds() throws Exception {
  Pipe pipe = Pipe.open();
  File tmpDirectory = tmpFolder.newFolder("folder");

  when(mockGcsUtil.getObjects(anyListOf(GcsPath.class)))
      .thenReturn(ImmutableList.of(StorageObjectOrIOException.create(
          new FileNotFoundException("some/path"))));
  when(mockGcsUtil.create(any(GcsPath.class), anyString())).thenReturn(pipe.sink());

  List<DataflowPackage> targets =
      defaultPackageUtil.stageClasspathElements(
          ImmutableList.of(tmpDirectory.getAbsolutePath()), STAGING_PATH, createOptions);
  DataflowPackage target = Iterables.getOnlyElement(targets);

  verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class));
  verify(mockGcsUtil).create(any(GcsPath.class), anyString());
  verifyNoMoreInteractions(mockGcsUtil);

  assertThat(target.getName(), RegexMatcher.matches("folder-" + HASH_PATTERN + ".jar"));
  assertThat(target.getLocation(), equalTo(STAGING_PATH + target.getName()));
  assertNull(new ZipInputStream(Channels.newInputStream(pipe.source())).getNextEntry());
}

项目：beam 文件：PackageUtilTest.java

@Test
public void testPackageUploadEventuallySucceeds() throws Exception {
  Pipe pipe = Pipe.open();
  File tmpFile = makeFileWithContents("file.txt", "This is a test!");
  when(mockGcsUtil.getObjects(anyListOf(GcsPath.class)))
      .thenReturn(ImmutableList.of(StorageObjectOrIOException.create(
          new FileNotFoundException("some/path"))));
  when(mockGcsUtil.create(any(GcsPath.class), anyString()))
      .thenThrow(new IOException("Fake Exception: 410 Gone")) // First attempt fails
      .thenReturn(pipe.sink());                               // second attempt succeeds

  try (PackageUtil directPackageUtil =
      PackageUtil.withExecutorService(MoreExecutors.newDirectExecutorService())) {
    directPackageUtil.stageClasspathElements(
        ImmutableList.of(tmpFile.getAbsolutePath()),
        STAGING_PATH,
        fastNanoClockAndSleeper,
        createOptions);
  } finally {
    verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class));
    verify(mockGcsUtil, times(2)).create(any(GcsPath.class), anyString());
    verifyNoMoreInteractions(mockGcsUtil);
  }
}

项目：beam 文件：PackageUtilTest.java

@Test
public void testPackageUploadIsNotSkippedWhenSizesAreDifferent() throws Exception {
  Pipe pipe = Pipe.open();
  File tmpDirectory = tmpFolder.newFolder("folder");
  tmpFolder.newFolder("folder", "empty_directory");
  tmpFolder.newFolder("folder", "directory");
  makeFileWithContents("folder/file.txt", "This is a test!");
  makeFileWithContents("folder/directory/file.txt", "This is also a test!");
  when(mockGcsUtil.getObjects(anyListOf(GcsPath.class)))
      .thenReturn(ImmutableList.of(StorageObjectOrIOException.create(
          createStorageObject(STAGING_PATH, Long.MAX_VALUE))));
  when(mockGcsUtil.create(any(GcsPath.class), anyString())).thenReturn(pipe.sink());

  defaultPackageUtil.stageClasspathElements(
      ImmutableList.of(tmpDirectory.getAbsolutePath()), STAGING_PATH, createOptions);

  verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class));
  verify(mockGcsUtil).create(any(GcsPath.class), anyString());
  verifyNoMoreInteractions(mockGcsUtil);
}

项目：beam 文件：PackageUtilTest.java

@Test
public void testPackageUploadWithExplicitPackageName() throws Exception {
  Pipe pipe = Pipe.open();
  File tmpFile = makeFileWithContents("file.txt", "This is a test!");
  final String overriddenName = "alias.txt";

  when(mockGcsUtil.getObjects(anyListOf(GcsPath.class)))
      .thenReturn(ImmutableList.of(StorageObjectOrIOException.create(
          new FileNotFoundException("some/path"))));
  when(mockGcsUtil.create(any(GcsPath.class), anyString())).thenReturn(pipe.sink());

  List<DataflowPackage> targets =
      defaultPackageUtil.stageClasspathElements(
          ImmutableList.of(overriddenName + "=" + tmpFile.getAbsolutePath()),
          STAGING_PATH,
          createOptions);
  DataflowPackage target = Iterables.getOnlyElement(targets);

  verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class));
  verify(mockGcsUtil).create(any(GcsPath.class), anyString());
  verifyNoMoreInteractions(mockGcsUtil);

  assertThat(target.getName(), equalTo(overriddenName));
  assertThat(target.getLocation(),
      RegexMatcher.matches(STAGING_PATH + "file-" + HASH_PATTERN + ".txt"));
}

项目：Pronghorn 文件：FuzzDataStageGenerator.java

@Override
protected void buildConstructors(Appendable target, String className) throws IOException {

    target.append("public ").append(className).append("(");

    if (generateRunnable) {
        target.append(") { \n");

        FieldReferenceOffsetManager from = MessageSchema.from(schema);
        if (!from.hasSimpleMessagesOnly) {
            target.append(tab).append("startup();\n");
        }

    } else {        
        target.append(GraphManager.class.getCanonicalName()).append(" gm, ");
        Appendables.appendClass(target, Pipe.class, schema.getClass()).append(" ").append(pipeVarName).append(") {\n");

        target.append("super(gm,NONE,").append(pipeVarName).append(");\n");
        target.append("this.").append(pipeVarName).append(" = ").append(pipeVarName).append(";\n"); 
        Appendables.appendStaticCall(target, Pipe.class, "from").append(pipeVarName).append(").validateGUID(FROM_GUID);\n");
    }

    target.append("}\n\n");

}

项目：pinenut 文件：PipeTest.java

public static void main(String[] args) throws Exception {
  Pipe pipe = Pipe.open();

  Pipe.SinkChannel sinkChannel = pipe.sink();
  Pipe.SourceChannel sourceChannel = pipe.source();
  String newData = "New String to write to file..." + System.currentTimeMillis();
  ByteBuffer buf = ByteBuffer.allocate(48);
  buf.clear();
  buf.put(newData.getBytes());
  buf.flip();

  ByteBuffer bufread = ByteBuffer.allocate(48);

  while (buf.hasRemaining()) {
    sinkChannel.write(buf);
    int bytesRead = sourceChannel.read(bufread);
    System.out.println(bytesRead);
  }

}

项目：nio-http 文件：ChannelQueue.java

public ChannelQueue(Selector readSelector) throws IOException {

        this.pipe = Pipe.open();
        this.sinkChannel = this.pipe.sink();
        this.sourceChannel = this.pipe.source();

        this.sinkChannel.configureBlocking(false);
        this.sourceChannel.configureBlocking(false);

        this.readSelector = readSelector;
        this.queue = new LinkedBlockingQueue();
        this.notificationQueue = 0;
        this.readBuffer = ByteBuffer.allocate(1);

        this.writeBuffer = ByteBuffer.allocate(1);
        this.writeBuffer.put((byte)0x1);
        this.writeBuffer.flip();

        this.sourceChannel.register(this.readSelector, SelectionKey.OP_READ);

    }

项目：ExpectIt 文件：SingleInputExpect.java

protected SingleInputExpect(
        final Pipe.SourceChannel source,
        final Pipe.SinkChannel sink,
        final InputStream input,
        final Charset charset,
        final Appendable echoInput,
        final Filter filter,
        final int bufferSize,
        final boolean autoFlushEcho) throws IOException {
    this.input = input;
    this.charset = charset;
    this.echoInput = echoInput;
    this.filter = filter;
    this.bufferSize = bufferSize;
    this.autoFlushEcho = autoFlushEcho;
    this.source = source;
    this.sink = sink;
    source.configureBlocking(false);
    buffer = new StringBuilder();
}

项目：ExpectIt 文件：MatcherTest.java

/**
 * Creates a mock input stream which send some data every SMALL_TIMEOUT ms.
 */
@Before
public void setup() throws Exception {
    mock = TestUtils.mockInputStream(text);
    final Pipe pipe = Pipe.open();
    input = new SingleInputExpect(
            pipe.source(),
            pipe.sink(),
            mock.getStream(),
            Charset.defaultCharset(),
            null,
            null,
            DEFAULT_BUFFER_SIZE,
            false);
    executor = Executors.newSingleThreadExecutor();
    input.start(executor);
    mock.waitUntilReady();
}

项目：JavaRA 文件：WsaFileCNC.java

/**
 * {@inheritDoc}
 */
@Override
public ReadableByteChannel getImagesData() {

    // Leverage the raw frame decoder as input to the colour decoder
    Pipe pipe = null;
    try {
        pipe = Pipe.open();
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
        return null;
    }
    decoderthreadpool.execute(new ColourFrameDecoder(getRawImageData(), pipe.sink()));
    return pipe.source();
}

项目：JavaRA 文件：WsaFileCNC.java

/**
 * {@inheritDoc}
 */
@Override
public ReadableByteChannel getRawImageData() {

    Pipe pipe;
    try {
        pipe = Pipe.open();
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
        return null;
    }
    decoderthreadpool.execute(new RawFrameDecoder(
            new DuplicateReadOnlyByteChannel(bytechannel), pipe.sink()));
    return pipe.source();
}

项目：JavaRA 文件：VqaFile.java

/**
 * {@inheritDoc}
 */
@Override
public ReadableByteChannel getImagesData() {

    Pipe pipe = null;
    try {
        pipe = Pipe.open();
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
        return null;
    }
    decoderthreadpool.execute(new ImageDataDecoder(
            new DuplicateReadOnlyByteChannel(bytechannel), pipe.sink()));
    return pipe.source();
}

项目：JavaRA 文件：VqaFile.java

/**
 * {@inheritDoc}
 */
@Override
public ReadableByteChannel getSoundData() {
    Pipe pipe = null;

    try {
        pipe = Pipe.open();
    } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
        return null;
    }

    decoderthreadpool.execute(new SoundDataDecoder(
            new DuplicateReadOnlyByteChannel(bytechannel), pipe.sink()));
    return pipe.source();
}

项目：OpenJSharp 文件：WindowsSelectorImpl.java

WindowsSelectorImpl(SelectorProvider sp) throws IOException {
    super(sp);
    pollWrapper = new PollArrayWrapper(INIT_CAP);
    wakeupPipe = Pipe.open();
    wakeupSourceFd = ((SelChImpl)wakeupPipe.source()).getFDVal();

    // Disable the Nagle algorithm so that the wakeup is more immediate
    SinkChannelImpl sink = (SinkChannelImpl)wakeupPipe.sink();
    (sink.sc).socket().setTcpNoDelay(true);
    wakeupSinkFd = ((SelChImpl)sink).getFDVal();

    pollWrapper.addWakeupSocket(wakeupSourceFd, 0);
}

项目：OpenJSharp 文件：DotNetSelectorImpl.java

DotNetSelectorImpl(SelectorProvider sp) throws IOException
{
    super(sp);
    wakeupPipe = Pipe.open();
    wakeupSourceFd = ((SelChImpl)wakeupPipe.source()).getFD().getSocket();

    // Disable the Nagle algorithm so that the wakeup is more immediate
    SinkChannelImpl sink = (SinkChannelImpl)wakeupPipe.sink();
    (sink.sc).socket().setTcpNoDelay(true);
    wakeupSinkFd = ((SelChImpl)sink).getFD().getSocket();
}

项目：jdk8u-jdk 文件：WindowsSelectorImpl.java

WindowsSelectorImpl(SelectorProvider sp) throws IOException {
    super(sp);
    pollWrapper = new PollArrayWrapper(INIT_CAP);
    wakeupPipe = Pipe.open();
    wakeupSourceFd = ((SelChImpl)wakeupPipe.source()).getFDVal();

    // Disable the Nagle algorithm so that the wakeup is more immediate
    SinkChannelImpl sink = (SinkChannelImpl)wakeupPipe.sink();
    (sink.sc).socket().setTcpNoDelay(true);
    wakeupSinkFd = ((SelChImpl)sink).getFDVal();

    pollWrapper.addWakeupSocket(wakeupSourceFd, 0);
}

项目：jdk8u-jdk 文件：PipeInterrupt.java

public void test() throws Exception {

        Thread tester = new Thread("PipeTester") {
            private Pipe testPipe = null;

            @Override
            public void run() {
                for (;;) {
                    boolean interrupted = this.isInterrupted();
                    try {
                        testPipe = Pipe.open();
                        close();
                        if (interrupted) {
                            if (!this.isInterrupted())
                               exc = new RuntimeException("interrupt status reset");
                            break;
                        }
                    } catch (IOException ioe) {
                        exc = ioe;
                    }
                }
            }

            private void close() throws IOException {
                if (testPipe != null) {
                    testPipe.sink().close();
                    testPipe.source().close();
                }
            }
        };

        tester.start();
        Thread.sleep(200);
        tester.interrupt();
        tester.join();

        if (exc != null)
            throw exc;
    }

项目：openjdk-jdk10 文件：WindowsSelectorImpl.java

WindowsSelectorImpl(SelectorProvider sp) throws IOException {
    super(sp);
    pollWrapper = new PollArrayWrapper(INIT_CAP);
    wakeupPipe = Pipe.open();
    wakeupSourceFd = ((SelChImpl)wakeupPipe.source()).getFDVal();

    // Disable the Nagle algorithm so that the wakeup is more immediate
    SinkChannelImpl sink = (SinkChannelImpl)wakeupPipe.sink();
    (sink.sc).socket().setTcpNoDelay(true);
    wakeupSinkFd = ((SelChImpl)sink).getFDVal();

    pollWrapper.addWakeupSocket(wakeupSourceFd, 0);
}

项目：openjdk-jdk10 文件：Open.java

static void test3() {
    SelectorProvider sp = SelectorProvider.provider();
    for (int i=0; i<11000; i++) {
        try {
            Pipe p = sp.openPipe();
        } catch (Exception e) {
            // Presumably "Too many open files"
        }
    }
}

项目：openjdk-jdk10 文件：PipeInterrupt.java

public void test() throws Exception {

        Thread tester = new Thread("PipeTester") {
            private Pipe testPipe = null;

            @Override
            public void run() {
                for (;;) {
                    boolean interrupted = this.isInterrupted();
                    try {
                        testPipe = Pipe.open();
                        close();
                        if (interrupted) {
                            if (!this.isInterrupted())
                               exc = new RuntimeException("interrupt status reset");
                            break;
                        }
                    } catch (IOException ioe) {
                        exc = ioe;
                    }
                }
            }

            private void close() throws IOException {
                if (testPipe != null) {
                    testPipe.sink().close();
                    testPipe.source().close();
                }
            }
        };

        tester.start();
        Thread.sleep(200);
        tester.interrupt();
        tester.join();

        if (exc != null)
            throw exc;
    }

项目：openjdk9 文件：WindowsSelectorImpl.java

WindowsSelectorImpl(SelectorProvider sp) throws IOException {
    super(sp);
    pollWrapper = new PollArrayWrapper(INIT_CAP);
    wakeupPipe = Pipe.open();
    wakeupSourceFd = ((SelChImpl)wakeupPipe.source()).getFDVal();

    // Disable the Nagle algorithm so that the wakeup is more immediate
    SinkChannelImpl sink = (SinkChannelImpl)wakeupPipe.sink();
    (sink.sc).socket().setTcpNoDelay(true);
    wakeupSinkFd = ((SelChImpl)sink).getFDVal();

    pollWrapper.addWakeupSocket(wakeupSourceFd, 0);
}

项目：openjdk9 文件：PipeInterrupt.java

public void test() throws Exception {

        Thread tester = new Thread("PipeTester") {
            private Pipe testPipe = null;

            @Override
            public void run() {
                for (;;) {
                    boolean interrupted = this.isInterrupted();
                    try {
                        testPipe = Pipe.open();
                        close();
                        if (interrupted) {
                            if (!this.isInterrupted())
                               exc = new RuntimeException("interrupt status reset");
                            break;
                        }
                    } catch (IOException ioe) {
                        exc = ioe;
                    }
                }
            }

            private void close() throws IOException {
                if (testPipe != null) {
                    testPipe.sink().close();
                    testPipe.source().close();
                }
            }
        };

        tester.start();
        Thread.sleep(200);
        tester.interrupt();
        tester.join();

        if (exc != null)
            throw exc;
    }

项目：jovial 文件：JovialTestExecuter.java

private InputStream createInput(Map<String, List<String>> config) {
    try {
        Pipe pipe = Pipe.open();
        OutputStream rawOutput = Channels.newOutputStream(pipe.sink());
        ObjectOutputStream output = new ObjectOutputStream(rawOutput);
        output.writeObject(config);
        return Channels.newInputStream(pipe.source());
    } catch (IOException e) {
        throw new UncheckedIOException("Could not open pipe.", e);
    }
}

项目：jdk8u_jdk 文件：WindowsSelectorImpl.java

WindowsSelectorImpl(SelectorProvider sp) throws IOException {
    super(sp);
    pollWrapper = new PollArrayWrapper(INIT_CAP);
    wakeupPipe = Pipe.open();
    wakeupSourceFd = ((SelChImpl)wakeupPipe.source()).getFDVal();

    // Disable the Nagle algorithm so that the wakeup is more immediate
    SinkChannelImpl sink = (SinkChannelImpl)wakeupPipe.sink();
    (sink.sc).socket().setTcpNoDelay(true);
    wakeupSinkFd = ((SelChImpl)sink).getFDVal();

    pollWrapper.addWakeupSocket(wakeupSourceFd, 0);
}

项目：jdk8u_jdk 文件：PipeInterrupt.java

public void test() throws Exception {

        Thread tester = new Thread("PipeTester") {
            private Pipe testPipe = null;

            @Override
            public void run() {
                for (;;) {
                    boolean interrupted = this.isInterrupted();
                    try {
                        testPipe = Pipe.open();
                        close();
                        if (interrupted) {
                            if (!this.isInterrupted())
                               exc = new RuntimeException("interrupt status reset");
                            break;
                        }
                    } catch (IOException ioe) {
                        exc = ioe;
                    }
                }
            }

            private void close() throws IOException {
                if (testPipe != null) {
                    testPipe.sink().close();
                    testPipe.source().close();
                }
            }
        };

        tester.start();
        Thread.sleep(200);
        tester.interrupt();
        tester.join();

        if (exc != null)
            throw exc;
    }

项目：lookaside_java-1.8.0-openjdk 文件：WindowsSelectorImpl.java

WindowsSelectorImpl(SelectorProvider sp) throws IOException {
    super(sp);
    pollWrapper = new PollArrayWrapper(INIT_CAP);
    wakeupPipe = Pipe.open();
    wakeupSourceFd = ((SelChImpl)wakeupPipe.source()).getFDVal();

    // Disable the Nagle algorithm so that the wakeup is more immediate
    SinkChannelImpl sink = (SinkChannelImpl)wakeupPipe.sink();
    (sink.sc).socket().setTcpNoDelay(true);
    wakeupSinkFd = ((SelChImpl)sink).getFDVal();

    pollWrapper.addWakeupSocket(wakeupSourceFd, 0);
}

项目：lookaside_java-1.8.0-openjdk 文件：PipeInterrupt.java

public void test() throws Exception {

        Thread tester = new Thread("PipeTester") {
            private Pipe testPipe = null;

            @Override
            public void run() {
                for (;;) {
                    boolean interrupted = this.isInterrupted();
                    try {
                        testPipe = Pipe.open();
                        close();
                        if (interrupted) {
                            if (!this.isInterrupted())
                               exc = new RuntimeException("interrupt status reset");
                            break;
                        }
                    } catch (IOException ioe) {
                        exc = ioe;
                    }
                }
            }

            private void close() throws IOException {
                if (testPipe != null) {
                    testPipe.sink().close();
                    testPipe.source().close();
                }
            }
        };

        tester.start();
        Thread.sleep(200);
        tester.interrupt();
        tester.join();

        if (exc != null)
            throw exc;
    }

项目：beam 文件：PackageUtilTest.java

@Test
public void testPackageUploadWithDirectorySucceeds() throws Exception {
  Pipe pipe = Pipe.open();
  File tmpDirectory = tmpFolder.newFolder("folder");
  tmpFolder.newFolder("folder", "empty_directory");
  tmpFolder.newFolder("folder", "directory");
  makeFileWithContents("folder/file.txt", "This is a test!");
  makeFileWithContents("folder/directory/file.txt", "This is also a test!");

  when(mockGcsUtil.getObjects(anyListOf(GcsPath.class)))
      .thenReturn(ImmutableList.of(StorageObjectOrIOException.create(
          new FileNotFoundException("some/path"))));
  when(mockGcsUtil.create(any(GcsPath.class), anyString())).thenReturn(pipe.sink());

  defaultPackageUtil.stageClasspathElements(
      ImmutableList.of(tmpDirectory.getAbsolutePath()), STAGING_PATH, createOptions);

  verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class));
  verify(mockGcsUtil).create(any(GcsPath.class), anyString());
  verifyNoMoreInteractions(mockGcsUtil);

  ZipInputStream inputStream = new ZipInputStream(Channels.newInputStream(pipe.source()));
  List<String> zipEntryNames = new ArrayList<>();
  for (ZipEntry entry = inputStream.getNextEntry(); entry != null;
      entry = inputStream.getNextEntry()) {
    zipEntryNames.add(entry.getName());
  }

  assertThat(zipEntryNames,
      containsInAnyOrder("directory/file.txt", "empty_directory/", "file.txt"));
}

项目：JavaNote 文件：PipeTest.java

public static void main(String[] args){
    Pipe pipe;
    Pipe.SinkChannel sinkChannel;
    Pipe.SourceChannel sourceChannel;

    try {
        pipe = Pipe.open();
        new Thread(new ReadWork(pipe.source())).start();
        new Thread(new WriteWork(pipe.sink())).start();

    } catch (IOException e) {
        e.printStackTrace();
    }
}

项目：qpid-proton-j 文件：EchoInputStreamWrapper.java

public static SourceChannel wrap(InputStream in) throws IOException {
    Pipe pipe = Pipe.open();
    new EchoInputStreamWrapper(in, pipe.sink()).start();
    SourceChannel result = pipe.source();
    result.configureBlocking(false);
    return result;
}

项目：Pronghorn 文件：PhastEncoderStageGenerator.java

/**
 * This method is to be overridden and called on by the super class. It should not be called anywhere but by the
 * super class
 *
 * @param target    where the code is being written in super class
 * @param className the base class name, without extensions or implementations
 * @throws IOException when the target can not be written to
 */
@Override
protected void buildConstructors(Appendable target, String className) throws IOException {
    target.append("public ").append(className).append("(");
    if (generateRunnable) {
        target.append(") { \n");

        FieldReferenceOffsetManager from = MessageSchema.from(schema);
        if (!from.hasSimpleMessagesOnly) {
            target.append(tab).append("startup();\n");
        }

    } else {
        target.append(GraphManager.class.getCanonicalName()).append(" gm, ");

        target.append("Pipe<" + schema.getClass().getSimpleName() + ">  " + inPipeName + ", ");
        target.append("Pipe<RawDataSchema> " + outPipeName );
        additionalArgs(target);
        target.append(") {\n");

        target.append(tab).append("super(gm," + inPipeName + ",").append(outPipeName).append(");\n");
        target.append(tab).append("this." + outPipeName + " = " + outPipeName + ";\n");
        target.append(tab + "this." + inPipeName + " = " + inPipeName + ";\n");
        target.append(tab);
        Appendables.appendStaticCall(target, Pipe.class, "from").append(inPipeName)
                .append(").validateGUID(FROM_GUID);\n");
    }
    target.append(tab + intDictionaryName + " = FROM.newIntDefaultsDictionary();\n");
    target.append(tab + longDictionaryName + " = FROM.newLongDefaultsDictionary();\n");
    target.append(tab + defIntDictionaryName + " = FROM.newIntDefaultsDictionary();\n");
    target.append(tab + defLongDictionaryName + " = FROM.newLongDefaultsDictionary();\n");
    additionalConstructorLogic(target);
    target.append("}\n\n");
}

项目：Pronghorn 文件：FuzzDataStageGenerator.java

@Override
protected void bodyOfNextMessageIdx(Appendable target) throws IOException {
    msgGenerator.preCall(target);

    int startsCount = MessageSchema.from(schema).messageStarts().length;

    if (startsCount==1) {
        target.append(tab).append("return ");
        Appendables.appendValue(target, MessageSchema.from(schema).messageStarts()[0]).append(";\n");
    } else {
        target.append(tab).append("return ");

        if (null==pipeVarName) {
            if (!(schema instanceof MessageSchemaDynamic)) {
                target.append(schema.getClass().getSimpleName()).append(".");
            }

            target.append("FROM");
        } else {
            Appendables.appendStaticCall(target, Pipe.class, "from").append(pipeVarName).append(")");
        }

        target.append(".messageStarts[(");
        msgGenerator.result(target).append(")%");
        Appendables.appendValue(target, startsCount).append("];\n");


    }
}

项目：Pronghorn 文件：PhastDecoderStageGenerator.java

/**
 * This method is to be overridden and called on by the super class. It should not be called anywhere but by the
 * super class
 *
 * @param target    where the code is being written in super class
 * @param className the base class name, without extensions or implementations
 * @throws IOException when the target can not be written to
 */
@Override
protected void buildConstructors(Appendable target, String className) throws IOException {

    target.append("public ").append(className).append("(");
    if (generateRunnable) {
        target.append(") { \n");

        FieldReferenceOffsetManager from = MessageSchema.from(schema);
        if (!from.hasSimpleMessagesOnly) {
            target.append(tab).append("startup();\n");
        }

    } else {
        target.append(GraphManager.class.getCanonicalName()).append(" gm, ");
        target.append("Pipe<RawDataSchema>  " + inPipeName + ", ");
        Appendables.appendClass(target, Pipe.class, schema.getClass()).append(" ").append(pipeVarName);
        additionalArgs(target);
        target.append(") {\n");

        target.append(tab).append("super(gm," + inPipeName + ",").append(pipeVarName).append(");\n");
        target.append(tab).append("this.").append(pipeVarName).append(" = ").append(pipeVarName).append(";\n");
        target.append(tab);
        Appendables.appendStaticCall(target, Pipe.class, "from").append(pipeVarName).append(").validateGUID(FROM_GUID);\n");
        target.append(tab + "this." + inPipeName + " = " + inPipeName + ";\n");
    }
    target.append(tab + intDictionaryName + " = FROM.newIntDefaultsDictionary();\n");
    target.append(tab + longDictionaryName + " = FROM.newLongDefaultsDictionary();\n");
    target.append(tab + defaultIntDictionaryName + " = FROM.newIntDefaultsDictionary();\n");
    target.append(tab + defaultLongDictionaryName + " = FROM.newLongDefaultsDictionary();\n");
    interfaceSetup(target);
    target.append("}\n\n");

}

项目：silverflash 文件：PipeTransport.java

private synchronized void open() throws IOException {
  if (inboundSource == null) {
    SelectorProvider provider = SelectorProvider.provider();
    Pipe inboundPipe = provider.openPipe();
    inboundSource = inboundPipe.source();
    inboundSink = inboundPipe.sink();
    Pipe outboundPipe = provider.openPipe();
    outboundSource = outboundPipe.source();
    outboundSink = outboundPipe.sink();
  }
}