import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
private void cancelKey(final SelectionKey key) {
    if ( log.isTraceEnabled() )
        log.trace("Adding key for cancel event:"+key);

    ObjectReader reader = (ObjectReader)key.attachment();
    if ( reader != null ) {
        reader.setCancelled(true);
        reader.finish();
    }
    Runnable cx = new Runnable() {
        @Override
        public void run() {
            if ( log.isTraceEnabled() )
                log.trace("Cancelling key:"+key);

            NioReceiver.cancelledKey(key);
        }
    };
    receiver.addEvent(cx);
}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
private void cancelKey(final SelectionKey key) {
	if (log.isTraceEnabled())
		log.trace("Adding key for cancel event:" + key);

	ObjectReader reader = (ObjectReader) key.attachment();
	if (reader != null) {
		reader.setCancelled(true);
		reader.finish();
	}
	Runnable cx = new Runnable() {
		@Override
		public void run() {
			if (log.isTraceEnabled())
				log.trace("Cancelling key:" + key);

			NioReceiver.cancelledKey(key);
		}
	};
	receiver.addEvent(cx);
}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
protected void registerForRead(final SelectionKey key, ObjectReader reader) {
    if ( log.isTraceEnabled() )
        log.trace("Adding key for read event:"+key);
    reader.finish();
    //register our OP_READ interest
    Runnable r = new Runnable() {
        @Override
        public void run() {
            try {
                if (key.isValid()) {
                    // cycle the selector so this key is active again
                    key.selector().wakeup();
                    // resume interest in OP_READ, OP_WRITE
                    int resumeOps = key.interestOps() | SelectionKey.OP_READ;
                    key.interestOps(resumeOps);
                    if ( log.isTraceEnabled() )
                        log.trace("Registering key for read:"+key);
                }
            } catch (CancelledKeyException ckx ) {
                NioReceiver.cancelledKey(key);
                if ( log.isTraceEnabled() )
                    log.trace("CKX Cancelling key:"+key);

            } catch (Exception x) {
                log.error("Error registering key for read:"+key,x);
            }
        }
    };
    receiver.addEvent(r);
}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
protected void socketTimeouts() {
        long now = System.currentTimeMillis();
        if ( (now-lastCheck) < getSelectorTimeout() ) return;
        //timeout
        Selector tmpsel = this.selector.get();
        Set<SelectionKey> keys =  (isListening()&&tmpsel!=null)?tmpsel.keys():null;
        if ( keys == null ) return;
        for (Iterator<SelectionKey> iter = keys.iterator(); iter.hasNext();) {
            SelectionKey key = iter.next();
            try {
//                if (key.interestOps() == SelectionKey.OP_READ) {
//                    //only timeout sockets that we are waiting for a read from
//                    ObjectReader ka = (ObjectReader) key.attachment();
//                    long delta = now - ka.getLastAccess();
//                    if (delta > (long) getTimeout()) {
//                        cancelledKey(key);
//                    }
//                }
//                else
                if ( key.interestOps() == 0 ) {
                    //check for keys that didn't make it in.
                    ObjectReader ka = (ObjectReader) key.attachment();
                    if ( ka != null ) {
                        long delta = now - ka.getLastAccess();
                        if (delta > getTimeout() && (!ka.isAccessed())) {
                            if (log.isWarnEnabled())
                                log.warn("Channel key is registered, but has had no interest ops for the last "+getTimeout()+" ms. (cancelled:"+ka.isCancelled()+"):"+key+" last access:"+new java.sql.Timestamp(ka.getLastAccess())+" Possible cause: all threads used, perform thread dump");
                            ka.setLastAccess(now);
                            //key.interestOps(SelectionKey.OP_READ);
                        }//end if
                    } else {
                        cancelledKey(key);
                    }//end if
                }//end if
            }catch ( CancelledKeyException ckx ) {
                cancelledKey(key);
            }
        }
        lastCheck = System.currentTimeMillis();
    }
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
public void listen() throws Exception {
    if (doListen()) {
        log.warn("ServerSocket already started");
        return;
    }
    setListen(true);

    while ( doListen() ) {
        Socket socket = null;
        if ( getTaskPool().available() < 1 ) {
            if ( log.isWarnEnabled() )
                log.warn("All BIO server replication threads are busy, unable to handle more requests until a thread is freed up.");
        }
        BioReplicationTask task = (BioReplicationTask)getTaskPool().getRxTask();
        if ( task == null ) continue; //should never happen
        try {
            socket = serverSocket.accept();
        }catch ( Exception x ) {
            if ( doListen() ) throw x;
        }
        if ( !doListen() ) {
            task.setDoRun(false);
            task.serviceSocket(null,null);
            getExecutor().execute(task);
            break; //regular shutdown
        }
        if ( socket == null ) continue;
        socket.setReceiveBufferSize(getRxBufSize());
        socket.setSendBufferSize(getTxBufSize());
        socket.setTcpNoDelay(getTcpNoDelay());
        socket.setKeepAlive(getSoKeepAlive());
        socket.setOOBInline(getOoBInline());
        socket.setReuseAddress(getSoReuseAddress());
        socket.setSoLinger(getSoLingerOn(),getSoLingerTime());
        socket.setSoTimeout(getTimeout());
        ObjectReader reader = new ObjectReader(socket);
        task.serviceSocket(socket,reader);
        getExecutor().execute(task);
    }//while
}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
protected void execute(ObjectReader reader) throws Exception{
    int pkgcnt = reader.count();

    if ( pkgcnt > 0 ) {
        ChannelMessage[] msgs = reader.execute();
        for ( int i=0; i<msgs.length; i++ ) {
            /**
             * Use send ack here if you want to ack the request to the remote 
             * server before completing the request
             * This is considered an asynchronized request
             */
            if (ChannelData.sendAckAsync(msgs[i].getOptions())) sendAck(Constants.ACK_COMMAND);
            try {
                //process the message
                getCallback().messageDataReceived(msgs[i]);
                /**
                 * Use send ack here if you want the request to complete on this
                 * server before sending the ack to the remote server
                 * This is considered a synchronized request
                 */
                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(Constants.ACK_COMMAND);
            }catch  ( Exception x ) {
                if (ChannelData.sendAckSync(msgs[i].getOptions())) sendAck(Constants.FAIL_ACK_COMMAND);
                log.error("Error thrown from messageDataReceived.",x);
            }
            if ( getUseBufferPool() ) {
                BufferPool.getBufferPool().returnBuffer(msgs[i].getMessage());
                msgs[i].setMessage(null);
            }
        }                       
    }

   
}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
/**
 * Called to initiate a unit of work by this worker thread on the provided
 * SelectionKey object. This method is synchronized, as is the run() method,
 * so only one key can be serviced at a given time. Before waking the worker
 * thread, and before returning to the main selection loop, this key's
 * interest set is updated to remove OP_READ. This will cause the selector
 * to ignore read-readiness for this channel while the worker thread is
 * servicing it.
 */
