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

private void run() throws Exception
{
    NioThread nt=new NioThread();
    final Socket s=new Socket("localhost", 9999);
    SourceChannel in=ConnectNio.inputStreamToPipe(s.getInputStream());
    in.configureBlocking(false);
    SinkChannel out=ConnectNio.outputStreamToPipe(s.getOutputStream(), s);
    out.configureBlocking(false);
    DualChannelProcessorMultiplexer multiplexer=new DualChannelProcessorMultiplexer(nt, in, out, false, 
            CoolRMINioRemoter.clientId, CoolRMINioRemoter.serverId);
    CoolRMINioClient cli=new CoolRMINioClient(getClass().getClassLoader(), false);
    cli.connect(multiplexer);
    multiplexer.start();
    nt.start();
    Iremote r= (Iremote)cli.getService(Iremote.class, Iremote.class.getName());
    System.out.println(""+r.getValue("Kitten"));
    System.out.println(""+r.getValue("Kitten"));
    System.out.println(""+r.getValue("Kitten"));
    System.out.println(""+r.getValue("Kitten"));
    cli.close();
    nt.close();
}

@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"));
}

private boolean isOpen(Object resource) {
    if (resource instanceof SourceChannel) {
        return ((SourceChannel)resource).isOpen();
    } else if (resource instanceof SinkChannel) {
        return ((SinkChannel)resource).isOpen();
    } else if (resource instanceof Selector) {
        return ((Selector)resource).isOpen();
    } else if (resource instanceof ServerSocketChannel) {
        return ((ServerSocketChannel)resource).isOpen();
    } else if (resource instanceof SocketChannel) {
        return ((SocketChannel)resource).isOpen();
    } else {
        throw new AssertionFailedError("Don't know how to check if this type is open: " + resource.getClass());
    }
}

@TestTargetNew(
    level = TestLevel.COMPLETE,
    notes = "",
    method = "SinkChannel",
    args = {java.nio.channels.spi.SelectorProvider.class}
)
public void testConstructor() throws IOException {
    SinkChannel channel =
            SelectorProvider.provider().openPipe().sink();
    assertNotNull(channel);
    assertSame(SelectorProvider.provider(),channel.provider());
    channel = Pipe.open().sink();
    assertNotNull(channel);
    assertSame(SelectorProvider.provider(),channel.provider());
}

/**
 * @tests java.nio.channels.Pipe#sink()
 */
@TestTargetNew(
    level = TestLevel.COMPLETE,
    notes = "",
    method = "sink",
    args = {}
)
public void test_sink() throws IOException {
    Pipe pipe = Pipe.open();
    SinkChannel sink = pipe.sink();
    assertTrue(sink.isBlocking());
}

HandshakeHandler(boolean clientMode, SourceChannel in, SinkChannel out)
        throws SSLException {
    this.in = in;
    this.out = out;
    engine = getEngine();
    engine.setUseClientMode(clientMode);
    String[] cipherSuites = engine.getSupportedCipherSuites();
    Set<String> enabledSuites = new HashSet<String>();
    for (String cipherSuite : cipherSuites) {
        if (cipherSuite.contains("anon")) {
            enabledSuites.add(cipherSuite);
        }
    }
    engine.setEnabledCipherSuites((String[]) enabledSuites.toArray(
            new String[enabledSuites.size()]));

    engine.beginHandshake();
    status = engine.getHandshakeStatus();

    if (clientMode) {
        LOGTAG = "CLIENT: ";
    } else {
        LOGTAG = "SERVER: ";
    }

    log("CipherSuites: " + Arrays.toString(engine.getEnabledCipherSuites()));
    log(status);

    readBuffer = ByteBuffer.allocate(200000);
    writeBuffer = ByteBuffer.allocate(20000);
}

