Java 类org.apache.catalina.tribes.ChannelMessage 实例源码

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    if ( access.addAndGet(1) == 1 ) txStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    try {
        super.sendMessage(destination, msg, payload);
    }catch ( ChannelException x ) {
        msgTxErr.addAndGet(1);
        if ( access.get() == 1 ) access.addAndGet(-1);
        throw x;
    } 
    mbTx += (bytes*destination.length)/(1024d*1024d);
    mbAppTx += bytes/(1024d*1024d);
    if ( access.addAndGet(-1) == 0 ) {
        long stop = System.currentTimeMillis();
        timeTx += (stop - txStart) / 1000d;
        if ((msgTxCnt.get() / interval) >= lastCnt) {
            lastCnt++;
            report(timeTx);
        }
    }
    msgTxCnt.addAndGet(1);
}

@Override
public void messageReceived(ChannelMessage msg) {
    if ( Arrays.contains(msg.getMessage().getBytesDirect(),0,COORD_ALIVE,0,COORD_ALIVE.length) ) {
        //ignore message, its an alive message
        fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_MSG_ARRIVE,this,"Alive Message"));

    } else if ( Arrays.contains(msg.getMessage().getBytesDirect(),0,COORD_HEADER,0,COORD_HEADER.length) ) {
        try {
            CoordinationMessage cmsg = new CoordinationMessage(msg.getMessage());
            Member[] cmbr = cmsg.getMembers();
            fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_MSG_ARRIVE,this,"Coord Msg Arrived("+Arrays.toNameString(cmbr)+")"));
            processCoordMessage(cmsg, msg.getAddress());
        }catch ( ChannelException x ) {
            log.error("Error processing coordination message. Could be fatal.",x);
        }
    } else {
        super.messageReceived(msg);
    }
}

@Override
public void messageReceived(ChannelMessage msg) {
    if ( !okToProcess(msg.getOptions()) ) {
        super.messageReceived(msg);
        return;
    }
    int msgnr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
    msg.getMessage().trim(4);
    MessageOrder order = new MessageOrder(msgnr,(ChannelMessage)msg.deepclone());
    try {
        inLock.writeLock().lock();
        if ( processIncoming(order) ) processLeftOvers(msg.getAddress(),false);
    }finally {
        inLock.writeLock().unlock();
    }
}

@Override
public void sendMessage(Member[] destination, ChannelMessage message) throws ChannelException {
    if ( !connected ) throw new ChannelException("Sender not connected.");
    ParallelNioSender sender = (ParallelNioSender)getSender();
    if (sender == null) {
        ChannelException cx = new ChannelException("Unable to retrieve a data sender, time out("+getMaxWait()+" ms) error.");
        for (int i = 0; i < destination.length; i++) cx.addFaultyMember(destination[i], new NullPointerException("Unable to retrieve a sender from the sender pool"));
        throw cx;
    } else {
        try {
            sender.sendMessage(destination, message);
            sender.keepalive();
        } catch (ChannelException x) {
            sender.disconnect();
            throw x;
        } finally {
            returnSender(sender);
            if (!connected) disconnect();
        }
    }
}

@Override
public void broadcast(ChannelMessage message) throws ChannelException {
    if (impl==null || (impl.startLevel & Channel.MBR_TX_SEQ)!=Channel.MBR_TX_SEQ )
        throw new ChannelException("Multicast send is not started or enabled.");

    byte[] data = XByteBuffer.createDataPackage((ChannelData)message);
    if (data.length>McastServiceImpl.MAX_PACKET_SIZE) {
        throw new ChannelException("Packet length["+data.length+"] exceeds max packet size of "+McastServiceImpl.MAX_PACKET_SIZE+" bytes.");
    }
    DatagramPacket packet = new DatagramPacket(data,0,data.length);
    try {
        impl.send(false, packet);
    } catch (Exception x) {
        throw new ChannelException(x);
    }
}

