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

@Override
protected FrameBuffer createFrameBuffer(TNonblockingTransport trans, SelectionKey selectionKey,
                                        AbstractSelectThread selectThread) {
    TrackingFrameBuffer frameBuffer = new TrackingFrameBuffer(trans, selectionKey, selectThread);
    if (trans instanceof TNonblockingSocket) {
        try {
            SocketChannel socketChannel = ((TNonblockingSocket) trans).getSocketChannel();
            InetAddress addr = ((InetSocketAddress) socketChannel.getRemoteAddress()).getAddress();
            clientAddresses.put(frameBuffer.getInputFramedTransport(), addr);
        } catch (IOException e) {
            log.warn("Exception while tracking client address", e);
            clientAddresses.remove(frameBuffer.getInputFramedTransport());
        }
    } else {
        log.warn("Unknown TNonblockingTransport instance: {}", trans.getClass().getName());
        clientAddresses.remove(frameBuffer.getInputFramedTransport());
    }
    return frameBuffer;
}

protected NioTcpMessageChannel(NioTcpMessageProcessor nioTcpMessageProcessor,
        SocketChannel socketChannel) throws IOException {
    super(nioTcpMessageProcessor.getSIPStack());
    super.myClientInputStream = socketChannel.socket().getInputStream();
    try {
        this.peerAddress = socketChannel.socket().getInetAddress();
        this.peerPort = socketChannel.socket().getPort();
        this.socketChannel = socketChannel;
        super.mySock = socketChannel.socket();
        // messages that we write out to him.
        nioParser = new NioPipelineParser(sipStack, this,
                this.sipStack.getMaxMessageSize());
        this.peerProtocol = nioTcpMessageProcessor.transport;
        lastActivityTimeStamp = System.currentTimeMillis();
        super.key = MessageChannel.getKey(peerAddress, peerPort, nioTcpMessageProcessor.transport);

           myAddress = nioTcpMessageProcessor.getIpAddress().getHostAddress();
           myPort = nioTcpMessageProcessor.getPort();

    } finally {
        if (logger.isLoggingEnabled(LogWriter.TRACE_DEBUG)) {
            logger.logDebug("Done creating NioTcpMessageChannel " + this + " socketChannel = " +socketChannel);
        }
    }

}

private void getData(ByteBuffer input, SocketChannel channel) {
    int id = input.getInt();
    byte[] entity = data.get(id);

    ByteBuffer outputBuffer = TCPServer.prepareEmptyBuffer();
    int entityOffset = 0;

    while ((entity.length - entityOffset) > TCPServer.MAX_MESSAGE_BYTES) {
        outputBuffer.rewind();
        outputBuffer.put(entity, entityOffset, TCPServer.MAX_MESSAGE_BYTES);
        TCPServer.sendMessage(channel, outputBuffer);
        entityOffset += TCPServer.MAX_MESSAGE_BYTES;
    }
    outputBuffer.rewind();
    outputBuffer.put(entity, entityOffset, entity.length - entityOffset);
    outputBuffer.limit(entity.length - entityOffset);
    TCPServer.sendMessage(channel, outputBuffer);
}

private void start0() throws IOException {
    while (!closed) {

        processQueues();
        selector.select();

        if (selector.selectedKeys().isEmpty()) {
            processQueues();
        }

        for (Iterator<SelectionKey> i = selector.selectedKeys().iterator(); i.hasNext(); ) {
            SelectionKey key = i.next();
            i.remove();

            if (key.isConnectable()) {
                SocketChannel socketChannel = (SocketChannel) key.channel();
                socketChannel.finishConnect();
                LOG.info("SocketChannel connected.");
            }

            if (key.isReadable()) {
                this.socketChannelReader.readFromKey(key);
            }
        }
    }
}

