import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
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();
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
@Override
public void onSelectableReadable(Event event) {
    Selectable selectable = event.getSelectable();

    // The onSelectableReadable event tells us that there is data
    // to be read, or the end of stream has been reached.
    SourceChannel channel = (SourceChannel)selectable.getChannel();
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    try {
        while(true) {
            int amount = channel.read(buffer);
            if (amount < 0) {
                selectable.terminate();
                selectable.getReactor().update(selectable);
            }
            if (amount <= 0) break;
            System.out.write(buffer.array(), 0, buffer.position());
            buffer.clear();
        }
    } catch(IOException ioException) {
        ioException.printStackTrace();
        selectable.terminate();
        selectable.getReactor().update(selectable);
    }
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
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;
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
public static void main(String[] args) throws IOException {
    if (args.length != 1) {
        System.err.println("Specify a file name as an argument.");
        System.exit(1);
    }
    FileInputStream inFile = new FileInputStream(args[0]);
    SourceChannel inChannel = EchoInputStreamWrapper.wrap(inFile);
    Reactor reactor = Proton.reactor(new Cat(inChannel));
    reactor.run();
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
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());
    }
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
/**
 * @tests java.nio.channels.Pipe#source()
 */
@TestTargetNew(
    level = TestLevel.COMPLETE,
    notes = "",
    method = "source",
    args = {}
)
public void test_source() throws IOException {
    Pipe pipe = Pipe.open();
    SourceChannel source = pipe.source();
    assertTrue(source.isBlocking());
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
@TestTargetNew(
    level = TestLevel.COMPLETE,
    notes = "",
    method = "SourceChannel",
    args = {java.nio.channels.spi.SelectorProvider.class}
)
public void testConstructor() throws IOException {
    SourceChannel channel =
            SelectorProvider.provider().openPipe().source();
    assertNotNull(channel);
    assertSame(SelectorProvider.provider(),channel.provider());
    channel = Pipe.open().source();
    assertNotNull(channel);
    assertSame(SelectorProvider.provider(),channel.provider());
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
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);
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
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);
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
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;
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
private Echo(SourceChannel channel) {
    this.channel = channel;
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
public static void main(String[] args) throws IOException {
    SourceChannel inChannel = EchoInputStreamWrapper.wrap(System.in);
    Reactor reactor = Proton.reactor(new Echo(inChannel));
    reactor.run();
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
private Cat(SourceChannel channel) {
    this.channel = channel;
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
/**
 * @tests java.nio.channels.Pipe#source()
 */
public void test_source() throws IOException {
	Pipe pipe = Pipe.open();
	SourceChannel source = pipe.source();
	assertTrue(source.isBlocking());
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
public FASTInputSourceChannel(SourceChannel channel) {
	this.sourceChannel = channel;
	assert(!channel.isBlocking()) : "Only non blocking SocketChannel is supported.";
}
 

import java.nio.channels.Pipe.SourceChannel; //导入依赖的package包/类
/**
 * 
 * @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);
}