@Override
public void messageReceived(ChannelMessage msg) {
    if (!okToProcess(msg.getOptions())) {
        super.messageReceived(msg);
        return;
    }
    int msgnr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(), msg.getMessage().getLength() - 4);
    msg.getMessage().trim(4);
    MessageOrder order = new MessageOrder(msgnr, (ChannelMessage) msg.deepclone());
    try {
        inLock.writeLock().lock();
        if (processIncoming(order))
            processLeftOvers(msg.getAddress(), false);
    } finally {
        inLock.writeLock().unlock();
    }
}

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    if ( access.addAndGet(1) == 1 ) txStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    try {
        super.sendMessage(destination, msg, payload);
    }catch ( ChannelException x ) {
        msgTxErr.addAndGet(1);
        if ( access.get() == 1 ) access.addAndGet(-1);
        throw x;
    } 
    mbTx += (bytes*destination.length)/(1024d*1024d);
    mbAppTx += bytes/(1024d*1024d);
    if ( access.addAndGet(-1) == 0 ) {
        long stop = System.currentTimeMillis();
        timeTx += (stop - txStart) / 1000d;
        if ((msgTxCnt.get() / interval) >= lastCnt) {
            lastCnt++;
            report(timeTx);
        }
    }
    msgTxCnt.addAndGet(1);
}

@Override
public void messageReceived(ChannelMessage msg) {
    if ( Arrays.contains(msg.getMessage().getBytesDirect(),0,COORD_ALIVE,0,COORD_ALIVE.length) ) {
        //ignore message, its an alive message
        fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_MSG_ARRIVE,this,"Alive Message"));

    } else if ( Arrays.contains(msg.getMessage().getBytesDirect(),0,COORD_HEADER,0,COORD_HEADER.length) ) {
        try {
            CoordinationMessage cmsg = new CoordinationMessage(msg.getMessage());
            Member[] cmbr = cmsg.getMembers();
            fireInterceptorEvent(new CoordinationEvent(CoordinationEvent.EVT_MSG_ARRIVE,this,"Coord Msg Arrived("+Arrays.toNameString(cmbr)+")"));
            processCoordMessage(cmsg, msg.getAddress());
        }catch ( ChannelException x ) {
            log.error("Error processing coordination message. Could be fatal.",x);
        }
    } else {
        super.messageReceived(msg);
    }
}

@Override
public void messageReceived(ChannelMessage msg) {
    if ( !okToProcess(msg.getOptions()) ) {
        super.messageReceived(msg);
        return;
    }
    int msgnr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
    msg.getMessage().trim(4);
    MessageOrder order = new MessageOrder(msgnr,(ChannelMessage)msg.deepclone());
    try {
        inLock.writeLock().lock();
        if ( processIncoming(order) ) processLeftOvers(msg.getAddress(),false);
    }finally {
        inLock.writeLock().unlock();
    }
}

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
        throws ChannelException {
    try {
        super.sendMessage(destination, msg, payload);
    } catch (ChannelException cx) {
        FaultyMember[] mbrs = cx.getFaultyMembers();
        for (int i = 0; i < mbrs.length; i++) {
            if (mbrs[i].getCause() != null && (!(mbrs[i].getCause() instanceof RemoteProcessException))) {// RemoteProcessException's
                                                                                                            // are
                                                                                                            // ok
                this.memberDisappeared(mbrs[i].getMember());
            } // end if
        } // for
        throw cx;
    }
}

@Override
public void messageReceived(ChannelMessage msg) {
    // catch incoming
    boolean process = true;
    if (okToProcess(msg.getOptions())) {
        // check to see if it is a testMessage, if so, process = false
        process = ((msg.getMessage().getLength() != TCP_FAIL_DETECT.length)
                || (!Arrays.equals(TCP_FAIL_DETECT, msg.getMessage().getBytes())));
    } // end if

    // ignore the message, it doesnt have the flag set
    if (process)
        super.messageReceived(msg);
    else if (log.isDebugEnabled())
        log.debug("Received a failure detector packet:" + msg);
}