private void handle(SelectionKey key) {
    SocketChannel channel = (SocketChannel) key.channel();
    if (key.isConnectable()) {
        try {
            if (channel.finishConnect()) {
                //connect finish
                this.logger.info("[Connecter] finish connect " + channel.getRemoteAddress().toString());
                IoWorker worker = this.workers.get(workersIndex);
                worker.dispatch(new JobBean(channel, this.chanToParam.get(channel)));
                workersIndex = (workersIndex + 1) % workers.size();
            }
        } catch (IOException e) {
            this.logger.info("[Connecter] finish connect error : " + e.toString());
            ClientParam clientParam = this.chanToParam.get(channel);
            if (clientParam.getOnConnectError() != null) {
                clientParam.getOnConnectError().onConnectError(e);
            }
            this.chanToParam.remove(channel);
            try {
                channel.close();
            } catch (IOException e1) {
                // already close
            }
        }
    }
}

/**
 * Expected to be used by the Poller to release resources on socket
 * close, errors etc.
 */
@Override
public void release(SocketChannel socket) {
    if (log.isDebugEnabled()) 
        log.debug("Iterating through our connections to release a socket channel:"+socket);
    boolean released = false;
    Iterator<java.util.Map.Entry<NioChannel, Processor<NioChannel>>> it = connections.entrySet().iterator();
    while (it.hasNext()) {
        java.util.Map.Entry<NioChannel, Processor<NioChannel>> entry = it.next();
        if (entry.getKey().getIOChannel()==socket) {
            it.remove();
            Processor<NioChannel> result = entry.getValue();
            result.recycle(true);
            unregister(result);
            released = true;
            break;
        }
    }
    if (log.isDebugEnabled()) 
        log.debug("Done iterating through our connections to release a socket channel:"+socket +" released:"+released);
}

void createOutgoingChannel(final short p_nodeID) throws NetworkException {
    try {
        m_outgoingChannel = SocketChannel.open();
        m_outgoingChannel.configureBlocking(false);
        m_outgoingChannel.socket().setSoTimeout(0);
        m_outgoingChannel.socket().setTcpNoDelay(true);
        m_outgoingChannel.socket().setReceiveBufferSize(32);
        m_outgoingChannel.socket().setSendBufferSize(m_bufferSize);
        int sendBufferSize = m_outgoingChannel.socket().getSendBufferSize();
        if (sendBufferSize < m_bufferSize) {
            // #if LOGGER >= WARN
            LOGGER.warn("Send buffer size could not be set properly. Check OS settings! Requested: %d, actual: %d", m_bufferSize, sendBufferSize);
            // #endif /* LOGGER >= WARN */
        }

        m_outgoingChannel.connect(m_nodeMap.getAddress(p_nodeID));
    } catch (final IOException ignored) {
        throw new NetworkException("Creating outgoing channel failed");
    }
}

@Override
public InetSocketAddress
getAddress()
{
    SocketChannel channel = ep.getSocketChannel();

    if ( channel != null ){

        Socket socket = channel.socket();

        if ( socket != null ){

            return((InetSocketAddress)socket.getLocalSocketAddress());
        }
    }

    return( null );
}

public SecureNioChannel(SocketChannel channel, SSLEngine engine, ApplicationBufferHandler bufHandler,
        NioSelectorPool pool) throws IOException {
    super(channel, bufHandler);
    this.sslEngine = engine;
    int appBufSize = sslEngine.getSession().getApplicationBufferSize();
    int netBufSize = sslEngine.getSession().getPacketBufferSize();
    // allocate network buffers - TODO, add in optional direct non-direct
    // buffers
    if (netInBuffer == null)
        netInBuffer = ByteBuffer.allocateDirect(netBufSize);
    if (netOutBuffer == null)
        netOutBuffer = ByteBuffer.allocateDirect(netBufSize);

    // selector pool for blocking operations
    this.pool = pool;

    // ensure that the application has a large enough read/write buffers
    // by doing this, we should not encounter any buffer overflow errors
    bufHandler.expand(bufHandler.getReadBuffer(), appBufSize);
    bufHandler.expand(bufHandler.getWriteBuffer(), appBufSize);
    reset();
}