public synchronized void serviceChannel(SelectionKey key) {
	if (log.isTraceEnabled())
		log.trace("About to service key:" + key);
	ObjectReader reader = (ObjectReader) key.attachment();
	if (reader != null)
		reader.setLastAccess(System.currentTimeMillis());
	this.key = key;
	key.interestOps(key.interestOps() & (~SelectionKey.OP_READ));
	key.interestOps(key.interestOps() & (~SelectionKey.OP_WRITE));
}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
protected void registerForRead(final SelectionKey key, ObjectReader reader) {
	if (log.isTraceEnabled())
		log.trace("Adding key for read event:" + key);
	reader.finish();
	// register our OP_READ interest
	Runnable r = new Runnable() {
		@Override
		public void run() {
			try {
				if (key.isValid()) {
					// cycle the selector so this key is active again
					key.selector().wakeup();
					// resume interest in OP_READ, OP_WRITE
					int resumeOps = key.interestOps() | SelectionKey.OP_READ;
					key.interestOps(resumeOps);
					if (log.isTraceEnabled())
						log.trace("Registering key for read:" + key);
				}
			} catch (CancelledKeyException ckx) {
				NioReceiver.cancelledKey(key);
				if (log.isTraceEnabled())
					log.trace("CKX Cancelling key:" + key);

			} catch (Exception x) {
				log.error("Error registering key for read:" + key, x);
			}
		}
	};
	receiver.addEvent(r);
}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
protected void execute(ObjectReader reader) throws Exception {
	int pkgcnt = reader.count();

	if (pkgcnt > 0) {
		ChannelMessage[] msgs = reader.execute();
		for (int i = 0; i < msgs.length; i++) {
			/**
			 * Use send ack here if you want to ack the request to the
			 * remote server before completing the request This is
			 * considered an asynchronous request
			 */
			if (ChannelData.sendAckAsync(msgs[i].getOptions()))
				sendAck(Constants.ACK_COMMAND);
			try {
				// process the message
				getCallback().messageDataReceived(msgs[i]);
				/**
				 * Use send ack here if you want the request to complete on
				 * this server before sending the ack to the remote server
				 * This is considered a synchronized request
				 */
				if (ChannelData.sendAckSync(msgs[i].getOptions()))
					sendAck(Constants.ACK_COMMAND);
			} catch (Exception x) {
				if (ChannelData.sendAckSync(msgs[i].getOptions()))
					sendAck(Constants.FAIL_ACK_COMMAND);
				log.error("Error thrown from messageDataReceived.", x);
			}
			if (getUseBufferPool()) {
				BufferPool.getBufferPool().returnBuffer(msgs[i].getMessage());
				msgs[i].setMessage(null);
			}
		}
	}

}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
protected void socketTimeouts() {
        long now = System.currentTimeMillis();
        if ( (now-lastCheck) < getSelectorTimeout() ) return;
        //timeout
        Selector tmpsel = selector;
        Set<SelectionKey> keys =  (isListening()&&tmpsel!=null)?tmpsel.keys():null;
        if ( keys == null ) return;
        for (Iterator<SelectionKey> iter = keys.iterator(); iter.hasNext();) {
            SelectionKey key = iter.next();
            try {
//                if (key.interestOps() == SelectionKey.OP_READ) {
//                    //only timeout sockets that we are waiting for a read from
//                    ObjectReader ka = (ObjectReader) key.attachment();
//                    long delta = now - ka.getLastAccess();
//                    if (delta > (long) getTimeout()) {
//                        cancelledKey(key);
//                    }
//                }
//                else
                if ( key.interestOps() == 0 ) {
                    //check for keys that didn't make it in.
                    ObjectReader ka = (ObjectReader) key.attachment();
                    if ( ka != null ) {
                        long delta = now - ka.getLastAccess();
                        if (delta > getTimeout() && (!ka.isAccessed())) {
                            if (log.isWarnEnabled())
                                log.warn("Channel key is registered, but has had no interest ops for the last "+getTimeout()+" ms. (cancelled:"+ka.isCancelled()+"):"+key+" last access:"+new java.sql.Timestamp(ka.getLastAccess())+" Possible cause: all threads used, perform thread dump");
                            ka.setLastAccess(now);
                            //key.interestOps(SelectionKey.OP_READ);
                        }//end if
                    } else {
                        cancelledKey(key);
                    }//end if
                }//end if
            }catch ( CancelledKeyException ckx ) {
                cancelledKey(key);
            }
        }
        lastCheck = System.currentTimeMillis();
    }
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
@Override
public synchronized void run() {
    if ( buffer == null ) {
        int size = getRxBufSize();
        if (key.channel() instanceof DatagramChannel) {
            size = ChannelReceiver.MAX_UDP_SIZE;
        }
        if ( (getOptions() & OPTION_DIRECT_BUFFER) == OPTION_DIRECT_BUFFER) {
            buffer = ByteBuffer.allocateDirect(size);
        } else {
            buffer = ByteBuffer.allocate(size);
        }
    } else {
        buffer.clear();
    }
    if (key == null) {
        return; // just in case
    }
    if ( log.isTraceEnabled() )
        log.trace("Servicing key:"+key);

    try {
        ObjectReader reader = (ObjectReader)key.attachment();
        if ( reader == null ) {
            if ( log.isTraceEnabled() )
                log.trace("No object reader, cancelling:"+key);
            cancelKey(key);
        } else {
            if ( log.isTraceEnabled() )
                log.trace("Draining channel:"+key);

            drainChannel(key, reader);
        }
    } catch (Exception e) {
        //this is common, since the sockets on the other
        //end expire after a certain time.
        if ( e instanceof CancelledKeyException ) {
            //do nothing
        } else if ( e instanceof IOException ) {
            //dont spew out stack traces for IO exceptions unless debug is enabled.
            if (log.isDebugEnabled()) log.debug ("IOException in replication worker, unable to drain channel. Probable cause: Keep alive socket closed["+e.getMessage()+"].", e);
            else log.warn ("IOException in replication worker, unable to drain channel. Probable cause: Keep alive socket closed["+e.getMessage()+"].");
        } else if ( log.isErrorEnabled() ) {
            //this is a real error, log it.
            log.error("Exception caught in TcpReplicationThread.drainChannel.",e);
        }
        cancelKey(key);
    } finally {

    }
    key = null;
    // done, ready for more, return to pool
    getTaskPool().returnWorker (this);
}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
public synchronized void serviceSocket(Socket socket, ObjectReader reader) {
    this.socket = socket;
    this.reader = reader;
}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
@Override
public synchronized void run() {
	if (buffer == null) {
		int size = getRxBufSize();
		if (key.channel() instanceof DatagramChannel) {
			size = ChannelReceiver.MAX_UDP_SIZE;
		}
		if ((getOptions() & OPTION_DIRECT_BUFFER) == OPTION_DIRECT_BUFFER) {
			buffer = ByteBuffer.allocateDirect(size);
		} else {
			buffer = ByteBuffer.allocate(size);
		}
	} else {
		buffer.clear();
	}
	if (key == null) {
		return; // just in case
	}
	if (log.isTraceEnabled())
		log.trace("Servicing key:" + key);

	try {
		ObjectReader reader = (ObjectReader) key.attachment();
		if (reader == null) {
			if (log.isTraceEnabled())
				log.trace("No object reader, cancelling:" + key);
			cancelKey(key);
		} else {
			if (log.isTraceEnabled())
				log.trace("Draining channel:" + key);

			drainChannel(key, reader);
		}
	} catch (Exception e) {
		// this is common, since the sockets on the other
		// end expire after a certain time.
		if (e instanceof CancelledKeyException) {
			// do nothing
		} else if (e instanceof IOException) {
			// dont spew out stack traces for IO exceptions unless debug is
			// enabled.
			if (log.isDebugEnabled())
				log.debug(
						"IOException in replication worker, unable to drain channel. Probable cause: Keep alive socket closed["
								+ e.getMessage() + "].",
						e);
			else
				log.warn(
						"IOException in replication worker, unable to drain channel. Probable cause: Keep alive socket closed["
								+ e.getMessage() + "].");
		} else if (log.isErrorEnabled()) {
			// this is a real error, log it.
			log.error("Exception caught in TcpReplicationThread.drainChannel.", e);
		}
		cancelKey(key);
	} finally {

	}
	key = null;
	// done, ready for more, return to pool
	getTaskPool().returnWorker(this);
}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
protected void socketTimeouts() {
	long now = System.currentTimeMillis();
	if ((now - lastCheck) < getSelectorTimeout())
		return;
	// timeout
	Selector tmpsel = this.selector.get();
	Set<SelectionKey> keys = (isListening() && tmpsel != null) ? tmpsel.keys() : null;
	if (keys == null)
		return;
	for (Iterator<SelectionKey> iter = keys.iterator(); iter.hasNext();) {
		SelectionKey key = iter.next();
		try {
			// if (key.interestOps() == SelectionKey.OP_READ) {
			// //only timeout sockets that we are waiting for a read from
			// ObjectReader ka = (ObjectReader) key.attachment();
			// long delta = now - ka.getLastAccess();
			// if (delta > (long) getTimeout()) {
			// cancelledKey(key);
			// }
			// }
			// else
			if (key.interestOps() == 0) {
				// check for keys that didn't make it in.
				ObjectReader ka = (ObjectReader) key.attachment();
				if (ka != null) {
					long delta = now - ka.getLastAccess();
					if (delta > getTimeout() && (!ka.isAccessed())) {
						if (log.isWarnEnabled())
							log.warn("Channel key is registered, but has had no interest ops for the last "
									+ getTimeout() + " ms. (cancelled:" + ka.isCancelled() + "):" + key
									+ " last access:" + new java.sql.Timestamp(ka.getLastAccess())
									+ " Possible cause: all threads used, perform thread dump");
						ka.setLastAccess(now);
						// key.interestOps(SelectionKey.OP_READ);
					} // end if
				} else {
					cancelledKey(key);
				} // end if
			} // end if
		} catch (CancelledKeyException ckx) {
			cancelledKey(key);
		}
	}
	lastCheck = System.currentTimeMillis();
}
 