@Override
public void sendMessage(Member[] destination, ChannelMessage message) throws ChannelException {
    if (!connected)
        throw new ChannelException("Sender not connected.");
    ParallelNioSender sender = (ParallelNioSender) getSender();
    if (sender == null) {
        ChannelException cx = new ChannelException(
                "Unable to retrieve a data sender, time out(" + getMaxWait() + " ms) error.");
        for (int i = 0; i < destination.length; i++)
            cx.addFaultyMember(destination[i],
                    new NullPointerException("Unable to retrieve a sender from the sender pool"));
        throw cx;
    } else {
        try {
            sender.sendMessage(destination, message);
            sender.keepalive();
        } catch (ChannelException x) {
            sender.disconnect();
            throw x;
        } finally {
            returnSender(sender);
            if (!connected)
                disconnect();
        }
    }
}

@Override
public void broadcast(ChannelMessage message) throws ChannelException {
    if (impl == null || (impl.startLevel & Channel.MBR_TX_SEQ) != Channel.MBR_TX_SEQ)
        throw new ChannelException("Multicast send is not started or enabled.");

    byte[] data = XByteBuffer.createDataPackage((ChannelData) message);
    if (data.length > McastServiceImpl.MAX_PACKET_SIZE) {
        throw new ChannelException("Packet length[" + data.length + "] exceeds max packet size of "
                + McastServiceImpl.MAX_PACKET_SIZE + " bytes.");
    }
    DatagramPacket packet = new DatagramPacket(data, 0, data.length);
    try {
        impl.send(false, packet);
    } catch (Exception x) {
        throw new ChannelException(x);
    }
}

/**
 * Send a message to one or more members in the cluster
 * 
 * @param destination
 *            Member[] - the destinations, null or zero length means all
 * @param msg
 *            ClusterMessage - the message to send
 * @param payload
 *            TBA
 */
@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
        throws ChannelException {
    if (destination == null)
        destination = membershipService.getMembers();
    if ((msg.getOptions() & Channel.SEND_OPTIONS_MULTICAST) == Channel.SEND_OPTIONS_MULTICAST) {
        membershipService.broadcast(msg);
    } else {
        clusterSender.sendMessage(msg, destination);
    }
    if (Logs.MESSAGES.isTraceEnabled()) {
        Logs.MESSAGES.trace("ChannelCoordinator - Sent msg:" + new UniqueId(msg.getUniqueId()) + " at "
                + new java.sql.Timestamp(System.currentTimeMillis()) + " to " + Arrays.toNameString(destination));
    }
}

/**
 * Send a message to one or more members in the cluster
 * @param destination Member[] - the destinations, null or zero length means all
 * @param msg ClusterMessage - the message to send
 * @param payload TBA
 */
@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    if ( destination == null ) destination = membershipService.getMembers();
    if ((msg.getOptions()&Channel.SEND_OPTIONS_MULTICAST) == Channel.SEND_OPTIONS_MULTICAST) {
        membershipService.broadcast(msg);
    } else {
        clusterSender.sendMessage(msg,destination);
    }
    if ( Logs.MESSAGES.isTraceEnabled() ) {
        Logs.MESSAGES.trace("ChannelCoordinator - Sent msg:" + new UniqueId(msg.getUniqueId()) + " at " +new java.sql.Timestamp(System.currentTimeMillis())+ " to "+Arrays.toNameString(destination));
    }
}

@Override
public void messageReceived(ChannelMessage msg) {
    if ( Logs.MESSAGES.isTraceEnabled() ) {
        Logs.MESSAGES.trace("ChannelCoordinator - Received msg:" + new UniqueId(msg.getUniqueId()) + " at " +new java.sql.Timestamp(System.currentTimeMillis())+ " from "+msg.getAddress().getName());
    }
    super.messageReceived(msg);
}