public Client(Selector selector, SocketChannel clientChannel, CloseListener<Client> closeListener) throws ClosedChannelException {
    id = nextId++;
    this.clientChannel = clientChannel;
    router = new Router(this, selector);
    pendingIdBuffer = createIntBuffer(id);

    SelectionHandler selectionHandler = (selectionKey) -> {
        if (selectionKey.isValid() && selectionKey.isWritable()) {
            processSend();
        }
        if (selectionKey.isValid() && selectionKey.isReadable()) {
            processReceive();
        }
        if (selectionKey.isValid()) {
            updateInterests();
        }
    };
    // on start, we are interested only in writing (we must first send the client id)
    interests = SelectionKey.OP_WRITE;
    selectionKey = clientChannel.register(selector, interests, selectionHandler);

    this.closeListener = closeListener;
}

@Override
public NonBlockingConnection startNewConnection() {
    try {
        SocketChannel socket = SocketChannel.open();
        NonBlockingConnection connection = new NonBlockingConnection(socket);

        // this connect is non-blocking and should always return false.
        boolean finished = ((SocketChannel) connection.getChannel()).connect(address);
        if (finished) {
            throw new IllegalStateException("async connect finished instantly?");
        }
        return connection;
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

public void run() {
    SocketChannel socketChannel = null;

    while (accept) {

        try {
            socketChannel = _serverChannel.accept();
            if (!accept) {
                break;
            }
            // socket.setSoTimeout(getTimeout());
        } catch (final IOException e) {
            if (accept) {
                logger.log(Level.WARNING, "ServerNIO died: "
                        + e.getMessage(), e);
            }
            break;
        }
        handleConnection(socketChannel);
    }
}

/**
 * testStartConnection
 */
@Test
public void testStartConnection() throws Exception {
    System.err.println("testStartConnection()");
    for (final HStoreCoordinator m : this.coordinators) {
        // Check that the messenger state is correct
        assert (m.isStarted());

        // Check that the messenger's listener thread is running
        assert (m.getListenerThread().isAlive());

        // Check that we can connect to the messenger's listening port
        int port = m.getLocalMessengerPort();
        SocketChannel channel = SocketChannel.open();
        channel.connect(new InetSocketAddress(port));
        assert (channel.isConnected());
    } // FOR
}

public Connection(SocketChannel channel, long lastContact) {
  this.channel = channel;
  this.lastContact = lastContact;
  this.data = null;
  this.dataLengthBuffer = ByteBuffer.allocate(4);
  this.unwrappedData = null;
  this.unwrappedDataLengthBuffer = ByteBuffer.allocate(4);
  this.socket = channel.socket();
  this.addr = socket.getInetAddress();
  if (addr == null) {
    this.hostAddress = "*Unknown*";
  } else {
    this.hostAddress = addr.getHostAddress();
  }
  this.remotePort = socket.getPort();
  this.responseQueue = new LinkedList<Call>();
  if (socketSendBufferSize != 0) {
    try {
      socket.setSendBufferSize(socketSendBufferSize);
    } catch (IOException e) {
      LOG.warn("Connection: unable to set socket send buffer size to " +
               socketSendBufferSize);
    }
  }
}

protected void removeSocket(SocketChannel channel) {
    if (logger.isLoggingEnabled(LogWriter.TRACE_DEBUG)) {
    logger.logDebug("Trying to remove cached socketChannel without key"
            + this + " socketChannel = " + channel);
}
    LinkedList<String> keys = new LinkedList<String>();
    synchronized(socketTable) {
        Set<Entry<String, SocketChannel>> e = socketTable.entrySet();
        for(Entry<String, SocketChannel> entry : e ) {
            SocketChannel sc = entry.getValue();
            if(sc.equals(channel)) {
                keys.add(entry.getKey());
            }
        }
        for(String key : keys) {
            if (logger.isLoggingEnabled(LogWriter.TRACE_DEBUG)) {
                logger.logDebug("Removing cached socketChannel without key"
                        + this + " socketChannel = " + channel + " key = " + key);
            }
            removeSocket(key);
        }
    }
  }

public int
write(
    SocketChannel       chan,
    ByteBuffer          buffer )

    throws IOException
{
    int pos = buffer.position();

    int len = chan.write( buffer );

    if ( len > 0 ){

        buffer.position( pos );

        analyse( buffer, len );
    }

    return( len );
}

private void doWrite(SelectionKey sk){
    SocketChannel channel=(SocketChannel)sk.channel();
    EchoClient echoClient=(EchoClient)sk.attachment();
    LinkedList<ByteBuffer> outq=echoClient.getOutputQueue();

    ByteBuffer bb=outq.getLast();
    try {
        int len=channel.write(bb);
        if(len==-1){
            disconnect(sk);
            return;
        }
        if(bb.remaining()==0){
            outq.removeLast();
        }
    }catch (Exception e){
        e.printStackTrace();
        System.out.println("fail to write to client");
        disconnect(sk);
    }

    if(outq.size()==0){
        sk.interestOps(SelectionKey.OP_READ);
    }

}

@Override
void connect(InetSocketAddress addr) throws IOException {
    SocketChannel sock = createSock();
    try {
       registerAndConnect(sock, addr);
    } catch (IOException e) {
        LOG.error("Unable to open socket to " + addr);
        sock.close();
        throw e;
    }
    initialized = false;

    /*
     * Reset incomingBuffer
     */
    lenBuffer.clear();
    incomingBuffer = lenBuffer;
}

public Connection(SocketChannel channel) {
    this.channel = channel;
    this.isClosed = new AtomicBoolean(false);
    this.startupTime = TimeUtil.currentTimeMillis();
    this.lastReadTime = startupTime;
    this.lastWriteTime = startupTime;
    this.id = ConnectIdGenerator.getINSTNCE().getId();
}

public static boolean connectEthernet(String ipPort) {
    SocketChannel channel = openAdbConnection();
    try {
        return send(channel, CONNECT_TCP_ALL + ipPort);
    } finally {
        try {
            channel.close();
        } catch (IOException ex) {
        }
    }

}

public void connect(NioThread nt, SocketAddress address, byte[] thisId, byte[] otherId) throws Exception
{
    SocketChannel sc=SocketChannel.open();
    sc.configureBlocking(false);
    sc.connect(address);
    connect(nt, sc, true, thisId, otherId);
}

/**
 * Mock a client channel with a connection request and a watches message
 * inside.
 * 
 * @return a socket channel
 * @throws IOException
 */
private SocketChannel createClientSocketChannel() throws IOException {

    SocketChannel socketChannel = mock(SocketChannel.class);
    Socket socket = mock(Socket.class);
    InetSocketAddress socketAddress = new InetSocketAddress(1234);
    when(socket.getRemoteSocketAddress()).thenReturn(socketAddress);
    when(socketChannel.socket()).thenReturn(socket);

    // Send watches packet to server connection
    final ByteBuffer connRequest = createConnRequest();
    final ByteBuffer watchesMessage = createWatchesMessage();
    final ByteBuffer request = ByteBuffer.allocate(connRequest.limit()
            + watchesMessage.limit());
    request.put(connRequest);
    request.put(watchesMessage);

    Answer<Integer> answer = new Answer<Integer>() {
        int i = 0;

        @Override
        public Integer answer(InvocationOnMock invocation) throws Throwable {
            Object[] args = invocation.getArguments();
            ByteBuffer bb = (ByteBuffer) args[0];
            for (int k = 0; k < bb.limit(); k++) {
                bb.put(request.get(i));
                i = i + 1;
            }
            return bb.limit();
        }
    };
    when(socketChannel.read(any(ByteBuffer.class))).thenAnswer(answer);
    return socketChannel;
}

private void read(SocketChannel channel) throws Exception {
    LinkedList<Byte> list = new LinkedList<>();
    ByteBuffer buf = ByteBuffer.allocate(1024);
    int bytesRead = channel.read(buf);
    // 如果读取到-1，则说明客户端关闭了该链接
    if (bytesRead == -1) {
        log.info("Close channel {}", channel.getRemoteAddress());
        channel.close();
        return;
    }
    // 非阻塞IO可以读取0个字节，这种数据应该手动丢弃
    if (bytesRead == 0) return;

    // 读取所有的数据
    while (bytesRead > 0) {
        buf.flip();
        while (buf.hasRemaining()) {
            list.add(buf.get());
        }
        buf.clear();
        bytesRead = channel.read(buf);
    }
    String request = new String(Bytes.toArray(list), Constants.DEFAULT_ENCODING);
    try {
        // 写回响应
        response(request, channel);
    } catch (Exception e) {
        e.printStackTrace();
        // 返回错误信息
        StringWriter stringWriter = new StringWriter();
        PrintWriter printWriter = new PrintWriter(stringWriter);
        e.printStackTrace(printWriter);
        serverError(stringWriter.toString(), channel);
    }
}

private void read(SelectionKey key) throws Exception {
    // 服务器可读消息，得到事件发生的socket通道
    SocketChannel channel = (SocketChannel) key.channel();
    // 穿件读取的缓冲区
    ByteBuffer buffer = ByteBuffer.allocate(10);
    channel.read(buffer);
    byte[] data = buffer.array();
    String msg = new String(data).trim();
    System.out.println("server receive from client: " + msg);
    ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());
    channel.write(outBuffer);
}

/**
 * 进行初始化Selector和Channel
 */
public void init(String ip, int port) throws IOException {
    //创建一个SocketChannel实例，并设置为非阻塞模式
    SocketChannel channel = SocketChannel.open();
    channel.configureBlocking(false);
    //创建Selector
    this.selector = SelectorProvider.provider().openSelector();
    //将SocketChannel绑定到Socket上
    //由于当前Channel是非阻塞的，因此，connect()方法返回时，连接并不一定建立成功，在后续使用这个连接时，还需要使用finishConnect()再次确认
    channel.connect(new InetSocketAddress(ip, port));
    //将这个Channel和Selector进行绑定，并注册了感兴趣的事件作为连接（OP_CONNECT）
    channel.register(selector, SelectionKey.OP_CONNECT);
}

public static SocketChannel newSocketChannel(FileDescriptor fd) {
    try {
        return new SocketChannelImpl(provider(), fd, false);
    } catch (IOException ioe) {
        throw new AssertionError(ioe);
    }
}

/**
 * {@inheritDoc}
 */
@Override
protected ConnectionRequest getConnectionRequest(SocketChannel handle) {
    SelectionKey key = handle.keyFor(selector);

    if ((key == null) || (!key.isValid())) {
        return null;
    }

    return (ConnectionRequest) key.attachment();
}

private void send(SocketChannel channel, RCONPacket packet) {
    if (!channel.keyFor(this.selector).isValid()) {
        return;
    }

    synchronized (this.sendQueues) {
        List<RCONPacket> queue = sendQueues.computeIfAbsent(channel, k -> new ArrayList<>());
        queue.add(packet);
    }

    this.selector.wakeup();
}

public static boolean send(SocketChannel mAdbConnection, String message) {
    if (mAdbConnection != null) {
        byte[] request = AdbHelper.formAdbRequest(message);
        try {
            AdbHelper.write(mAdbConnection, request);
            AdbHelper.AdbResponse resp = AdbHelper.readAdbResponse(mAdbConnection, false);
            return resp.okay;

        } catch (Exception e) {
        }
    }
    return false;
}

SSLDelegate(SSLEngine eng, SocketChannel chan, HttpClientImpl client, String sn)
{
    this.engine = eng;
    this.chan = chan;
    this.client = client;
    this.wrapper = new EngineWrapper(chan, engine);
    this.sslParameters = engine.getSSLParameters();
    this.serverName = sn;
}

public MeviusClient(InetSocketAddress addr, MeviusHandler handler) throws IOException {
    this.sc = SocketChannel.open(addr);
    sc.configureBlocking(false);
    this.handler = handler;
    (el = new EventListener(sc)).start();
    uuid = UUID.randomUUID();
    self = true;
    KeyPair kp = MeviusCipherKey.randomRSAKeyPair(512).getKey();
    privatekey = kp.getPrivate();
    sc.write(convert(kp.getPublic()));
    handler.connection(ConnectionType.CLIENT_CONNECT_TO_SERVER, this);
}

private void handRead(SelectionKey key) throws IOException {
    SocketChannel sc = (SocketChannel) key.channel();
    ByteBuffer byteBuffer = ByteBuffer.allocate(1024); // 1k
    sc.read(byteBuffer);
    byteBuffer.flip();
    byte[] data = new byte[byteBuffer.limit()];
    byteBuffer.get(data);
    key.interestOps(SelectionKey.OP_WRITE);
    System.out.println("server read");
    System.out.println(new String(data, "UTF-8"));

}

void forwardPacket(DatagramPacket outPacket, IpPacket parsedPacket) throws DaedalusVpnService.VpnNetworkException {
    Socket dnsSocket;
    try {
        // Packets to be sent to the real DNS server will need to be protected from the VPN
        dnsSocket = SocketChannel.open().socket();

        service.protect(dnsSocket);

        SocketAddress address = new InetSocketAddress(outPacket.getAddress(), DNSServerHelper.getPortOrDefault(outPacket.getAddress(), outPacket.getPort()));
        dnsSocket.connect(address, 5000);
        dnsSocket.setSoTimeout(5000);
        Logger.info("TcpProvider: Sending DNS query request");
        DataOutputStream dos = new DataOutputStream(dnsSocket.getOutputStream());
        byte[] packet = processUdpPacket(outPacket, parsedPacket);
        dos.writeShort(packet.length);
        dos.write(packet);
        dos.flush();

        if (parsedPacket != null) {
            dnsIn.add(new TcpProvider.WaitingOnSocketPacket(dnsSocket, parsedPacket));
        } else {
            dnsSocket.close();
        }
    } catch (IOException e) {
        if (e.getCause() instanceof ErrnoException) {
            ErrnoException errnoExc = (ErrnoException) e.getCause();
            if ((errnoExc.errno == OsConstants.ENETUNREACH) || (errnoExc.errno == OsConstants.EPERM)) {
                throw new DaedalusVpnService.VpnNetworkException("Cannot send message:", e);
            }
        }
        Log.w(TAG, "handleDnsRequest: Could not send packet to upstream", e);
    }
}

public SocketChannelHelper(String address, int port)
    {
        this.address = address;
        this.port = port;

        try
        {
    this.channel = SocketChannel.open(new InetSocketAddress(address, port));
}
        catch (IOException e)
        {
    e.printStackTrace();
}
    }

public static void closeConnection(SocketChannel connection) throws InterruptedException, IOException {
    synchronized (m_executors) {
        ExecutorPair p = m_executors.remove(connection);
        assert(p != null);
        p.shutdown();
    }
    connection.close();
}

private void connect(Selector finalSelector) {
    AbstractConnection c;
    while ((c = connectQueue.poll()) != null) {
        try {
            SocketChannel channel = (SocketChannel) c.getChannel();
            channel.register(finalSelector, SelectionKey.OP_CONNECT, c);
            channel.connect(new InetSocketAddress(c.host, c.port));

        } catch (Exception e) {
            LOGGER.info("error:", e);
            c.close(e.toString());
        }
    }
}

/**
  * Close the given connection.
  * @param channel to close
  */
 public void
 closeConnection(
SocketChannel channel )
 {
  closeConnection( channel, 0 );
 }

private void pushRequest(Request request) throws IOException {

        // 如果尚未连接到服务器，则会创建一个连接
        SocketChannel socketChannel =
                socketChannelClientHandler.openSocketChannel(clientConfig.pickAddress());

        RequestContext requestContext = new RequestContext();
        requestContext.setSocketChannel(socketChannel);
        requestContext.setRequest(request);

        this.socketChannelClientHandler.push(requestContext);
    }

public void addWorker(final UUID fdtSessionID, final SocketChannel sc) throws Exception {
    final FDTSession fdtSession = fdtSessionMap.get(fdtSessionID);
    if (fdtSession != null) {
        fdtSession.transportProvider.addWorkerStream(sc, true);
    } else {
        logger.log(Level.WARNING, "\n\n [ FDTSessionManager ] No such session " + fdtSessionID + " for worker: "
                + sc + ". The channel will be closed");
        Utils.closeIgnoringExceptions(sc);
    }
}

@Override
public SocketChannel get(int retryNumber, Throwable lastError) {
    if(lastError != null) {
        logger.error(lastError.getMessage(), lastError);
    }
    try {
        return SocketChannel.open(new InetSocketAddress(host, port));
    } catch (IOException e) {
        throw new IllegalStateException(e.getMessage(), e);
    }
}