import org.apache.catalina.tribes.io.ObjectReader; //导入依赖的package包/类
public void listen() throws Exception {
	if (doListen()) {
		log.warn("ServerSocket already started");
		return;
	}
	setListen(true);

	while (doListen()) {
		Socket socket = null;
		if (getTaskPool().available() < 1) {
			if (log.isWarnEnabled())
				log.warn(
						"All BIO server replication threads are busy, unable to handle more requests until a thread is freed up.");
		}
		BioReplicationTask task = (BioReplicationTask) getTaskPool().getRxTask();
		if (task == null)
			continue; // should never happen
		try {
			socket = serverSocket.accept();
		} catch (Exception x) {
			if (doListen())
				throw x;
		}
		if (!doListen()) {
			task.setDoRun(false);
			task.serviceSocket(null, null);
			getExecutor().execute(task);
			break; // regular shutdown
		}
		if (socket == null)
			continue;
		socket.setReceiveBufferSize(getRxBufSize());
		socket.setSendBufferSize(getTxBufSize());
		socket.setTcpNoDelay(getTcpNoDelay());
		socket.setKeepAlive(getSoKeepAlive());
		socket.setOOBInline(getOoBInline());
		socket.setReuseAddress(getSoReuseAddress());
		socket.setSoLinger(getSoLingerOn(), getSoLingerTime());
		socket.setSoTimeout(getTimeout());
		ObjectReader reader = new ObjectReader(socket);
		task.serviceSocket(socket, reader);
		getExecutor().execute(task);
	} // while
}