@Override
public void messageReceived(ChannelMessage msg) {
    boolean isFrag = XByteBuffer.toBoolean(msg.getMessage().getBytesDirect(), msg.getMessage().getLength() - 1);
    msg.getMessage().trim(1);
    if (isFrag) {
        defrag(msg);
    } else {
        super.messageReceived(msg);
    }
}

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

}

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    try {
        super.sendMessage(destination, msg, payload);
    }catch ( ChannelException cx ) {
        FaultyMember[] mbrs = cx.getFaultyMembers();
        for ( int i=0; i<mbrs.length; i++ ) {
            if ( mbrs[i].getCause()!=null &&
                 (!(mbrs[i].getCause() instanceof RemoteProcessException)) ) {//RemoteProcessException's are ok
                this.memberDisappeared(mbrs[i].getMember());
            }//end if
        }//for
        throw cx;
    }
}

@Override
public void messageReceived(ChannelMessage msg) {
    //catch incoming
    boolean process = true;
    if ( okToProcess(msg.getOptions()) ) {
        //check to see if it is a testMessage, if so, process = false
        process = ( (msg.getMessage().getLength() != TCP_FAIL_DETECT.length) ||
                    (!Arrays.equals(TCP_FAIL_DETECT,msg.getMessage().getBytes()) ) );
    }//end if

    //ignore the message, it doesnt have the flag set
    if ( process ) super.messageReceived(msg);
    else if ( log.isDebugEnabled() ) log.debug("Received a failure detector packet:"+msg);
}

@Override
public void messageReceived(ChannelMessage msg) {
    //catch incoming 
    boolean process = true;
    if ( okToProcess(msg.getOptions()) ) {
        //check to see if it is a ping message, if so, process = false
        process = ( (msg.getMessage().getLength() != TCP_PING_DATA.length) ||
                    (!Arrays.equals(TCP_PING_DATA,msg.getMessage().getBytes()) ) );
    }//end if

    //ignore the message, it doesnt have the flag set
    if ( process ) super.messageReceived(msg);
    else if ( log.isDebugEnabled() ) log.debug("Received a TCP ping packet:"+msg);
}

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
        throws ChannelException {
    if (!okToProcess(msg.getOptions())) {
        super.sendMessage(destination, msg, payload);
        return;
    }
    ChannelException cx = null;
    for (int i = 0; i < destination.length; i++) {
        try {
            int nr = 0;
            try {
                outLock.writeLock().lock();
                nr = incCounter(destination[i]);
            } finally {
                outLock.writeLock().unlock();
            }
            // reduce byte copy
            msg.getMessage().append(nr);
            try {
                getNext().sendMessage(new Member[] { destination[i] }, msg, payload);
            } finally {
                msg.getMessage().trim(4);
            }
        } catch (ChannelException x) {
            if (cx == null)
                cx = x;
            cx.addFaultyMember(x.getFaultyMembers());
        }
    } // for
    if (cx != null)
        throw cx;
}

@Override
public void messageReceived(ChannelMessage msg) {
    boolean isFrag = XByteBuffer.toBoolean(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-1);
    msg.getMessage().trim(1);
    if ( isFrag ) {
        defrag(msg);
    } else {
        super.messageReceived(msg);
    }
}

public FragCollection getFragCollection(FragKey key, ChannelMessage msg) {
    FragCollection coll = fragpieces.get(key);
    if ( coll == null ) {
        synchronized (fragpieces) {
            coll = fragpieces.get(key);
            if ( coll == null ) {
                coll = new FragCollection(msg);
                fragpieces.put(key, coll);
            }
        }
    } 
    return coll;
}