void prepareEngines() throws IOException {
    Pipe clientSendPipe = Pipe.open();
    Pipe serverSendPipe = Pipe.open();

    SinkChannel clientSink = clientSendPipe.sink();
    SourceChannel serverSource = clientSendPipe.source();
    SinkChannel serverSink = serverSendPipe.sink();
    SourceChannel clientSource = serverSendPipe.source();

    clientEngine = new HandshakeHandler(true, clientSource, clientSink);
    serverEngine = new HandshakeHandler(false, serverSource, serverSink);
}

PipeSelectableChannel(final SinkChannel out, final SourceChannel in) {
    if (out == null) {
        throw new NullPointerException("out == null");
    }
    if (in == null) {
        throw new NullPointerException("in == null");
    }
    if ((in.validOps() & out.validOps()) != 0) {
        throw new IllegalArgumentException("sink and source have overlapping validOps");
    }
    this.out = out;
    this.in = in;
}

private EchoInputStreamWrapper(InputStream in, SinkChannel out) {
    this.in = in;
    this.out = out;
    setName(getClass().getName() + "-" + idCounter.incrementAndGet());
    setDaemon(true);
}

public mai(Context paramContext, File paramFile, String paramString, long paramLong1, long paramLong2, Map<String, String> paramMap)
{
  if ((paramLong1 < 0L) || ((paramLong2 != -1L) && (paramLong1 > paramLong2))) {
    throw new IllegalArgumentException("Invalid stream limits");
  }
  this.f = paramLong1;
  this.d = paramLong2;
  long l1;
  if (paramFile == null)
  {
    l1 = 0L;
    this.e = l1;
    if (this.f >= this.e) {
      break label216;
    }
    this.b = new RandomAccessFile(paramFile, "r");
    this.b.seek(this.f);
  }
  for (;;)
  {
    if (paramString == null) {
      break label224;
    }
    Pipe localPipe = Pipe.open();
    this.c = Channels.newInputStream(localPipe.source());
    Pipe.SinkChannel localSinkChannel = localPipe.sink();
    this.a = ixd.a(paramContext).a(paramString, 4, paramMap, localSinkChannel, this);
    long l2 = Math.max(this.e, this.f);
    if (l2 != 0L)
    {
      new StringBuilder(41).append("Starting request at: ").append(l2);
      this.a.a(l2);
    }
    this.a.f();
    return;
    l1 = paramFile.length();
    break;
    label216:
    this.b = null;
  }
  label224:
  this.a = null;
  this.c = null;
}

public OutPipe(SinkChannel sink) throws IOException {
    s = sink;
    s.configureBlocking(true);
}

public OutPipe(SinkChannel sink) throws IOException {
    s = sink;
    s.configureBlocking(true);
}

/**
 * @tests java.nio.channels.Pipe#sink()
 */
public void test_sink() throws IOException {
    Pipe pipe = Pipe.open();
    SinkChannel sink = pipe.sink();
    assertTrue(sink.isBlocking());
}

/**
 * @tests java.nio.channels.Pipe#sink()
 */
public void test_sink() throws IOException {
    Pipe pipe = Pipe.open();
    SinkChannel sink = pipe.sink();
    assertTrue(sink.isBlocking());
}

/**
 * @tests java.nio.channels.Pipe#sink()
 */
public void test_sink() throws IOException {
    Pipe pipe = Pipe.open();
    SinkChannel sink = pipe.sink();
    assertTrue(sink.isBlocking());
}

/**
 * 
 * @param t
 * @param c
 * @param client
 * @param thisId Identifier of this multiplexer endpoint. This is sent to the client on connection.
 * @param remoteId Required identifier of the other endpoint. This is checked to be equal to the value received from the client.
 */
public DualChannelProcessorMultiplexer(NioThread t, SourceChannel source, SinkChannel sink, boolean client, byte[] thisId, byte[] remoteId) {
    super(thisId, remoteId);
    cpSink=new CPSink(t, sink, client);
    cpSource=new CPSource(t, source, client);
}