clientnet.java

短信开发用于文件交换处理转发的类模块

        }

        private int connect() {

            close();

            try {
                SocketChannel channel = SocketChannel.open();
                channel.configureBlocking(false);

                sk = channel.register(selector, SelectionKey.OP_CONNECT
                        | SelectionKey.OP_READ);

                channel.connect(socketAddress);

                if (selector.select(connectTimeout) > 0) {

                    Iterator readyItor = selector.selectedKeys().iterator();

                    while (readyItor.hasNext()) {

                        SelectionKey key = (SelectionKey) readyItor.next();
                        readyItor.remove();

                        if (key != sk) {
                            continue;
                        }

                        SocketChannel keyChannel = (SocketChannel) key
                                .channel();

                        if (key.isConnectable()) {

                            if (keyChannel.isConnectionPending()) {
                                keyChannel.finishConnect();
                            }
                            key.interestOps(SelectionKey.OP_READ);
                        } else {
                            if (log.isEnabledFor(Priority.ERROR)) {
                                log.error("unknown event");
                            }
                            return ClientNet.ERR_UNKNOWNEVENT;
                        }

                    }
                    return 0;
                } else {
                    if (log.isEnabledFor(Priority.ERROR)) {
                        log.error("connect timeout.");
                    }
                    return ClientNet.ERR_TIMEOUT;
                }
            } catch (Throwable err) {
                if (log.isEnabledFor(Priority.ERROR)) {
                    log.error(null, err);
                }
            }
            return ClientNet.ERR_TIMEOUT;
        }

        private void close() {
            if (sk != null) {
                try {
                    sk.channel().close();
                    sk.cancel();
                } catch (Throwable ex) {
                    if (log.isEnabledFor(Priority.ERROR)) {
                        log.error(null, ex);
                    }
                }
                sk = null;
            }
            using = false;
        }

        protected void finalize() {
            close();
            if (selector != null) {
                try {
                    selector.close();
                } catch (Throwable ex) {
                    if (log.isEnabledFor(org.apache.log4j.Priority.ERROR)) {
                        log.error(null, ex);
                    }
                }
                selector = null;
            }
        }

        public boolean isConnected() {
            return sk != null && sk.channel().isOpen();
        }

        public int queryEntity(EntityHead entity, int timeout) throws Exception {

            if (inPool || using || !this.isConnected()) {
                log.error("net is unavaiable.");
                return ClientNet.ERR_NOTCONNECTED;
            }

            int command = entity.getCommandID();

            using = true;
            buffer.clear();
            entity.writePackage(buffer);
            buffer.flip();

            try {
                while (buffer.remaining() > 0) {
                    ((SocketChannel) sk.channel()).write(buffer);
                }

                if (selector.select(timeout) > 0) {
                    Iterator readyItor = selector.selectedKeys().iterator();

                    while (readyItor.hasNext()) {

                        SelectionKey key = (SelectionKey) readyItor.next();

                        readyItor.remove();

                        if (key == sk && key.isReadable()) {
                            SocketChannel keyChannel = (SocketChannel) key
                                    .channel();
                            buffer.clear();

                            // long lasttime = System.currentTimeMillis();
                            // int datalen = 0;
                            // int len = 0;
                            // do {
                            // try {
                            // int temp = keyChannel.read(buffer);
                            // if(temp > 0){
                            // datalen += temp;
                            // }
                            // if(buffer.remaining() > 4 && len == 0){
                            // buffer.flip();
                            // len = buffer.getInt(buffer.position());
                            // buffer.compact();
                            // }
                            // log.debug("len = " + len + " datalen = " +
                            // datalen + " temp = " + temp + " time = " +
                            // (System.currentTimeMillis() - lasttime));
                            // } catch (Exception err) {
                            // close();
                            // using = false;
                            // return ERR_CONNECTFAIL;
                            // }
                            // } while (datalen < len ||
                            // System.currentTimeMillis() - lasttime >
                            // timeout);

                            int datalen = keyChannel.read(buffer);
                            buffer.flip();

                            boolean dataOK = false;
                            if (buffer.remaining() > 8) {
                                int len = buffer.getInt(0); // len
                                int cmd = buffer.getInt(4); // cmd

                                if ((len - 8) <= buffer.remaining()
                                        && cmd == (command | 0x80000000)) {
                                    entity.readPackage(buffer); // obj
                                    dataOK = true;
                                }
                            }
                            using = false;

                            if (!dataOK) {
                                close();
                            }

                            return dataOK ? 0 : ClientNet.ERR_RECVFAIL;
                        }
                    }
                    log.error("request unknow event");
                    close();
                    return ClientNet.ERR_UNKNOWNEVENT;
                } else {
                    log.error("request timeout");
                    close();
                    return ClientNet.ERR_TIMEOUT;
                }
            } catch (Throwable err) {
                log.error(null, err);
                close();
                return ClientNet.ERR_EXCEPTION;
            }
        }
    }

    public ClientNet() {
    }

    public ClientNet(String ip, int port, int timeout) {
        socketAddress = new InetSocketAddress(ip, port);
        connectTimeout = timeout;
    }

    static private Hashtable _connPool = new Hashtable();

    public innerNioSocket Connect() {
        innerNioSocket _socket;
        synchronized (_connPool) {
            ArrayList conns = (ArrayList) _connPool.get(this.socketAddress);
            if (conns == null || conns.size() == 0) {
                if (conns == null) {
                    conns = new ArrayList();
                    _connPool.put(socketAddress, conns);
                }
                innerNioSocket client = new innerNioSocket();
                client.connect();
                client.inPool = false;
                if (!client.isConnected()) {
                    conns.add(client);
                    client.inPool = true;
                }
                _socket = client.isConnected() ? client : null;
            } else {

                innerNioSocket client = (innerNioSocket) conns.remove(0);
                client.inPool = false;
                if (!client.isConnected()) {
                    client.connect();
                }
                if (!client.isConnected()) {
                    conns.add(client);
                    client.inPool = true;
                }
                _socket = client.isConnected() ? client : null;
            }
        }

        return _socket; // != null ? 0 : -1;
    }

    public void Disconnect(innerNioSocket _socket) {
        if (_socket != null && !_socket.inPool) {
            synchronized (_connPool) {
                ArrayList conns = (ArrayList) _connPool.get(socketAddress);
                _socket.inPool = true;
                conns.add(_socket);
                _socket = null;
            }
        }
    }

}