public void frag(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    int size = msg.getMessage().getLength();

    int count = ((size / maxSize )+(size%maxSize==0?0:1));
    ChannelMessage[] messages = new ChannelMessage[count];
    int remaining = size;
    for ( int i=0; i<count; i++ ) {
        ChannelMessage tmp = (ChannelMessage)msg.clone();
        int offset = (i*maxSize);
        int length = Math.min(remaining,maxSize);
        tmp.getMessage().clear();
        tmp.getMessage().append(msg.getMessage().getBytesDirect(),offset,length);
        //add the msg nr
        //tmp.getMessage().append(XByteBuffer.toBytes(i),0,4);
        tmp.getMessage().append(i);
        //add the total nr of messages
        //tmp.getMessage().append(XByteBuffer.toBytes(count),0,4);
        tmp.getMessage().append(count);
        //add true as the frag flag
        //byte[] flag = XByteBuffer.toBytes(true);
        //tmp.getMessage().append(flag,0,flag.length);
        tmp.getMessage().append(true);
        messages[i] = tmp;
        remaining -= length;

    }
    for ( int i=0; i<messages.length; i++ ) {
        super.sendMessage(destination,messages[i],payload);
    }
}

public void addMessage(ChannelMessage msg) {
    //remove the total messages
    msg.getMessage().trim(4);
    //get the msg nr
    int nr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
    //remove the msg nr
    msg.getMessage().trim(4);
    frags[nr] = msg.getMessage();

}

public ChannelMessage assemble() {
    if ( !complete() ) throw new IllegalStateException("Fragments are missing.");
    int buffersize = 0;
    for (int i=0; i<frags.length; i++ ) buffersize += frags[i].getLength();
    XByteBuffer buf = new XByteBuffer(buffersize,false);
    msg.setMessage(buf);
    for ( int i=0; i<frags.length; i++ ) {
        msg.getMessage().append(frags[i].getBytesDirect(),0,frags[i].getLength());
    }
    return msg;
}

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    try {
        byte[] data = compress(msg.getMessage().getBytes());
        msg.getMessage().trim(msg.getMessage().getLength());
        msg.getMessage().append(data,0,data.length);
        getNext().sendMessage(destination, msg, payload);
    } catch ( IOException x ) {
        log.error("Unable to compress byte contents");
        throw new ChannelException(x);
    }
}

@Override
public void messageReceived(ChannelMessage msg) {
    try {
        byte[] data = decompress(msg.getMessage().getBytes());
        msg.getMessage().trim(msg.getMessage().getLength());
        msg.getMessage().append(data,0,data.length);
        getPrevious().messageReceived(msg);
    } catch ( IOException x ) {
        log.error("Unable to decompress byte contents",x);
    }
}

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


}

@Override
public void messageReceived(ChannelMessage msg) {
    try {
        byte[] data = decompress(msg.getMessage().getBytes());
        msg.getMessage().trim(msg.getMessage().getLength());
        msg.getMessage().append(data, 0, data.length);
        getPrevious().messageReceived(msg);
    } catch (IOException x) {
        log.error("Unable to decompress byte contents", x);
    }
}

@Override
public void messageReceived(ChannelMessage msg) {
    if (Arrays.equals(domain, msg.getAddress().getDomain())) {
        super.messageReceived(msg);
    } else {
        if (logCounter.incrementAndGet() >= logInterval) {
            logCounter.set(0);
            if (log.isWarnEnabled())
                log.warn("Received message from cluster[" + msg.getAddress() + "] was refused.");
        }
    }
}

@Override
public void messageReceived(ChannelMessage msg) {
    if (first) {
        first = false;
        start = System.currentTimeMillis();
    }
    mb += ( (double) len) / 1024 / 1024;
    synchronized (this) {count++;}
    if ( ( (count) % 10000) == 0) {
        long time = System.currentTimeMillis();
        seconds = ( (double) (time - start)) / 1000;
        System.out.println("Throughput " + df.format(mb / seconds) + " MB/seconds, messages "+count+", total "+mb+" MB.");
    }
}

/**
 * Send a message to one or more members in the cluster
 * @param destination Member[] - the destinations, null or zero length means all
 * @param msg ClusterMessage - the message to send
 * @param payload TBA
 */
@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    if ( destination == null ) destination = membershipService.getMembers();
    if ((msg.getOptions()&Channel.SEND_OPTIONS_MULTICAST) == Channel.SEND_OPTIONS_MULTICAST) {
        membershipService.broadcast(msg);
    } else {
        clusterSender.sendMessage(msg,destination);
    }
    if ( Logs.MESSAGES.isTraceEnabled() ) {
        Logs.MESSAGES.trace("ChannelCoordinator - Sent msg:" + new UniqueId(msg.getUniqueId()) + " at " +new java.sql.Timestamp(System.currentTimeMillis())+ " to "+Arrays.toNameString(destination));
    }
}

@Override
public void messageReceived(ChannelMessage msg) {
    if ( Logs.MESSAGES.isTraceEnabled() ) {
        Logs.MESSAGES.trace("ChannelCoordinator - Received msg:" + new UniqueId(msg.getUniqueId()) + " at " +new java.sql.Timestamp(System.currentTimeMillis())+ " from "+msg.getAddress().getName());
    }
    super.messageReceived(msg);
}

@Override
public void messageReceived(ChannelMessage msg) {
    if ( rxStart == 0 ) rxStart = System.currentTimeMillis();
    long bytes = XByteBuffer.getDataPackageLength(((ChannelData)msg).getDataPackageLength());
    mbRx += bytes/(1024d*1024d);
    msgRxCnt.addAndGet(1);
    if ( msgRxCnt.get() % interval == 0 ) report(timeTx);
    super.messageReceived(msg);

}

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    try {
        super.sendMessage(destination, msg, payload);
    }catch ( ChannelException cx ) {
        FaultyMember[] mbrs = cx.getFaultyMembers();
        for ( int i=0; i<mbrs.length; i++ ) {
            if ( mbrs[i].getCause()!=null &&
                 (!(mbrs[i].getCause() instanceof RemoteProcessException)) ) {//RemoteProcessException's are ok
                this.memberDisappeared(mbrs[i].getMember());
            }//end if
        }//for
        throw cx;
    }
}

@Override
public void messageReceived(ChannelMessage msg) {
    //catch incoming 
    boolean process = true;
    if ( okToProcess(msg.getOptions()) ) {
        //check to see if it is a ping message, if so, process = false
        process = ( (msg.getMessage().getLength() != TCP_PING_DATA.length) ||
                    (!Arrays.equals(TCP_PING_DATA,msg.getMessage().getBytes()) ) );
    }//end if

    //ignore the message, it doesnt have the flag set
    if ( process ) super.messageReceived(msg);
    else if ( log.isDebugEnabled() ) log.debug("Received a TCP ping packet:"+msg);
}

@Override
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
    if ( !okToProcess(msg.getOptions()) ) {
        super.sendMessage(destination, msg, payload);
        return;
    }
    ChannelException cx = null;
    for (int i=0; i<destination.length; i++ ) {
        try {
            int nr = 0;
            try {
                outLock.writeLock().lock();
                nr = incCounter(destination[i]);
            } finally {
                outLock.writeLock().unlock();
            }
            //reduce byte copy
            msg.getMessage().append(nr);
            try {
                getNext().sendMessage(new Member[] {destination[i]}, msg, payload);
            } finally {
                msg.getMessage().trim(4);
            }
        }catch ( ChannelException x ) {
            if ( cx == null ) cx = x;
            cx.addFaultyMember(x.getFaultyMembers());
        }
    }//for
    if ( cx != null ) throw cx;
}

@Override
public boolean addToQueue(ChannelMessage msg, Member[] destination, InterceptorPayload payload) {
    final LinkObject obj = new LinkObject(msg,destination,payload);
    Runnable r = new Runnable() {
        @Override
        public void run() {
            sendAsyncData(obj);
        }
    };
    executor.execute(r);
    return true;
}