init

java/org/apache/catalina/tribes/group/interceptors/DomainFilterInterceptor.java

@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.membership.Membership;
+import org.apache.catalina.tribes.util.StringManager;
+import org.apache.juli.logging.Log;
+import org.apache.juli.logging.LogFactory;
+
+/**
+ * <p>Title: Member domain filter interceptor </p>
+ *
+ * <p>Description: Filters membership based on domain.
+ * </p>
+ *
+ * @version 1.0
+ */
+public class DomainFilterInterceptor extends ChannelInterceptorBase
+        implements DomainFilterInterceptorMBean {
+
+    private static final Log log = LogFactory.getLog(DomainFilterInterceptor.class);
+    protected static final StringManager sm = StringManager.getManager(DomainFilterInterceptor.class);
+    protected volatile Membership membership = null;
+
+    protected byte[] domain = new byte[0];
+    protected int logInterval = 100;
+    private final AtomicInteger logCounter = new AtomicInteger(logInterval);
+
+    @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(sm.getString("domainFilterInterceptor.message.refused", msg.getAddress()));
+            }
+        }
+    }//messageReceived
+
+
+    @Override
+    public void memberAdded(Member member) {
+        if ( membership == null ) setupMembership();
+        boolean notify = false;
+        synchronized (membership) {
+            notify = Arrays.equals(domain,member.getDomain());
+            if ( notify ) notify = membership.memberAlive(member);
+        }
+        if ( notify ) {
+            super.memberAdded(member);
+        } else {
+            if(log.isInfoEnabled()) log.info(sm.getString("domainFilterInterceptor.member.refused", member));
+        }
+    }
+
+    @Override
+    public void memberDisappeared(Member member) {
+        if ( membership == null ) setupMembership();
+        boolean notify = false;
+        synchronized (membership) {
+            notify = Arrays.equals(domain,member.getDomain());
+            if ( notify ) membership.removeMember(member);
+        }
+        if ( notify ) super.memberDisappeared(member);
+    }
+
+    @Override
+    public boolean hasMembers() {
+        if ( membership == null ) setupMembership();
+        return membership.hasMembers();
+    }
+
+    @Override
+    public Member[] getMembers() {
+        if ( membership == null ) setupMembership();
+        return membership.getMembers();
+    }
+
+    @Override
+    public Member getMember(Member mbr) {
+        if ( membership == null ) setupMembership();
+        return membership.getMember(mbr);
+    }
+
+    @Override
+    public Member getLocalMember(boolean incAlive) {
+        return super.getLocalMember(incAlive);
+    }
+
+
+    protected synchronized void setupMembership() {
+        if ( membership == null ) {
+            membership = new Membership(super.getLocalMember(true));
+        }
+
+    }
+
+    @Override
+    public byte[] getDomain() {
+        return domain;
+    }
+
+    public void setDomain(byte[] domain) {
+        this.domain = domain;
+    }
+
+    public void setDomain(String domain) {
+        if ( domain == null ) return;
+        if (domain.startsWith("{"))
+            setDomain(org.apache.catalina.tribes.util.Arrays.fromString(domain));
+        else
+            setDomain(org.apache.catalina.tribes.util.Arrays.convert(domain));
+    }
+
+    @Override
+    public int getLogInterval() {
+        return logInterval;
+    }
+
+    @Override
+    public void setLogInterval(int logInterval) {
+        this.logInterval = logInterval;
+    }
+
+}

java/org/apache/catalina/tribes/group/interceptors/DomainFilterInterceptorMBean.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+public interface DomainFilterInterceptorMBean {
+
+    public int getOptionFlag();
+
+    public byte[] getDomain();
+
+    public int getLogInterval();
+
+    public void setLogInterval(int logInterval);
+
+}

java/org/apache/catalina/tribes/group/interceptors/EncryptInterceptorMBean.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+public interface EncryptInterceptorMBean {
+
+    // Config
+    public int getOptionFlag();
+    public void setOptionFlag(int optionFlag);
+
+    public void setEncryptionAlgorithm(String algorithm);
+    public String getEncryptionAlgorithm();
+    public void setEncryptionKey(byte[] key);
+    public byte[] getEncryptionKey();
+    public void setProviderName(String provider);
+    public String getProviderName();
+}

java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptor.java

@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Set;
+
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.group.InterceptorPayload;
+import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.util.StringManager;
+import org.apache.juli.logging.Log;
+import org.apache.juli.logging.LogFactory;
+
+/**
+ *
+ * The fragmentation interceptor splits up large messages into smaller messages and assembles them on the other end.
+ * This is very useful when you don't want large messages hogging the sending sockets
+ * and smaller messages can make it through.
+ *
+ * <br><b>Configuration Options</b><br>
+ * FragmentationInterceptor.expire=&lt;milliseconds&gt; - how long do we keep the fragments in memory and wait for the rest to arrive <b>default=60,000ms -&gt; 60seconds</b>
+ * This setting is useful to avoid OutOfMemoryErrors<br>
+ * FragmentationInterceptor.maxSize=&lt;max message size&gt; - message size in bytes <b>default=1024*100 (around a tenth of a MB)</b><br>
+ * @version 1.0
+ */
+public class FragmentationInterceptor extends ChannelInterceptorBase implements FragmentationInterceptorMBean {
+    private static final Log log = LogFactory.getLog(FragmentationInterceptor.class);
+    protected static final StringManager sm = StringManager.getManager(FragmentationInterceptor.class);
+
+    protected final HashMap<FragKey, FragCollection> fragpieces = new HashMap<>();
+    private int maxSize = 1024*100;
+    private long expire = 1000 * 60; //one minute expiration
+    protected final boolean deepclone = true;
+
+
+    @Override
+    public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
+        int size = msg.getMessage().getLength();
+        boolean frag = (size>maxSize) && okToProcess(msg.getOptions());
+        if ( frag ) {
+            frag(destination, msg, payload);
+        } else {
+            msg.getMessage().append(frag);
+            super.sendMessage(destination, msg, payload);
+        }
+    }
+
+    @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 removeFragCollection(FragKey key) {
+        fragpieces.remove(key);
+    }
+
+    public void defrag(ChannelMessage msg ) {
+        FragKey key = new FragKey(msg.getUniqueId());
+        FragCollection coll = getFragCollection(key,msg);
+        coll.addMessage((ChannelMessage)msg.deepclone());
+
+        if ( coll.complete() ) {
+            removeFragCollection(key);
+            ChannelMessage complete = coll.assemble();
+            super.messageReceived(complete);
+
+        }
+    }
+
+    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);
+        }
+    }
+
+    @Override
+    public void heartbeat() {
+        try {
+            Set<FragKey> set = fragpieces.keySet();
+            Object[] keys = set.toArray();
+            for ( int i=0; i<keys.length; i++ ) {
+                FragKey key = (FragKey)keys[i];
+                if ( key != null && key.expired(getExpire()) )
+                    removeFragCollection(key);
+            }
+        }catch ( Exception x ) {
+            if ( log.isErrorEnabled() ) {
+                log.error(sm.getString("fragmentationInterceptor.heartbeat.failed"),x);
+            }
+        }
+        super.heartbeat();
+    }
+
+    @Override
+    public int getMaxSize() {
+        return maxSize;
+    }
+
+    @Override
+    public long getExpire() {
+        return expire;
+    }
+
+    @Override
+    public void setMaxSize(int maxSize) {
+        this.maxSize = maxSize;
+    }
+
+    @Override
+    public void setExpire(long expire) {
+        this.expire = expire;
+    }
+
+    public static class FragCollection {
+        private final long received = System.currentTimeMillis();
+        private final ChannelMessage msg;
+        private final XByteBuffer[] frags;
+        public FragCollection(ChannelMessage msg) {
+            //get the total messages
+            int count = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
+            frags = new XByteBuffer[count];
+            this.msg = msg;
+        }
+
+        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 boolean complete() {
+            boolean result = true;
+            for ( int i=0; (i<frags.length) && (result); i++ ) result = (frags[i] != null);
+            return result;
+        }
+
+        public ChannelMessage assemble() {
+            if ( !complete() ) throw new IllegalStateException(sm.getString("fragmentationInterceptor.fragments.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;
+        }
+
+        public boolean expired(long expire) {
+            return (System.currentTimeMillis()-received)>expire;
+        }
+    }
+
+    public static class FragKey {
+        private final byte[] uniqueId;
+        private final long received = System.currentTimeMillis();
+        public FragKey(byte[] id ) {
+            this.uniqueId = id;
+        }
+        @Override
+        public int hashCode() {
+            return XByteBuffer.toInt(uniqueId,0);
+        }
+
+        @Override
+        public boolean equals(Object o ) {
+            if ( o instanceof FragKey ) {
+            return Arrays.equals(uniqueId,((FragKey)o).uniqueId);
+        } else return false;
+
+        }
+
+        public boolean expired(long expire) {
+            return (System.currentTimeMillis()-received)>expire;
+        }
+
+    }
+
+}

java/org/apache/catalina/tribes/group/interceptors/FragmentationInterceptorMBean.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+public interface FragmentationInterceptorMBean {
+
+    // Attributes
+    public int getMaxSize();
+
+    public long getExpire();
+
+    public void setMaxSize(int maxSize);
+
+    public void setExpire(long expire);
+}

java/org/apache/catalina/tribes/group/interceptors/GzipInterceptor.java

@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.group.InterceptorPayload;
+import org.apache.catalina.tribes.util.StringManager;
+import org.apache.juli.logging.Log;
+import org.apache.juli.logging.LogFactory;
+
+
+/**
+ * @version 1.0
+ */
+public class GzipInterceptor extends ChannelInterceptorBase {
+
+    private static final Log log = LogFactory.getLog(GzipInterceptor.class);
+    protected static final StringManager sm = StringManager.getManager(GzipInterceptor.class);
+
+    public static final int DEFAULT_BUFFER_SIZE = 2048;
+
+    @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);
+            super.sendMessage(destination, msg, payload);
+        } catch ( IOException x ) {
+            log.error(sm.getString("gzipInterceptor.compress.failed"));
+            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);
+            super.messageReceived(msg);
+        } catch ( IOException x ) {
+            log.error(sm.getString("gzipInterceptor.decompress.failed"),x);
+        }
+    }
+
+    public static byte[] compress(byte[] data) throws IOException {
+        ByteArrayOutputStream bout = new ByteArrayOutputStream();
+        GZIPOutputStream gout = new GZIPOutputStream(bout);
+        gout.write(data);
+        gout.flush();
+        gout.close();
+        return bout.toByteArray();
+    }
+
+    /**
+     * @param data  Data to decompress
+     * @return      Decompressed data
+     * @throws IOException Compression error
+     */
+    public static byte[] decompress(byte[] data) throws IOException {
+        ByteArrayOutputStream bout =
+            new ByteArrayOutputStream(DEFAULT_BUFFER_SIZE);
+        ByteArrayInputStream bin = new ByteArrayInputStream(data);
+        GZIPInputStream gin = new GZIPInputStream(bin);
+        byte[] tmp = new byte[DEFAULT_BUFFER_SIZE];
+        int length = gin.read(tmp);
+        while (length > -1) {
+            bout.write(tmp, 0, length);
+            length = gin.read(tmp);
+        }
+        return bout.toByteArray();
+    }
+}

java/org/apache/catalina/tribes/group/interceptors/LocalStrings.properties

@@ -0,0 +1,82 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+domainFilterInterceptor.member.refused=Member [{0}] was refused to join cluster
+domainFilterInterceptor.message.refused=Received message from cluster[{0}] was refused.
+
+encryptInterceptor.algorithm.required=Encryption algorithm is required, fully-specified e.g. AES/CBC/PKCS5Padding
+encryptInterceptor.algorithm.unsupported-mode=EncryptInterceptor does not support block cipher mode [{0}]
+encryptInterceptor.decrypt.error.short-message=Failed to decrypt message: premature end-of-message
+encryptInterceptor.decrypt.failed=Failed to decrypt message
+encryptInterceptor.encrypt.failed=Failed to encrypt message
+encryptInterceptor.init.failed=Failed to initialize EncryptInterceptor
+encryptInterceptor.key.required=Encryption key is required
+encryptInterceptor.tcpFailureDetector.ordering=EncryptInterceptor must be upstream of TcpFailureDetector. Please re-order EncryptInterceptor to be listed before TcpFailureDetector in your channel interceptor pipeline.
+
+fragmentationInterceptor.fragments.missing=Fragments are missing.
+fragmentationInterceptor.heartbeat.failed=Unable to perform heartbeat clean up in the frag interceptor
+
+gzipInterceptor.compress.failed=Unable to compress byte contents
+gzipInterceptor.decompress.failed=Unable to decompress byte contents
+
+messageDispatchInterceptor.AsyncMessage.failed=Error while processing async message.
+messageDispatchInterceptor.completeMessage.failed=Unable to report back completed message.
+messageDispatchInterceptor.errorMessage.failed=Unable to report back error message.
+messageDispatchInterceptor.queue.full=Asynchronous queue is full, reached its limit of [{0}] bytes, current:[{1}] bytes.
+messageDispatchInterceptor.unableAdd.queue=Unable to add the message to the async queue, queue bug?
+messageDispatchInterceptor.warning.optionflag=Warning, you are overriding the asynchronous option flag, this will disable the Channel.SEND_OPTIONS_ASYNCHRONOUS that other apps might use.
+
+nonBlockingCoordinator.electionMessage.sendfailed=Unable to send election message to:[{0}]
+nonBlockingCoordinator.heartbeat.failed=Unable to perform heartbeat.
+nonBlockingCoordinator.heartbeat.inconsistency=Heartbeat found inconsistency, restart election
+nonBlockingCoordinator.memberAdded.failed=Unable to start election when member was added.
+nonBlockingCoordinator.memberAlive.failed=Unable to perform member alive check, assuming member down.
+nonBlockingCoordinator.memberDisappeared.failed=Unable to start election when member was removed.
+nonBlockingCoordinator.processCoordinationMessage.failed=Error processing coordination message. Could be fatal.
+
+orderInterceptor.messageAdded.sameCounter=Message added has the same counter, synchronization bug. Disable the order interceptor
+
+staticMembershipInterceptor.no.failureDetector=There is no TcpFailureDetector. Automatic detection of static members does not work properly. By defining the StaticMembershipInterceptor under the TcpFailureDetector, automatic detection of the static members will work.
+staticMembershipInterceptor.no.pingInterceptor=There is no TcpPingInterceptor. The health check of static members does not work properly. By defining the TcpPingInterceptor, the health check of static members will work.
+staticMembershipInterceptor.sendLocalMember.failed=Local member notification failed.
+staticMembershipInterceptor.sendShutdown.failed=Shutdown notification failed.
+
+tcpFailureDetector.already.disappeared=Verification complete. Member already disappeared[{0}]
+tcpFailureDetector.failureDetection.failed=Unable to perform failure detection check, assuming member down.[{0}]
+tcpFailureDetector.heartbeat.failed=Unable to perform heartbeat on the TcpFailureDetector.
+tcpFailureDetector.member.disappeared=Verification complete. Member disappeared[{0}]
+tcpFailureDetector.memberDisappeared.verify=Received memberDisappeared[{0}] message. Will verify.
+tcpFailureDetector.performBasicCheck.memberAdded=Member added, even though we weren''t notified:[{0}]
+tcpFailureDetector.still.alive=Verification complete. Member still alive[{0}]
+tcpFailureDetector.suspectMember.alive=Suspect member, confirmed alive.[{0}]
+tcpFailureDetector.suspectMember.dead=Suspect member, confirmed dead.[{0}]
+
+tcpPingInterceptor.ping.failed=Unable to send TCP ping.
+tcpPingInterceptor.pingFailed.pingThread=Unable to send ping from TCP ping thread.
+
+throughputInterceptor.report=ThroughputInterceptor Report[\n\
+\tTx Msg:{0} messages\n\
+\tSent:{1} MB (total)\n\
+\tSent:{2} MB (application)\n\
+\tTime:{3} seconds\n\
+\tTx Speed:{4} MB/sec (total)\n\
+\tTx Speed:{5} MB/sec (application)\n\
+\tError Msg:{6}\n\
+\tRx Msg:{7} messages\n\
+\tRx Speed:{8} MB/sec (since 1st msg)\n\
+\tReceived:{9} MB]\n
+
+twoPhaseCommitInterceptor.heartbeat.failed=Unable to perform heartbeat on the TwoPhaseCommit interceptor.
+twoPhaseCommitInterceptor.originalMessage.missing=Received a confirmation, but original message is missing. Id:[{0}]

java/org/apache/catalina/tribes/group/interceptors/LocalStrings_de.properties

@@ -0,0 +1,32 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+domainFilterInterceptor.member.refused=Mitglied [{0}] wurde nicht zum Cluster zugelassen
+
+encryptInterceptor.decrypt.error.short-message=Konnte die Nachricht nicht entschlüsseln: Vorzeitiges Ende der Nachricht
+encryptInterceptor.decrypt.failed=Nachricht konnte nicht entschlüsselt werden
+
+messageDispatchInterceptor.queue.full=Asynchrone Warteschlange ist voll. Das Limit von [{0}] Bytes ist erreicht mit aktuell [{1}] Bytes.
+
+nonBlockingCoordinator.processCoordinationMessage.failed=Fehler beim Verarbeiten der Koordinationsnachricht. Könnte Fatal sein.
+
+staticMembershipInterceptor.sendShutdown.failed=Benachrichtigung über den Shutdown schlug fehl.
+
+tcpFailureDetector.failureDetection.failed=Überprüng zur Fehlererkennung fehlgeschlagen, Mitglied [{0}]
+tcpFailureDetector.still.alive=Verifikation abgeschlossen. Member sind immer noch am Leben [{0}]
+
+tcpPingInterceptor.ping.failed=Konnte kein TCP Ping senden.
+
+twoPhaseCommitInterceptor.heartbeat.failed=Kann den Heartbeat auf dem TwoPhaseCommit Interceptor nicht durchführen.

java/org/apache/catalina/tribes/group/interceptors/LocalStrings_es.properties

@@ -0,0 +1,47 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+domainFilterInterceptor.message.refused=El mensaje [{0}] recibido del cluster fue rechazado
+
+encryptInterceptor.decrypt.error.short-message=Fallo al descifrar el mensaje: fin-de-mensaje prematuro
+
+messageDispatchInterceptor.queue.full=La cola asincrónica esta llena, se alcanzó el limite de  [{0}] bytes, actualmente:[{1}] bytes.\n
+
+nonBlockingCoordinator.memberAlive.failed=No se puede verificar si el miembro esta vivo, asumiendo que el miembro esta inactivo.
+nonBlockingCoordinator.processCoordinationMessage.failed=Error procesando el mensaje de coordinación. Puede ser fatal.\n
+
+staticMembershipInterceptor.no.pingInterceptor=No existe TcpPingInterceptor. El verificador de estado de miembros estáticos no trabaja correctamente. Al definir el TcpPingInterceptor, el verificador de estado de miembros estáticos trabajará correctamente.
+staticMembershipInterceptor.sendShutdown.failed=El aviso de apagado falló.
+
+tcpFailureDetector.failureDetection.failed=No se pudo realizar la verificación de detección de fallos, se asume que el membro esta abajo.[{0}]
+tcpFailureDetector.heartbeat.failed=Imposible ejecutar heartbeat en el TcpFailureDetector.
+tcpFailureDetector.member.disappeared=Verificación completada. Miembro desaparecido[{0}]
+tcpFailureDetector.still.alive=Verificación completa. El miembro aun esta vivo [{0}]
+tcpFailureDetector.suspectMember.alive=Se confima que esta vivo el miembro.[{0}]\n
+
+tcpPingInterceptor.ping.failed=Imposible enviar ping TCP
+
+throughputInterceptor.report=ThroughputInterceptor Reporte[\n\
+\tTx Msg:{0} mensajes\n\
+\tEnviados:{2} MB (aplicación)\n\
+\tTiempo:{3} segundos\n\
+\tTx Speed:{4} MB/seg(total)\n\
+\tTx Speed::{5} MB/seg(aplicación)\n\
+\tMsg error:{6}\n\
+\tRx Msg:{7} mensajes\n\
+\tRx Speed:{8} MB/sec (desde 1er msg)\n\
+\tRecivido:{9} MB]
+
+twoPhaseCommitInterceptor.heartbeat.failed=Incapáz de ejecutar heartbeat en el interceptor TwoPhaseCommit

java/org/apache/catalina/tribes/group/interceptors/LocalStrings_fr.properties

@@ -0,0 +1,82 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+domainFilterInterceptor.member.refused=Le membre [{0}] a été refusé dans le cluster
+domainFilterInterceptor.message.refused=Le message reçu du cluster [{0}] a été refusé
+
+encryptInterceptor.algorithm.required=Un algorithme de cryptage est requis, avec une spécification complète telle que AES/CBC/PKCS5Padding
+encryptInterceptor.algorithm.unsupported-mode=L''EncryptInterceptor ne supporte pas le mode de chiffrage de bloc [{0}]
+encryptInterceptor.decrypt.error.short-message=Erreur de décryptage du message: fin de message prématuré
+encryptInterceptor.decrypt.failed=Echec de décryptage du message
+encryptInterceptor.encrypt.failed=Erreur de cryptage du message
+encryptInterceptor.init.failed=Echec de l'initalisation d'EncryptInterceptor
+encryptInterceptor.key.required=Une clé de cryptage est requise
+encryptInterceptor.tcpFailureDetector.ordering=EncryptInterceptor doit être en amont de TcpFailureDetector, l'EncryptInterceptor doit être repositionné pour être listé avant TcpFailureDetector dans le pipeline d'intercepteurs du canal
+
+fragmentationInterceptor.fragments.missing=Les fragments sont manquants
+fragmentationInterceptor.heartbeat.failed=Impossible d'effectuer le nettoyage périodique de l'intercepteur de fragments
+
+gzipInterceptor.compress.failed=Impossible de compresser un contenu binaire
+gzipInterceptor.decompress.failed=Impossible de décompresser le contenu des octets
+
+messageDispatchInterceptor.AsyncMessage.failed=Erreur lors du traitement du message asynchrone
+messageDispatchInterceptor.completeMessage.failed=Impossible de renvoyer le message complet
+messageDispatchInterceptor.errorMessage.failed=Impossible d'envoyer le message d'erreur
+messageDispatchInterceptor.queue.full=La file d''attente asynchrone est pleine, ayant atteint sa limite de [{0}] octets. Actuellement: [{1}] octets.
+messageDispatchInterceptor.unableAdd.queue=Impossible d'ajouter le message à la file asynchrone.  Bogue de file ?
+messageDispatchInterceptor.warning.optionflag=Attention, vous passez outre le drapeau d'option d'asynchronicité ("asynchronous option flag"), cela désactivera Channel.SEND_OPTIONS_ASYNCHRONOUS, que d'autres applications sont susceptibles d'utiliser.
+
+nonBlockingCoordinator.electionMessage.sendfailed=Impossible d''envoyer le message d''élection à: [{0}]
+nonBlockingCoordinator.heartbeat.failed=Impossible d'effectuer le signal périodique
+nonBlockingCoordinator.heartbeat.inconsistency=Le coordinateur à trouvé un état inconsistant, redémarrage de l'élection
+nonBlockingCoordinator.memberAdded.failed=Impossible de démarrer une élection quand le membre a été ajouté
+nonBlockingCoordinator.memberAlive.failed=Impossible d'effectuer le test de vie du membre, assume membre inactif.
+nonBlockingCoordinator.memberDisappeared.failed=Impossible de démarrer une élection lorsqu'un membre a été enlevé
+nonBlockingCoordinator.processCoordinationMessage.failed=Echec de traitement de message de coordination. Pourrait être fatal.
+
+orderInterceptor.messageAdded.sameCounter=Le message ajouté a le même compteur, à cause d'un bug de synchronisation, l'intercepteur d'ordre doit être désactivé
+
+staticMembershipInterceptor.no.failureDetector=Il n'y a pas de détecteur TcpFailureDetector. La détection automatique de membres statiques ne fonctionne pas correctement.  Par la définition d'un intercepteur StaticMembershipInterceptor sous le TcpFailureDetector, cette détection automatique fonctionnera.
+staticMembershipInterceptor.no.pingInterceptor=Il n'y a pas de TcpPingInterceptor. Le test de bonne santé des membres statiques ne fonctionne pas correctement. En définissant le TcpPingInterceptor, le test de bonne santé des membres statiques fonctionnera.
+staticMembershipInterceptor.sendLocalMember.failed=La notification du membre local a échouée
+staticMembershipInterceptor.sendShutdown.failed=La notification d'arrêt a échoué
+
+tcpFailureDetector.already.disappeared=La vérification est terminée, le membre avait déjà disparu [{0}]
+tcpFailureDetector.failureDetection.failed=Impossible d''effectuer le test de détection de faute.  Membre [{0}] supposé inactif.
+tcpFailureDetector.heartbeat.failed=Incapable de faire une pulsation ("heatbeat") sur le TcpFailureDector
+tcpFailureDetector.member.disappeared=La vérfication est complète, le membre a disparu [{0}]
+tcpFailureDetector.memberDisappeared.verify=Reçu un message memberDisappeared[{0}], qui sera vérifié
+tcpFailureDetector.performBasicCheck.memberAdded=Le membre a été ajouté bien qu''aucune notification n''ait été reçue: [{0}]
+tcpFailureDetector.still.alive=Vérification terminée. Le membre [{0}] vit toujours
+tcpFailureDetector.suspectMember.alive=Membre suspect, confirmé vivant.[{0}]
+tcpFailureDetector.suspectMember.dead=Un membre suspect a été confirmé mort [{0}]
+
+tcpPingInterceptor.ping.failed=Impossible d'envoyer un ping TCP.
+tcpPingInterceptor.pingFailed.pingThread=Impossible d'envoyer un ping à partir du thread des ping TCP
+
+throughputInterceptor.report=Rapport de l''intercepteur du débit ("ThroughputInterceptor Report") [\n\
+\tMsg Transmis (Tx Msg):{0} messages\n\
+\tEnvoyé (Sent):{1} MB (total)\n\
+\tEnvoyé (Sent):{2} MB (application)\n\
+\tDurée (Time):{3} secondes\n\
+\tVitesse d''écriture (Tx Speed):{4} MB/sec (total)\n\
+\tVitesse d''écriture (Tx Speed):{5} MB/sec (application)\n\
+\tMsg d''erreur (Error Msg):{6}\n\
+\tMsg Reçus (Rx Msg):{7} messages\n\
+\tVitesse de Réception (Rx Speed):{8} MB/sec (depuis le 1er message)\n\
+\tReçu:{9} MB]
+
+twoPhaseCommitInterceptor.heartbeat.failed=Impossible d'exécuter un battement de coeur (heartbeat) sur l'intercepteur (interceptor) "TwoPhaseCommit".
+twoPhaseCommitInterceptor.originalMessage.missing=Reçue une confirmation mais le message d''origine manque, id: [{0}]

java/org/apache/catalina/tribes/group/interceptors/LocalStrings_ja.properties

@@ -0,0 +1,82 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+domainFilterInterceptor.member.refused=メンバーはクラスター [{0}] への参加を拒否されました。
+domainFilterInterceptor.message.refused=クラスター [{0}] から受信したメッセージは拒否されました。
+
+encryptInterceptor.algorithm.required=暗号化アルゴリズムが必要です。完全指定。 AES/CBC/PKCS5Padding
+encryptInterceptor.algorithm.unsupported-mode=EncryptInterceptorはブロック暗号モード [{0}]をサポートしていません。
+encryptInterceptor.decrypt.error.short-message=メッセージの復号に失敗: メッセージの末尾が途切れています
+encryptInterceptor.decrypt.failed=メッセージの復号に失敗しました。
+encryptInterceptor.encrypt.failed=メッセージを暗号化できません。
+encryptInterceptor.init.failed=EncryptInterceptorの初期化に失敗しました
+encryptInterceptor.key.required=暗号化キーが必要です。
+encryptInterceptor.tcpFailureDetector.ordering=EncryptInterceptorはTcpFailureDetectorの上流になければなりません。 チャネルインターセプターパイプラインのTcpFailureDetectorの前にリストされるようにEncryptInterceptorを再設定してください。
+
+fragmentationInterceptor.fragments.missing=フラグメントが見つかりません。
+fragmentationInterceptor.heartbeat.failed=fragmentationInterceptorでハートビートクリーンアップを実行できません。
+
+gzipInterceptor.compress.failed=バイトデータを圧縮できません。
+gzipInterceptor.decompress.failed=圧縮されたバイトデータを展開できません。
+
+messageDispatchInterceptor.AsyncMessage.failed=非同期メッセージの処理中にエラーが発生しました。
+messageDispatchInterceptor.completeMessage.failed=完了したメッセージを報告できません。
+messageDispatchInterceptor.errorMessage.failed=エラーメッセージを返すことができません。
+messageDispatchInterceptor.queue.full=非同期キューが満杯です。現在は [{1}] バイトで上限の [{0}] バイトに達しています。
+messageDispatchInterceptor.unableAdd.queue=非同期キューにメッセージを登録できませんでした。キューの不具合かもしれません。
+messageDispatchInterceptor.warning.optionflag=警告です。非同期オプションフラグを上書きしたため、他のアプリケーションが使用する可能性のある Channel.SEND_OPTIONS_ASYNCHRONOUS は無効化されます。
+
+nonBlockingCoordinator.electionMessage.sendfailed=メンバー [{0}] に調停メッセージを送信できません。
+nonBlockingCoordinator.heartbeat.failed=ハートビートを実行できません。
+nonBlockingCoordinator.heartbeat.inconsistency=ハートビートが不一致を発見し、イレクションを再開します。
+nonBlockingCoordinator.memberAdded.failed=メンバーが追加されたときにイレクションを開始できません。
+nonBlockingCoordinator.memberAlive.failed=動作チェックが実行できなかったため、メンバーは停止しているものとして扱います。
+nonBlockingCoordinator.memberDisappeared.failed=メンバーが削除されたときにイレクションを開始できません。
+nonBlockingCoordinator.processCoordinationMessage.failed=調停メッセージを処理できませんでした。致命的な問題が発生している可能性があります。
+
+orderInterceptor.messageAdded.sameCounter=同じカウンタにメッセージが追加されました。同期バグがあります。 Order インターセプタを無効にして下さい。
+
+staticMembershipInterceptor.no.failureDetector=TcpFailureDetector がありません。静的メンバーの自動検出機能は正常に動作しません。TcpFailureDetector 配下に StaticMembershipInterceptor を定義すれば、静的メンバーの自動検出機能が動作するでしょう。
+staticMembershipInterceptor.no.pingInterceptor=TcpPingInterceptorが存在しないため、静的メンバーのヘルスチェックは正常に機能しません。TcpPingInterceptorを定義すれば機能するでしょう。
+staticMembershipInterceptor.sendLocalMember.failed=ローカルメンバーの通知は失敗しました。
+staticMembershipInterceptor.sendShutdown.failed=シャットダウン通知が失敗しました
+
+tcpFailureDetector.already.disappeared=検証完了。メンバーはすでに離脱していることを確認しました [{0}]
+tcpFailureDetector.failureDetection.failed=故障検出チェックが実行できないため、メンバーが停止しているものとして扱います。
+tcpFailureDetector.heartbeat.failed=TcpFailureDetector のハートビートチェックができませんでした。
+tcpFailureDetector.member.disappeared=メンバ検証が完了しました。 メンバーが消えました[{0}]
+tcpFailureDetector.memberDisappeared.verify=memberDisappeared[{0}]メッセージを受信しました。 メンバ検証します。
+tcpFailureDetector.performBasicCheck.memberAdded=私たちに通知されなかったにもかかわらず、メンバーが追加されました：[{0}]
+tcpFailureDetector.still.alive=故障検出チェックが完了しました。メンバー [{0}] は正常です。
+tcpFailureDetector.suspectMember.alive=疑わしいクラスタメンバーの生存を確認しました。 [{0}]
+tcpFailureDetector.suspectMember.dead=疑義メンバが死亡したことが確認されました。[{0}]
+
+tcpPingInterceptor.ping.failed=TCP の ping メッセージを送信できませんでした。
+tcpPingInterceptor.pingFailed.pingThread=TCP pingスレッドからpingを送信できません。
+
+throughputInterceptor.report=ThroughputInterceptor Report[\n\
+\        送信メッセージ (Tx Msg):{0} messages\n\
+\        送信済み (Sent):{1} MB (total)\n\
+\        送信済み (Sent):{2} MB (application)\n\
+\        時間 (Time):{3} seconds\n\
+\        送信速度 (Tx Speed):{4} MB/sec (total)\n\
+\        送信速度 (Tx Speed):{5} MB/sec (application)\n\
+\        エラーメッセージ (Error Msg):{6}\n\
+\        受信メッセージ (Rx Msg):{7} messages\n\
+\        受信速度 (Rx Speed):{8} MB/sec (since 1st msg)\n\
+\        受信済み (Received):{9} MB]
+
+twoPhaseCommitInterceptor.heartbeat.failed=TwoPhaseCommit インターセプターのハートビートが失敗しました。
+twoPhaseCommitInterceptor.originalMessage.missing=確認を受信しましたが、元のメッセージがありません。 Id：[{0}]

java/org/apache/catalina/tribes/group/interceptors/LocalStrings_ko.properties

@@ -0,0 +1,83 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+domainFilterInterceptor.member.refused=멤버 [{0}]이(가) 클러스터에 참가하는 것이 거부되었습니다.
+domainFilterInterceptor.message.refused=클러스터 [{0}](으)로부터 받은 메시지가 거부되었습니다.
+
+encryptInterceptor.algorithm.required=암호화 알고리즘을 완전하게 지정해야 합니다. 예) AES/CBC/PKCS5Padding.
+encryptInterceptor.algorithm.unsupported-mode=EncryptInterceptor가 블록 cipher 모드 [{0}]을(를) 지원하지 않습니다.
+encryptInterceptor.decrypt.error.short-message=메시지를 해독하지 못했습니다: 메시지가 너무 일찍 끝났습니다 (premature end-of-message).
+encryptInterceptor.decrypt.failed=메시지를 해독하지 못했습니다.
+encryptInterceptor.encrypt.failed=메시지를 암호화하지 못했습니다.
+encryptInterceptor.init.failed=EncryptInterceptor를 초기화하지 못했습니다.
+encryptInterceptor.key.required=암호화 키가 필수적입니다.
+encryptInterceptor.tcpFailureDetector.ordering=EncryptInterceptor는 반드시 TcpFailureDetector 보다 먼저 위치해야 합니다. 채널 인터셉터 파이프라인 내에서, EncryptInterceptor가 TcpFailureDetector 보다 먼저 위치하도록 조정하십시오.
+
+fragmentationInterceptor.fragments.missing=Fragment들이 없습니다.
+fragmentationInterceptor.heartbeat.failed=FragmentationInterceptor에서 heartbeat를 clean up 할 수 없습니다.
+
+gzipInterceptor.compress.failed=바이트 컨텐트들을 압축할 수 없습니다.
+gzipInterceptor.decompress.failed=바이트 컨텐트들의 압축을 풀 수 없습니다.
+
+messageDispatchInterceptor.AsyncMessage.failed=비동기 메시지를 처리하는 중 오류 발생
+messageDispatchInterceptor.completeMessage.failed=완료된 메시지를 되돌려 보고할 수 없습니다.
+messageDispatchInterceptor.errorMessage.failed=오류 메시지를 되돌려 보고할 수 없습니다.
+messageDispatchInterceptor.queue.full=비동기 큐가 꽉 차서 한계값인 [{0}] 바이트에 도달했습니다. 현재 값: [{1}] 바이트.
+messageDispatchInterceptor.unableAdd.queue=비동기 큐에 메시지를 추가할 수 없습니다. 큐의 버그일까요?
+messageDispatchInterceptor.warning.optionflag=경고: 귀하는 비동기 옵션 플래그를 오버라이드하고 있는데, 이는 다른 애플리케이션들이 사용할 수도 있는 Channel.SEND_OPTIONS_ASYNCHRONOUS 옵션을 사용 불능 상태로 만들 것입니다.
+
+nonBlockingCoordinator.electionMessage.sendfailed=Election 메시지를 [{0}]에 보낼 수 없습니다.
+nonBlockingCoordinator.heartbeat.failed=Heartbeat를 수행할 수 없습니다.
+nonBlockingCoordinator.heartbeat.inconsistency=Heartbeat가 일관되지 않은 상태로 발견되었습니다. Election을 다시 시작합니다.
+nonBlockingCoordinator.memberAdded.failed=멤버가 추가되었을 때, election을 시작할 수 없었습니다.
+nonBlockingCoordinator.memberAlive.failed=멤버가 살아있는지 점검할 수 없습니다. 아마도 해당 멤버가 다운된 것 같습니다.
+nonBlockingCoordinator.memberDisappeared.failed=멤버가 제거되었을 때, election을 시작할 수 없었습니다.
+nonBlockingCoordinator.processCoordinationMessage.failed=CoordinationMessage 처리 중 오류 발생. 치명적인 오류일 수 있습니다.
+
+orderInterceptor.messageAdded.sameCounter=추가된 메시지가 동일한 카운터를 가지고 있습니다. 동기화 결함입니다. OrderInterceptor를 사용불능 상태로 설정하십시오.
+
+staticMembershipInterceptor.no.failureDetector=TcpFailureDetector가 없습니다. 정적 멤버들에 대한 자동 탐지가 정상 동작하지 않을 것입니다. TcpFailureDetector 아래에 StaticMembershipInterceptor를 정의하게 되면, 정적 멤버들에 대한 자동 탐지가 정상 동작할 것입니다.
+staticMembershipInterceptor.no.pingInterceptor=TcpPingInterceptor가 존재하지 않습니다. 정적 멤버들에 대한 heath check는 제대로 동작하지 않을 것입니다. TcpPingInterceptor를 정의함으로써, 정적 멤버들에 대한 health check가 정상 동작할 것입니다.
+staticMembershipInterceptor.sendLocalMember.failed=로컬 멤버 통지 실패
+staticMembershipInterceptor.sendShutdown.failed=시스템을 셧다운하기 위한 통지가 실패했습니다.
+
+tcpFailureDetector.already.disappeared=검증 완료. 멤버가 이미 사라졌습니다: [{0}]
+tcpFailureDetector.failureDetection.failed=멤버에 대한 실패 탐지 점검을 수행할 수 없습니다. 아마도 해당 멤버 [{0}]이(가) 다운된 것 같습니다.
+tcpFailureDetector.heartbeat.failed=TcpFailureDetector에서 heartbeat 점검을 수행할 수 없습니다.
+tcpFailureDetector.member.disappeared=검증 완료. 멤버가 사라졌습니다: [{0}]
+tcpFailureDetector.memberDisappeared.verify=멤버 사라짐 메시지를 받았습니다: [{0}]. 이를 검증할 것입니다.
+tcpFailureDetector.performBasicCheck.memberAdded=통지 받지는 못했지만, 멤버가 추가되었습니다: [{0}]
+tcpFailureDetector.still.alive=검증 완료. 멤버가 아직 살아 있습니다: [{0}]
+tcpFailureDetector.suspectMember.alive=의심 멤버 서버가 살아 있음을 확인했습니다. [{0}]
+tcpFailureDetector.suspectMember.dead=의심 멤버가 다운된 것으로 확인됨: [{0}]
+
+tcpPingInterceptor.ping.failed=TCP ping을 보낼 수 없습니다.
+tcpPingInterceptor.pingFailed.pingThread=TCP ping 쓰레드로부터, ping을 전송할 수 없습니다.
+
+throughputInterceptor.report=ThroughputInterceptor의 보고 [\n\
+\tTx Msg:{0} 메시지(들)\n\
+\tSent:{1} MB (전체)\n\
+\tSent:{2} MB (애플리케이션)\n\
+\tTime:{3} 초\n\
+\tTx Speed:{4} MB/sec (전체)\n\
+\tTx Speed:{5} MB/sec (애플리케이션)\n\
+\tError Msg:{6}\n\
+\tRx Msg:{7} 메시지\n\
+\tRx Speed:{8} MB/sec (첫번째 메시지 이후로)\n\
+\tReceived:{9} MB]\n\
+\n
+
+twoPhaseCommitInterceptor.heartbeat.failed=해당 TwoPhaseCommit 인터셉터에서 heartbeat를 수행할 수 없습니다.
+twoPhaseCommitInterceptor.originalMessage.missing=확인 플래그를 받았지만, 원본 메시지가 없습니다. ID:[{0}]

java/org/apache/catalina/tribes/group/interceptors/LocalStrings_ru.properties

@@ -0,0 +1,22 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+encryptInterceptor.decrypt.error.short-message=Невозможно расшифровать сообщение: слишком мало символов
+
+nonBlockingCoordinator.memberAlive.failed=Невозможно проверить участника, считаем что упал
+
+staticMembershipInterceptor.sendShutdown.failed=Не удалось сообщить об отключении.
+
+tcpFailureDetector.still.alive=Проверка завершена. Участник ещё жив [{0}]

java/org/apache/catalina/tribes/group/interceptors/LocalStrings_zh_CN.properties

@@ -0,0 +1,60 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+domainFilterInterceptor.member.refused=成员被拒绝加入集群 cluster[{0}]
+domainFilterInterceptor.message.refused=从集群[{0}]中接收的消息被拒绝
+
+encryptInterceptor.decrypt.error.short-message=解密消息失败: 结尾消息提前结束
+encryptInterceptor.decrypt.failed=无法解密信息
+encryptInterceptor.encrypt.failed=无法加密信息
+encryptInterceptor.init.failed=初始化EncryptInterceptor失败
+encryptInterceptor.tcpFailureDetector.ordering=加密拦截器必须位于TCP故障检测器的上游。请重新订购加密拦截器，将其列在通道拦截器管道中的TCP故障检测器之前。
+
+messageDispatchInterceptor.errorMessage.failed=无法回传错误信息
+messageDispatchInterceptor.queue.full=异步队列已满，达到 [{0}] 字节的限制，当前：[{1}] 字节
+messageDispatchInterceptor.unableAdd.queue=无法将消息添加到异步队列，队列 bug？
+messageDispatchInterceptor.warning.optionflag=警告！你正在覆盖异步选项标志，这将禁用其它程序可能用到的 Channel.SEND_OPTIONS_ASYNCHRONOUS。
+
+nonBlockingCoordinator.heartbeat.inconsistency=心跳发现不一致，重新启动选举
+nonBlockingCoordinator.memberAlive.failed=无法执行成员活动检查，猜测成员下线。
+nonBlockingCoordinator.processCoordinationMessage.failed=处理协调消息时出错。 可能是致命的。
+
+staticMembershipInterceptor.no.failureDetector=没有TcpFailureDetector。 自动检测静态成员无法正常工作。 通过在TcpFailureDetector下定义StaticMembershipInterceptor，可以自动检测静态成员。
+staticMembershipInterceptor.no.pingInterceptor=在没有TcpPingInterceptor的情况下，静态成员的健康检查不会正常工作。只有定义了TcpPingInterceptor，才能使健康检查正常进行。
+staticMembershipInterceptor.sendShutdown.failed=关闭通知失败。
+
+tcpFailureDetector.failureDetection.failed=无法进行失败监测，假定成员宕机。[{0}]
+tcpFailureDetector.heartbeat.failed=TCP心跳检测器无法执行心跳
+tcpFailureDetector.member.disappeared=认证完成。成员消失[{0}]
+tcpFailureDetector.memberDisappeared.verify=(:收到的membermissed[{0}]消息。将验证。
+tcpFailureDetector.still.alive=验证完成。成员 [{0}] 仍然存活
+tcpFailureDetector.suspectMember.alive=验证可疑成员服务器还活着。[{0}]
+
+tcpPingInterceptor.ping.failed=无法发送 TCP ping
+tcpPingInterceptor.pingFailed.pingThread=不能从ping 线程发送ping
+
+throughputInterceptor.report=吞吐量拦截器 报告[\n\
+\   传输消息: {0} 消息数.\n\
+\   发送: {1} MB(总共)\n\
+\   发送: {2} MB (应用)\n\
+\   耗时: {3} 秒\n\
+\   传输速率: {4}MB/sec (总共)\n\
+\   传输速率: {5}MB/sec (应用)\n\
+\  错误消息: {6}\n\
+\  接收消息: {7} 消息数\n\
+\  接收速率: {8} MB/sec (从第一个消息开始)\n\
+\  收到: {9}MB]
+
+twoPhaseCommitInterceptor.heartbeat.failed=无法在两阶段提交拦截器上执行心跳。

java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java

@@ -0,0 +1,335 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.ErrorHandler;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.UniqueId;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.group.InterceptorPayload;
+import org.apache.catalina.tribes.util.ExecutorFactory;
+import org.apache.catalina.tribes.util.StringManager;
+import org.apache.catalina.tribes.util.TcclThreadFactory;
+import org.apache.juli.logging.Log;
+import org.apache.juli.logging.LogFactory;
+
+/**
+ * The message dispatcher is a way to enable asynchronous communication
+ * through a channel. The dispatcher will look for the
+ * <code>Channel.SEND_OPTIONS_ASYNCHRONOUS</code> flag to be set, if it is, it
+ * will queue the message for delivery and immediately return to the sender.
+ */
+public class MessageDispatchInterceptor extends ChannelInterceptorBase
+        implements MessageDispatchInterceptorMBean {
+
+    private static final Log log = LogFactory.getLog(MessageDispatchInterceptor.class);
+    protected static final StringManager sm =
+            StringManager.getManager(MessageDispatchInterceptor.class);
+
+    protected long maxQueueSize = 1024*1024*64; //64MB
+    protected volatile boolean run = false;
+    protected boolean useDeepClone = true;
+    protected boolean alwaysSend = true;
+
+    protected final AtomicLong currentSize = new AtomicLong(0);
+    protected ExecutorService executor = null;
+    protected int maxThreads = 10;
+    protected int maxSpareThreads = 2;
+    protected long keepAliveTime = 5000;
+
+
+    public MessageDispatchInterceptor() {
+        setOptionFlag(Channel.SEND_OPTIONS_ASYNCHRONOUS);
+    }
+
+
+    @Override
+    public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload)
+            throws ChannelException {
+        boolean async = (msg.getOptions() &
+                Channel.SEND_OPTIONS_ASYNCHRONOUS) == Channel.SEND_OPTIONS_ASYNCHRONOUS;
+        if (async && run) {
+            if ((getCurrentSize()+msg.getMessage().getLength()) > maxQueueSize) {
+                if (alwaysSend) {
+                    super.sendMessage(destination,msg,payload);
+                    return;
+                } else {
+                    throw new ChannelException(sm.getString("messageDispatchInterceptor.queue.full",
+                            Long.toString(maxQueueSize), Long.toString(getCurrentSize())));
+                }
+            }
+            //add to queue
+            if (useDeepClone) {
+                msg = (ChannelMessage)msg.deepclone();
+            }
+            if (!addToQueue(msg, destination, payload)) {
+                throw new ChannelException(
+                        sm.getString("messageDispatchInterceptor.unableAdd.queue"));
+            }
+            addAndGetCurrentSize(msg.getMessage().getLength());
+        } else {
+            super.sendMessage(destination, msg, payload);
+        }
+    }
+
+
+    public boolean addToQueue(final ChannelMessage msg, final Member[] destination,
+            final InterceptorPayload payload) {
+        Runnable r = new Runnable() {
+            @Override
+            public void run() {
+                sendAsyncData(msg, destination, payload);
+            }
+        };
+        executor.execute(r);
+        return true;
+    }
+
+
+    public void startQueue() {
+        if (run) {
+            return;
+        }
+        String channelName = "";
+        if (getChannel().getName() != null) channelName = "[" + getChannel().getName() + "]";
+        executor = ExecutorFactory.newThreadPool(maxSpareThreads, maxThreads, keepAliveTime,
+                TimeUnit.MILLISECONDS,
+                new TcclThreadFactory("MessageDispatchInterceptor.MessageDispatchThread" + channelName));
+        run = true;
+    }
+
+
+    public void stopQueue() {
+        run = false;
+        executor.shutdownNow();
+        setAndGetCurrentSize(0);
+    }
+
+
+    @Override
+    public void setOptionFlag(int flag) {
+        if ( flag != Channel.SEND_OPTIONS_ASYNCHRONOUS ) {
+            log.warn(sm.getString("messageDispatchInterceptor.warning.optionflag"));
+        }
+        super.setOptionFlag(flag);
+    }
+
+
+    public void setMaxQueueSize(long maxQueueSize) {
+        this.maxQueueSize = maxQueueSize;
+    }
+
+
+    public void setUseDeepClone(boolean useDeepClone) {
+        this.useDeepClone = useDeepClone;
+    }
+
+    @Override
+    public long getMaxQueueSize() {
+        return maxQueueSize;
+    }
+
+
+    public boolean getUseDeepClone() {
+        return useDeepClone;
+    }
+
+    @Override
+    public long getCurrentSize() {
+        return currentSize.get();
+    }
+
+
+    public long addAndGetCurrentSize(long inc) {
+        return currentSize.addAndGet(inc);
+    }
+
+
+    public long setAndGetCurrentSize(long value) {
+        currentSize.set(value);
+        return value;
+    }
+
+    @Override
+    public long getKeepAliveTime() {
+        return keepAliveTime;
+    }
+
+    @Override
+    public int getMaxSpareThreads() {
+        return maxSpareThreads;
+    }
+
+    @Override
+    public int getMaxThreads() {
+        return maxThreads;
+    }
+
+
+    public void setKeepAliveTime(long keepAliveTime) {
+        this.keepAliveTime = keepAliveTime;
+    }
+
+
+    public void setMaxSpareThreads(int maxSpareThreads) {
+        this.maxSpareThreads = maxSpareThreads;
+    }
+
+
+    public void setMaxThreads(int maxThreads) {
+        this.maxThreads = maxThreads;
+    }
+
+    @Override
+    public boolean isAlwaysSend() {
+        return alwaysSend;
+    }
+
+    @Override
+    public void setAlwaysSend(boolean alwaysSend) {
+        this.alwaysSend = alwaysSend;
+    }
+
+
+    @Override
+    public void start(int svc) throws ChannelException {
+        //start the thread
+        if (!run ) {
+            synchronized (this) {
+                // only start with the sender
+                if ( !run && ((svc & Channel.SND_TX_SEQ)==Channel.SND_TX_SEQ) ) {
+                    startQueue();
+                }
+            }
+        }
+        super.start(svc);
+    }
+
+
+    @Override
+    public void stop(int svc) throws ChannelException {
+        //stop the thread
+        if (run) {
+            synchronized (this) {
+                if ( run && ((svc & Channel.SND_TX_SEQ)==Channel.SND_TX_SEQ)) {
+                    stopQueue();
+                }
+            }
+        }
+
+        super.stop(svc);
+    }
+
+
+    protected void sendAsyncData(ChannelMessage msg, Member[] destination,
+            InterceptorPayload payload) {
+        ErrorHandler handler = null;
+        if (payload != null) {
+            handler = payload.getErrorHandler();
+        }
+        try {
+            super.sendMessage(destination, msg, null);
+            try {
+                if (handler != null) {
+                    handler.handleCompletion(new UniqueId(msg.getUniqueId()));
+                }
+            } catch ( Exception ex ) {
+                log.error(sm.getString("messageDispatchInterceptor.completeMessage.failed"),ex);
+            }
+        } catch ( Exception x ) {
+            ChannelException cx = null;
+            if (x instanceof ChannelException) {
+                cx = (ChannelException) x;
+            } else {
+                cx = new ChannelException(x);
+            }
+            if (log.isDebugEnabled()) {
+                log.debug(sm.getString("messageDispatchInterceptor.AsyncMessage.failed"),x);
+            }
+            try {
+                if (handler != null) {
+                    handler.handleError(cx, new UniqueId(msg.getUniqueId()));
+                }
+            } catch ( Exception ex ) {
+                log.error(sm.getString("messageDispatchInterceptor.errorMessage.failed"),ex);
+            }
+        } finally {
+            addAndGetCurrentSize(-msg.getMessage().getLength());
+        }
+    }
+
+    // ---------------------------------------------- stats of the thread pool
+    /**
+     * Return the current number of threads that are managed by the pool.
+     * @return the current number of threads that are managed by the pool
+     */
+    @Override
+    public int getPoolSize() {
+        if (executor instanceof ThreadPoolExecutor) {
+            return ((ThreadPoolExecutor) executor).getPoolSize();
+        } else {
+            return -1;
+        }
+    }
+
+    /**
+     * Return the current number of threads that are in use.
+     * @return the current number of threads that are in use
+     */
+    @Override
+    public int getActiveCount() {
+        if (executor instanceof ThreadPoolExecutor) {
+            return ((ThreadPoolExecutor) executor).getActiveCount();
+        } else {
+            return -1;
+        }
+    }
+
+    /**
+     * Return the total number of tasks that have ever been scheduled for execution by the pool.
+     * @return the total number of tasks that have ever been scheduled for execution by the pool
+     */
+    @Override
+    public long getTaskCount() {
+        if (executor instanceof ThreadPoolExecutor) {
+            return ((ThreadPoolExecutor) executor).getTaskCount();
+        } else {
+            return -1;
+        }
+    }
+
+    /**
+     * Return the total number of tasks that have completed execution by the pool.
+     * @return the total number of tasks that have completed execution by the pool
+     */
+    @Override
+    public long getCompletedTaskCount() {
+        if (executor instanceof ThreadPoolExecutor) {
+            return ((ThreadPoolExecutor) executor).getCompletedTaskCount();
+        } else {
+            return -1;
+        }
+    }
+}

java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptorMBean.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+public interface MessageDispatchInterceptorMBean {
+
+    public int getOptionFlag();
+
+    public boolean isAlwaysSend();
+
+    public void setAlwaysSend(boolean alwaysSend);
+
+    public long getMaxQueueSize();
+
+    public long getCurrentSize();
+
+    public long getKeepAliveTime();
+
+    public int getMaxSpareThreads();
+
+    public int getMaxThreads();
+
+    // pool stats
+    public int getPoolSize();
+
+    public int getActiveCount();
+
+    public long getTaskCount();
+
+    public long getCompletedTaskCount();
+
+}

java/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java

@@ -0,0 +1,334 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.util.HashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.group.InterceptorPayload;
+import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.util.StringManager;
+
+
+/**
+ *
+ * The order interceptor guarantees that messages are received in the same order they were
+ * sent.
+ * This interceptor works best with the ack=true setting. <br>
+ * There is no point in
+ * using this with the replicationMode="fastasynchqueue" as this mode guarantees ordering.<BR>
+ * If you are using the mode ack=false replicationMode=pooled, and have a lot of concurrent threads,
+ * this interceptor can really slow you down, as many messages will be completely out of order
+ * and the queue might become rather large. If this is the case, then you might want to set
+ * the value OrderInterceptor.maxQueue = 25 (meaning that we will never keep more than 25 messages in our queue)
+ * <br><b>Configuration Options</b><br>
+ * OrderInterceptor.expire=&lt;milliseconds&gt; - if a message arrives out of order, how long before we act on it <b>default=3000ms</b><br>
+ * OrderInterceptor.maxQueue=&lt;max queue size&gt; - how much can the queue grow to ensure ordering.
+ *   This setting is useful to avoid OutOfMemoryErrors<b>default=Integer.MAX_VALUE</b><br>
+ * OrderInterceptor.forwardExpired=&lt;boolean&gt; - this flag tells the interceptor what to
+ * do when a message has expired or the queue has grown larger than the maxQueue value.
+ * true means that the message is sent up the stack to the receiver that will receive and out of order message
+ * false means, forget the message and reset the message counter. <b>default=true</b>
+ *
+ *
+ * @version 1.1
+ */
+public class OrderInterceptor extends ChannelInterceptorBase {
+    protected static final StringManager sm = StringManager.getManager(OrderInterceptor.class);
+    private final HashMap<Member, Counter> outcounter = new HashMap<>();
+    private final HashMap<Member, Counter> incounter = new HashMap<>();
+    private final HashMap<Member, MessageOrder> incoming = new HashMap<>();
+    private long expire = 3000;
+    private boolean forwardExpired = true;
+    private int maxQueue = Integer.MAX_VALUE;
+
+    final ReentrantReadWriteLock inLock = new ReentrantReadWriteLock(true);
+    final ReentrantReadWriteLock outLock= new ReentrantReadWriteLock(true);
+
+    @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;
+                outLock.writeLock().lock();
+                try {
+                    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) {
+        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());
+        inLock.writeLock().lock();
+        try {
+            if ( processIncoming(order) ) processLeftOvers(msg.getAddress(),false);
+        } finally {
+            inLock.writeLock().unlock();
+        }
+    }
+    protected void processLeftOvers(Member member, boolean force) {
+        MessageOrder tmp = incoming.get(member);
+        if ( force ) {
+            Counter cnt = getInCounter(member);
+            cnt.setCounter(Integer.MAX_VALUE);
+        }
+        if ( tmp!= null ) processIncoming(tmp);
+    }
+    /**
+     *
+     * @param order MessageOrder
+     * @return boolean - true if a message expired and was processed
+     */
+    protected boolean processIncoming(MessageOrder order) {
+        boolean result = false;
+        Member member = order.getMessage().getAddress();
+        Counter cnt = getInCounter(member);
+
+        MessageOrder tmp = incoming.get(member);
+        if ( tmp != null ) {
+            order = MessageOrder.add(tmp,order);
+        }
+
+
+        while ( (order!=null) && (order.getMsgNr() <= cnt.getCounter())  ) {
+            //we are right on target. process orders
+            if ( order.getMsgNr() == cnt.getCounter() ) cnt.inc();
+            else if ( order.getMsgNr() > cnt.getCounter() ) cnt.setCounter(order.getMsgNr());
+            super.messageReceived(order.getMessage());
+            order.setMessage(null);
+            order = order.next;
+        }
+        MessageOrder head = order;
+        MessageOrder prev = null;
+        tmp = order;
+        //flag to empty out the queue when it larger than maxQueue
+        boolean empty = order!=null?order.getCount()>=maxQueue:false;
+        while ( tmp != null ) {
+            //process expired messages or empty out the queue
+            if ( tmp.isExpired(expire) || empty ) {
+                //reset the head
+                if ( tmp == head ) head = tmp.next;
+                cnt.setCounter(tmp.getMsgNr()+1);
+                if ( getForwardExpired() )
+                    super.messageReceived(tmp.getMessage());
+                tmp.setMessage(null);
+                tmp = tmp.next;
+                if ( prev != null ) prev.next = tmp;
+                result = true;
+            } else {
+                prev = tmp;
+                tmp = tmp.next;
+            }
+        }
+        if ( head == null ) incoming.remove(member);
+        else incoming.put(member, head);
+        return result;
+    }
+
+    @Override
+    public void memberAdded(Member member) {
+        //notify upwards
+        super.memberAdded(member);
+    }
+
+    @Override
+    public void memberDisappeared(Member member) {
+        //reset counters - lock free
+        incounter.remove(member);
+        outcounter.remove(member);
+        //clear the remaining queue
+        processLeftOvers(member,true);
+        //notify upwards
+        super.memberDisappeared(member);
+    }
+
+    protected int incCounter(Member mbr) {
+        Counter cnt = getOutCounter(mbr);
+        return cnt.inc();
+    }
+
+    protected Counter getInCounter(Member mbr) {
+        Counter cnt = incounter.get(mbr);
+        if ( cnt == null ) {
+            cnt = new Counter();
+            cnt.inc(); //always start at 1 for incoming
+            incounter.put(mbr,cnt);
+        }
+        return cnt;
+    }
+
+    protected Counter getOutCounter(Member mbr) {
+        Counter cnt = outcounter.get(mbr);
+        if ( cnt == null ) {
+            cnt = new Counter();
+            outcounter.put(mbr,cnt);
+        }
+        return cnt;
+    }
+
+    protected static class Counter {
+        private final AtomicInteger value = new AtomicInteger(0);
+
+        public int getCounter() {
+            return value.get();
+        }
+
+        public void setCounter(int counter) {
+            this.value.set(counter);
+        }
+
+        public int inc() {
+            return value.addAndGet(1);
+        }
+    }
+
+    protected static class MessageOrder {
+        private final long received = System.currentTimeMillis();
+        private MessageOrder next;
+        private final int msgNr;
+        private ChannelMessage msg = null;
+        public MessageOrder(int msgNr,ChannelMessage msg) {
+            this.msgNr = msgNr;
+            this.msg = msg;
+        }
+
+        public boolean isExpired(long expireTime) {
+            return (System.currentTimeMillis()-received) > expireTime;
+        }
+
+        public ChannelMessage getMessage() {
+            return msg;
+        }
+
+        public void setMessage(ChannelMessage msg) {
+            this.msg = msg;
+        }
+
+        public void setNext(MessageOrder order) {
+            this.next = order;
+        }
+        public MessageOrder getNext() {
+            return next;
+        }
+
+        public int getCount() {
+            int counter = 1;
+            MessageOrder tmp = next;
+            while ( tmp != null ) {
+                counter++;
+                tmp = tmp.next;
+            }
+            return counter;
+        }
+
+        @SuppressWarnings("null") // prev cannot be null
+        public static MessageOrder add(MessageOrder head, MessageOrder add) {
+            if ( head == null ) return add;
+            if ( add == null ) return head;
+            if ( head == add ) return add;
+
+            if ( head.getMsgNr() > add.getMsgNr() ) {
+                add.next = head;
+                return add;
+            }
+
+            MessageOrder iter = head;
+            MessageOrder prev = null;
+            while ( iter.getMsgNr() < add.getMsgNr() && (iter.next !=null ) ) {
+                prev = iter;
+                iter = iter.next;
+            }
+            if ( iter.getMsgNr() < add.getMsgNr() ) {
+                //add after
+                add.next = iter.next;
+                iter.next = add;
+            } else if (iter.getMsgNr() > add.getMsgNr()) {
+                //add before
+                prev.next = add; // prev cannot be null here, warning suppressed
+                add.next = iter;
+
+            } else {
+                throw new ArithmeticException(sm.getString("orderInterceptor.messageAdded.sameCounter"));
+            }
+
+            return head;
+        }
+
+        public int getMsgNr() {
+            return msgNr;
+        }
+
+
+    }
+
+    public void setExpire(long expire) {
+        this.expire = expire;
+    }
+
+    public void setForwardExpired(boolean forwardExpired) {
+        this.forwardExpired = forwardExpired;
+    }
+
+    public void setMaxQueue(int maxQueue) {
+        this.maxQueue = maxQueue;
+    }
+
+    public long getExpire() {
+        return expire;
+    }
+
+    public boolean getForwardExpired() {
+        return forwardExpired;
+    }
+
+    public int getMaxQueue() {
+        return maxQueue;
+    }
+
+}

java/org/apache/catalina/tribes/group/interceptors/SimpleCoordinator.java

@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.AbsoluteOrder;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+
+/**
+ * A dinky coordinator, just uses a sorted version of the member array.
+ *
+ * @author rnewson
+ *
+ */
+public class SimpleCoordinator extends ChannelInterceptorBase {
+
+    private Member[] view;
+
+    private final AtomicBoolean membershipChanged = new AtomicBoolean();
+
+    private void membershipChanged() {
+        membershipChanged.set(true);
+    }
+
+    @Override
+    public void memberAdded(final Member member) {
+        super.memberAdded(member);
+        membershipChanged();
+        installViewWhenStable();
+    }
+
+    @Override
+    public void memberDisappeared(final Member member) {
+        super.memberDisappeared(member);
+        membershipChanged();
+        installViewWhenStable();
+    }
+
+    /**
+     * Override to receive view changes.
+     *
+     * @param view The members array
+     */
+    protected void viewChange(final Member[] view) {
+    }
+
+    @Override
+    public void start(int svc) throws ChannelException {
+        super.start(svc);
+        installViewWhenStable();
+    }
+
+    private void installViewWhenStable() {
+        int stableCount = 0;
+
+        while (stableCount < 10) {
+            if (membershipChanged.compareAndSet(true, false)) {
+                stableCount = 0;
+            } else {
+                stableCount++;
+            }
+            try {
+                TimeUnit.MILLISECONDS.sleep(250);
+            } catch (final InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+
+        final Member[] members = getMembers();
+        final Member[] view = new Member[members.length+1];
+        System.arraycopy(members, 0, view, 0, members.length);
+        view[members.length] = getLocalMember(false);
+        Arrays.sort(view, AbsoluteOrder.comp);
+        if (Arrays.equals(view, this.view)) {
+            return;
+        }
+        this.view = view;
+        viewChange(view);
+    }
+
+    @Override
+    public void stop(int svc) throws ChannelException {
+        super.stop(svc);
+    }
+
+    public Member[] getView() {
+        return view;
+    }
+
+    public Member getCoordinator() {
+        return view == null ? null : view[0];
+    }
+
+    public boolean isCoordinator() {
+        return view == null ? false : getLocalMember(false).equals(
+                getCoordinator());
+    }
+
+}

java/org/apache/catalina/tribes/group/interceptors/StaticMembershipInterceptor.java

@@ -0,0 +1,238 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ChannelInterceptor;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.AbsoluteOrder;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.io.ChannelData;
+import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.util.StringManager;
+import org.apache.juli.logging.Log;
+import org.apache.juli.logging.LogFactory;
+
+public class StaticMembershipInterceptor extends ChannelInterceptorBase
+        implements StaticMembershipInterceptorMBean {
+
+    private static final Log log = LogFactory.getLog(StaticMembershipInterceptor.class);
+    protected static final StringManager sm =
+            StringManager.getManager(StaticMembershipInterceptor.class);
+
+    protected static final byte[] MEMBER_START = new byte[] {
+        76, 111, 99, 97, 108, 32, 83, 116, 97, 116, 105, 99, 77, 101, 109, 98, 101, 114, 32, 78,
+        111, 116, 105, 102, 105, 99, 97, 116, 105, 111, 110, 32, 68, 97, 116, 97};
+
+    protected static final byte[] MEMBER_STOP = new byte[] {
+        76, 111, 99, 97, 108, 32, 83, 116, 97, 116, 105, 99, 77, 101, 109, 98, 101, 114, 32, 83,
+        104, 117, 116, 100, 111, 119, 110, 32, 68, 97, 116, 97};
+
+    protected final ArrayList<Member> members = new ArrayList<>();
+    protected Member localMember = null;
+
+    public StaticMembershipInterceptor() {
+        super();
+    }
+
+    public void addStaticMember(Member member) {
+        synchronized (members) {
+            if (!members.contains(member)) members.add(member);
+        }
+    }
+
+    public void removeStaticMember(Member member) {
+        synchronized (members) {
+            if (members.contains(member)) members.remove(member);
+        }
+    }
+
+    public void setLocalMember(Member member) {
+        this.localMember = member;
+        localMember.setLocal(true);
+    }
+
+    @Override
+    public void messageReceived(ChannelMessage msg) {
+        if (msg.getMessage().getLength() == MEMBER_START.length &&
+                Arrays.equals(MEMBER_START, msg.getMessage().getBytes())) {
+            // receive member start
+            Member member = getMember(msg.getAddress());
+            if (member != null) {
+                super.memberAdded(member);
+            }
+        } else if (msg.getMessage().getLength() == MEMBER_STOP.length &&
+                    Arrays.equals(MEMBER_STOP, msg.getMessage().getBytes())) {
+            // receive member shutdown
+            Member member = getMember(msg.getAddress());
+            if (member != null) {
+                try {
+                    member.setCommand(Member.SHUTDOWN_PAYLOAD);
+                    super.memberDisappeared(member);
+                } finally {
+                    member.setCommand(new byte[0]);
+                }
+            }
+        } else {
+            super.messageReceived(msg);
+        }
+    }
+
+    /**
+     * has members
+     */
+    @Override
+    public boolean hasMembers() {
+        return super.hasMembers() || (members.size()>0);
+    }
+
+    /**
+     * Get all current cluster members
+     * @return all members or empty array
+     */
+    @Override
+    public Member[] getMembers() {
+        if ( members.size() == 0 ) return super.getMembers();
+        else {
+            synchronized (members) {
+                Member[] others = super.getMembers();
+                Member[] result = new Member[members.size() + others.length];
+                for (int i = 0; i < others.length; i++) result[i] = others[i];
+                for (int i = 0; i < members.size(); i++) result[i + others.length] = members.get(i);
+                AbsoluteOrder.absoluteOrder(result);
+                return result;
+            }//sync
+        }//end if
+    }
+
+    /**
+     *
+     * @param mbr Member
+     * @return Member
+     */
+    @Override
+    public Member getMember(Member mbr) {
+        if ( members.contains(mbr) ) return members.get(members.indexOf(mbr));
+        else return super.getMember(mbr);
+    }
+
+    /**
+     * Return the member that represents this node.
+     *
+     * @return Member
+     */
+    @Override
+    public Member getLocalMember(boolean incAlive) {
+        if (this.localMember != null ) return localMember;
+        else return super.getLocalMember(incAlive);
+    }
+
+    /**
+     * {@inheritDoc}
+     * <p>
+     * Sends notifications upwards.
+     */
+    @Override
+    public void start(int svc) throws ChannelException {
+        if ( (Channel.SND_RX_SEQ&svc)==Channel.SND_RX_SEQ ) super.start(Channel.SND_RX_SEQ);
+        if ( (Channel.SND_TX_SEQ&svc)==Channel.SND_TX_SEQ ) super.start(Channel.SND_TX_SEQ);
+        final ChannelInterceptorBase base = this;
+        for (final Member member : members) {
+            Thread t = new Thread() {
+                @Override
+                public void run() {
+                    base.memberAdded(member);
+                    if (getfirstInterceptor().getMember(member) != null) {
+                        sendLocalMember(new Member[]{member});
+                    }
+                }
+            };
+            t.start();
+        }
+        super.start(svc & (~Channel.SND_RX_SEQ) & (~Channel.SND_TX_SEQ));
+
+        // check required interceptors
+        TcpFailureDetector failureDetector = null;
+        TcpPingInterceptor pingInterceptor = null;
+        ChannelInterceptor prev = getPrevious();
+        while (prev != null) {
+            if (prev instanceof TcpFailureDetector ) failureDetector = (TcpFailureDetector) prev;
+            if (prev instanceof TcpPingInterceptor) pingInterceptor = (TcpPingInterceptor) prev;
+            prev = prev.getPrevious();
+        }
+        if (failureDetector == null) {
+            log.warn(sm.getString("staticMembershipInterceptor.no.failureDetector"));
+        }
+        if (pingInterceptor == null) {
+            log.warn(sm.getString("staticMembershipInterceptor.no.pingInterceptor"));
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     * <p>
+     * Sends local member shutdown.
+     */
+    @Override
+    public void stop(int svc) throws ChannelException {
+        // Sends local member shutdown.
+        Member[] members = getfirstInterceptor().getMembers();
+        sendShutdown(members);
+        super.stop(svc);
+    }
+
+    protected void sendLocalMember(Member[] members) {
+        try {
+            sendMemberMessage(members, MEMBER_START);
+        } catch (ChannelException cx) {
+            log.warn(sm.getString("staticMembershipInterceptor.sendLocalMember.failed"),cx);
+        }
+    }
+
+    protected void sendShutdown(Member[] members) {
+        try {
+            sendMemberMessage(members, MEMBER_STOP);
+        } catch (ChannelException cx) {
+            log.warn(sm.getString("staticMembershipInterceptor.sendShutdown.failed"),cx);
+        }
+    }
+
+    protected ChannelInterceptor getfirstInterceptor() {
+        ChannelInterceptor result = null;
+        ChannelInterceptor now = this;
+        do {
+            result = now;
+            now = now.getPrevious();
+        } while (now.getPrevious() != null);
+        return result;
+    }
+
+    protected void sendMemberMessage(Member[] members, byte[] message) throws ChannelException {
+        if ( members == null || members.length == 0 ) return;
+        ChannelData data = new ChannelData(true);
+        data.setAddress(getLocalMember(false));
+        data.setTimestamp(System.currentTimeMillis());
+        data.setOptions(getOptionFlag());
+        data.setMessage(new XByteBuffer(message, false));
+        super.sendMessage(members, data, null);
+    }
+}

java/org/apache/catalina/tribes/group/interceptors/StaticMembershipInterceptorMBean.java

@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+import org.apache.catalina.tribes.Member;
+
+public interface StaticMembershipInterceptorMBean  {
+
+    public int getOptionFlag();
+
+    public Member getLocalMember(boolean incAlive);
+}

java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetector.java

@@ -0,0 +1,415 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.net.ConnectException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.NoRouteToHostException;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.util.Arrays;
+import java.util.HashMap;
+
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ChannelException.FaultyMember;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.RemoteProcessException;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.group.InterceptorPayload;
+import org.apache.catalina.tribes.io.ChannelData;
+import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.membership.Membership;
+import org.apache.catalina.tribes.membership.StaticMember;
+import org.apache.catalina.tribes.util.StringManager;
+import org.apache.juli.logging.Log;
+import org.apache.juli.logging.LogFactory;
+
+/**
+ * <p>Title: A perfect failure detector </p>
+ *
+ * <p>Description: The TcpFailureDetector is a useful interceptor
+ * that adds reliability to the membership layer.</p>
+ * <p>
+ * If the network is busy, or the system is busy so that the membership receiver thread
+ * is not getting enough time to update its table, members can be &quot;timed out&quot;
+ * This failure detector will intercept the memberDisappeared message(unless its a true shutdown message)
+ * and connect to the member using TCP.
+ * </p>
+ * <p>
+ * The TcpFailureDetector works in two ways. <br>
+ * 1. It intercepts memberDisappeared events
+ * 2. It catches send errors
+ * </p>
+ *
+ * @version 1.0
+ */
+public class TcpFailureDetector extends ChannelInterceptorBase implements TcpFailureDetectorMBean {
+
+    private static final Log log = LogFactory.getLog(TcpFailureDetector.class);
+    protected static final StringManager sm = StringManager.getManager(TcpFailureDetector.class);
+
+    protected static final byte[] TCP_FAIL_DETECT = new byte[] {
+        79, -89, 115, 72, 121, -126, 67, -55, -97, 111, -119, -128, -95, 91, 7, 20,
+        125, -39, 82, 91, -21, -15, 67, -102, -73, 126, -66, -113, -127, 103, 30, -74,
+        55, 21, -66, -121, 69, 126, 76, -88, -65, 10, 77, 19, 83, 56, 21, 50,
+        85, -10, -108, -73, 58, -6, 64, 120, -111, 4, 125, -41, 114, -124, -64, -43};
+
+    protected long connectTimeout = 1000;//1 second default
+
+    protected boolean performSendTest = true;
+
+    protected boolean performReadTest = false;
+
+    protected long readTestTimeout = 5000;//5 seconds
+
+    protected Membership membership = null;
+
+    protected final HashMap<Member, Long> removeSuspects = new HashMap<>();
+
+    protected final HashMap<Member, Long> addSuspects = new HashMap<>();
+
+    protected int removeSuspectsTimeout = 300; // 5 minutes
+
+    @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);
+    }//messageReceived
+
+
+    @Override
+    public void memberAdded(Member member) {
+        if ( membership == null ) setupMembership();
+        boolean notify = false;
+        synchronized (membership) {
+            if (removeSuspects.containsKey(member)) {
+                //previously marked suspect, system below picked up the member again
+                removeSuspects.remove(member);
+            } else if (membership.getMember(member) == null){
+                //if we add it here, then add it upwards too
+                //check to see if it is alive
+                if (memberAlive(member)) {
+                    membership.memberAlive(member);
+                    notify = true;
+                } else {
+                    if (member instanceof StaticMember) {
+                        addSuspects.put(member, Long.valueOf(System.currentTimeMillis()));
+                    }
+                }
+            }
+        }
+        if ( notify ) super.memberAdded(member);
+    }
+
+    @Override
+    public void memberDisappeared(Member member) {
+        if ( membership == null ) setupMembership();
+        boolean shutdown = Arrays.equals(member.getCommand(),Member.SHUTDOWN_PAYLOAD);
+        if (shutdown) {
+            synchronized (membership) {
+                if (!membership.contains(member)) return;
+                membership.removeMember(member);
+                removeSuspects.remove(member);
+                if (member instanceof StaticMember) {
+                    addSuspects.put(member, Long.valueOf(System.currentTimeMillis()));
+                }
+            }
+            super.memberDisappeared(member);
+        } else {
+            boolean notify = false;
+            if(log.isInfoEnabled())
+                log.info(sm.getString("tcpFailureDetector.memberDisappeared.verify", member));
+            synchronized (membership) {
+                if (!membership.contains(member)) {
+                    if(log.isInfoEnabled())
+                        log.info(sm.getString("tcpFailureDetector.already.disappeared", member));
+                    return;
+                }
+                //check to see if the member really is gone
+                if (!memberAlive(member)) {
+                    //not correct, we need to maintain the map
+                    membership.removeMember(member);
+                    removeSuspects.remove(member);
+                    if (member instanceof StaticMember) {
+                        addSuspects.put(member, Long.valueOf(System.currentTimeMillis()));
+                    }
+                    notify = true;
+                } else {
+                    //add the member as suspect
+                    removeSuspects.put(member, Long.valueOf(System.currentTimeMillis()));
+                }
+            }
+            if ( notify ) {
+                if(log.isInfoEnabled())
+                    log.info(sm.getString("tcpFailureDetector.member.disappeared", member));
+                super.memberDisappeared(member);
+            } else {
+                if(log.isInfoEnabled())
+                    log.info(sm.getString("tcpFailureDetector.still.alive", member));
+            }
+        }
+    }
+
+    @Override
+    public boolean hasMembers() {
+        if ( membership == null ) setupMembership();
+        return membership.hasMembers();
+    }
+
+    @Override
+    public Member[] getMembers() {
+        if ( membership == null ) setupMembership();
+        return membership.getMembers();
+    }
+
+    @Override
+    public Member getMember(Member mbr) {
+        if ( membership == null ) setupMembership();
+        return membership.getMember(mbr);
+    }
+
+    @Override
+    public Member getLocalMember(boolean incAlive) {
+        return super.getLocalMember(incAlive);
+    }
+
+    @Override
+    public void heartbeat() {
+        super.heartbeat();
+        checkMembers(false);
+    }
+
+    @Override
+    public void checkMembers(boolean checkAll) {
+        try {
+            if (membership == null) setupMembership();
+            synchronized (membership) {
+                if (!checkAll) performBasicCheck();
+                else performForcedCheck();
+            }
+        } catch (Exception x) {
+            log.warn(sm.getString("tcpFailureDetector.heartbeat.failed"),x);
+        }
+    }
+
+    protected void performForcedCheck() {
+        //update all alive times
+        Member[] members = super.getMembers();
+        for (int i = 0; members != null && i < members.length; i++) {
+            if (memberAlive(members[i])) {
+                if (membership.memberAlive(members[i])) super.memberAdded(members[i]);
+                addSuspects.remove(members[i]);
+            } else {
+                if (membership.getMember(members[i])!=null) {
+                    membership.removeMember(members[i]);
+                    removeSuspects.remove(members[i]);
+                    if (members[i] instanceof StaticMember) {
+                        addSuspects.put(members[i], Long.valueOf(System.currentTimeMillis()));
+                    }
+                    super.memberDisappeared(members[i]);
+                }
+            } //end if
+        } //for
+
+    }
+
+    protected void performBasicCheck() {
+        //update all alive times
+        Member[] members = super.getMembers();
+        for (int i = 0; members != null && i < members.length; i++) {
+            if (addSuspects.containsKey(members[i]) && membership.getMember(members[i]) == null) {
+                // avoid temporary adding member.
+                continue;
+            }
+            if (membership.memberAlive(members[i])) {
+                //we don't have this one in our membership, check to see if he/she is alive
+                if (memberAlive(members[i])) {
+                    log.warn(sm.getString("tcpFailureDetector.performBasicCheck.memberAdded", members[i]));
+                    super.memberAdded(members[i]);
+                } else {
+                    membership.removeMember(members[i]);
+                } //end if
+            } //end if
+        } //for
+
+        //check suspect members if they are still alive,
+        //if not, simply issue the memberDisappeared message
+        Member[] keys = removeSuspects.keySet().toArray(new Member[removeSuspects.size()]);
+        for (int i = 0; i < keys.length; i++) {
+            Member m = keys[i];
+            if (membership.getMember(m) != null && (!memberAlive(m))) {
+                membership.removeMember(m);
+                if (m instanceof StaticMember) {
+                    addSuspects.put(m, Long.valueOf(System.currentTimeMillis()));
+                }
+                super.memberDisappeared(m);
+                removeSuspects.remove(m);
+                if(log.isInfoEnabled())
+                    log.info(sm.getString("tcpFailureDetector.suspectMember.dead", m));
+            } else {
+                if (removeSuspectsTimeout > 0) {
+                    long timeNow = System.currentTimeMillis();
+                    int timeIdle = (int) ((timeNow - removeSuspects.get(m).longValue()) / 1000L);
+                    if (timeIdle > removeSuspectsTimeout) {
+                        removeSuspects.remove(m); // remove suspect member
+                    }
+                }
+            }
+        }
+
+        //check add suspects members if they are alive now,
+        //if they are, simply issue the memberAdded message
+        keys = addSuspects.keySet().toArray(new Member[addSuspects.size()]);
+        for (int i = 0; i < keys.length; i++) {
+            Member m = keys[i];
+            if ( membership.getMember(m) == null && (memberAlive(m))) {
+                membership.memberAlive(m);
+                super.memberAdded(m);
+                addSuspects.remove(m);
+                if(log.isInfoEnabled())
+                    log.info(sm.getString("tcpFailureDetector.suspectMember.alive", m));
+            } //end if
+        }
+    }
+
+    protected synchronized void setupMembership() {
+        if ( membership == null ) {
+            membership = new Membership(super.getLocalMember(true));
+        }
+
+    }
+
+    protected boolean memberAlive(Member mbr) {
+        return memberAlive(mbr,TCP_FAIL_DETECT,performSendTest,performReadTest,readTestTimeout,connectTimeout,getOptionFlag());
+    }
+
+    protected boolean memberAlive(Member mbr, byte[] msgData,
+                                         boolean sendTest, boolean readTest,
+                                         long readTimeout, long conTimeout,
+                                         int optionFlag) {
+        //could be a shutdown notification
+        if ( Arrays.equals(mbr.getCommand(),Member.SHUTDOWN_PAYLOAD) ) return false;
+
+        try (Socket socket = new Socket()) {
+            InetAddress ia = InetAddress.getByAddress(mbr.getHost());
+            InetSocketAddress addr = new InetSocketAddress(ia, mbr.getPort());
+            socket.setSoTimeout((int)readTimeout);
+            socket.connect(addr, (int) conTimeout);
+            if ( sendTest ) {
+                ChannelData data = new ChannelData(true);
+                data.setAddress(getLocalMember(false));
+                data.setMessage(new XByteBuffer(msgData,false));
+                data.setTimestamp(System.currentTimeMillis());
+                int options = optionFlag | Channel.SEND_OPTIONS_BYTE_MESSAGE;
+                if ( readTest ) options = (options | Channel.SEND_OPTIONS_USE_ACK);
+                else options = (options & (~Channel.SEND_OPTIONS_USE_ACK));
+                data.setOptions(options);
+                byte[] message = XByteBuffer.createDataPackage(data);
+                socket.getOutputStream().write(message);
+                if ( readTest ) {
+                    int length = socket.getInputStream().read(message);
+                    return length > 0;
+                }
+            }//end if
+            return true;
+        } catch (SocketTimeoutException | ConnectException | NoRouteToHostException noop) {
+            //do nothing, we couldn't connect
+        } catch (Exception x) {
+            log.error(sm.getString("tcpFailureDetector.failureDetection.failed", mbr),x);
+        }
+        return false;
+    }
+
+    @Override
+    public long getReadTestTimeout() {
+        return readTestTimeout;
+    }
+
+    @Override
+    public boolean getPerformSendTest() {
+        return performSendTest;
+    }
+
+    @Override
+    public boolean getPerformReadTest() {
+        return performReadTest;
+    }
+
+    @Override
+    public long getConnectTimeout() {
+        return connectTimeout;
+    }
+
+    @Override
+    public int getRemoveSuspectsTimeout() {
+        return removeSuspectsTimeout;
+    }
+
+    @Override
+    public void setPerformReadTest(boolean performReadTest) {
+        this.performReadTest = performReadTest;
+    }
+
+    @Override
+    public void setPerformSendTest(boolean performSendTest) {
+        this.performSendTest = performSendTest;
+    }
+
+    @Override
+    public void setReadTestTimeout(long readTestTimeout) {
+        this.readTestTimeout = readTestTimeout;
+    }
+
+    @Override
+    public void setConnectTimeout(long connectTimeout) {
+        this.connectTimeout = connectTimeout;
+    }
+
+    @Override
+    public void setRemoveSuspectsTimeout(int removeSuspectsTimeout) {
+        this.removeSuspectsTimeout = removeSuspectsTimeout;
+    }
+
+}

java/org/apache/catalina/tribes/group/interceptors/TcpFailureDetectorMBean.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+public interface TcpFailureDetectorMBean {
+
+    public int getOptionFlag();
+
+    // Attributes
+    public long getConnectTimeout();
+
+    public boolean getPerformSendTest();
+
+    public boolean getPerformReadTest();
+
+    public long getReadTestTimeout();
+
+    public int getRemoveSuspectsTimeout();
+
+    public void setPerformReadTest(boolean performReadTest);
+
+    public void setPerformSendTest(boolean performSendTest);
+
+    public void setReadTestTimeout(long readTestTimeout);
+
+    public void setConnectTimeout(long connectTimeout);
+
+    public void setRemoveSuspectsTimeout(int removeSuspectsTimeout);
+
+    // Operations
+    public void checkMembers(boolean checkAll);
+}

java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptor.java

@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.lang.ref.WeakReference;
+import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ChannelInterceptor;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.io.ChannelData;
+import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.util.StringManager;
+import org.apache.juli.logging.Log;
+import org.apache.juli.logging.LogFactory;
+
+/**
+ *
+ * Sends a ping to all members.
+ * Configure this interceptor with the TcpFailureDetector below it,
+ * and the TcpFailureDetector will act as the membership guide.
+ * @version 1.0
+ */
+
+public class TcpPingInterceptor extends ChannelInterceptorBase implements TcpPingInterceptorMBean {
+
+    private static final Log log = LogFactory.getLog(TcpPingInterceptor.class);
+    protected static final StringManager sm = StringManager.getManager(TcpPingInterceptor.class);
+
+    protected static final byte[] TCP_PING_DATA = new byte[] {
+        79, -89, 115, 72, 121, -33, 67, -55, -97, 111, -119, -128, -95, 91, 7, 20,
+        125, -39, 82, 91, -21, -33, 67, -102, -73, 126, -66, -113, -127, 103, 30, -74,
+        55, 21, -66, -121, 69, 33, 76, -88, -65, 10, 77, 19, 83, 56, 21, 50,
+        85, -10, -108, -73, 58, -33, 33, 120, -111, 4, 125, -41, 114, -124, -64, -43};
+
+    protected long interval = 1000; //1 second
+
+    protected boolean useThread = false;
+    protected boolean staticOnly = false;
+    protected volatile boolean running = true;
+    protected PingThread thread = null;
+    protected static final AtomicInteger cnt = new AtomicInteger(0);
+
+    WeakReference<TcpFailureDetector> failureDetector = null;
+    WeakReference<StaticMembershipInterceptor> staticMembers = null;
+
+    @Override
+    public synchronized void start(int svc) throws ChannelException {
+        super.start(svc);
+        running = true;
+        if ( thread == null && useThread) {
+            thread = new PingThread();
+            thread.setDaemon(true);
+            String channelName = "";
+            if (getChannel().getName() != null) channelName = "[" + getChannel().getName() + "]";
+            thread.setName("TcpPingInterceptor.PingThread" + channelName +"-"+cnt.addAndGet(1));
+            thread.start();
+        }
+
+        //acquire the interceptors to invoke on send ping events
+        ChannelInterceptor next = getNext();
+        while ( next != null ) {
+            if ( next instanceof TcpFailureDetector )
+                failureDetector = new WeakReference<>((TcpFailureDetector)next);
+            if ( next instanceof StaticMembershipInterceptor )
+                staticMembers = new WeakReference<>((StaticMembershipInterceptor)next);
+            next = next.getNext();
+        }
+
+    }
+
+    @Override
+    public synchronized void stop(int svc) throws ChannelException {
+        running = false;
+        if (thread != null) {
+            thread.interrupt();
+            thread = null;
+        }
+        super.stop(svc);
+    }
+
+    @Override
+    public void heartbeat() {
+        super.heartbeat();
+        if (!getUseThread()) sendPing();
+    }
+
+    @Override
+    public long getInterval() {
+        return interval;
+    }
+
+    public void setInterval(long interval) {
+        this.interval = interval;
+    }
+
+    public void setUseThread(boolean useThread) {
+        this.useThread = useThread;
+    }
+
+    public void setStaticOnly(boolean staticOnly) {
+        this.staticOnly = staticOnly;
+    }
+
+    @Override
+    public boolean getUseThread() {
+        return useThread;
+    }
+
+    public boolean getStaticOnly() {
+        return staticOnly;
+    }
+
+    protected void sendPing() {
+        TcpFailureDetector tcpFailureDetector =
+                failureDetector != null ? failureDetector.get() : null;
+        if (tcpFailureDetector != null) {
+            // We have a reference to the failure detector
+            // Piggy back on it
+            tcpFailureDetector.checkMembers(true);
+        } else {
+            StaticMembershipInterceptor smi =
+                    staticOnly && staticMembers != null ? staticMembers.get() : null;
+            if (smi != null) {
+                sendPingMessage(smi.getMembers());
+            } else {
+                sendPingMessage(getMembers());
+            }
+        }
+    }
+
+    protected void sendPingMessage(Member[] members) {
+        if ( members == null || members.length == 0 ) return;
+        ChannelData data = new ChannelData(true);//generates a unique Id
+        data.setAddress(getLocalMember(false));
+        data.setTimestamp(System.currentTimeMillis());
+        data.setOptions(getOptionFlag());
+        data.setMessage(new XByteBuffer(TCP_PING_DATA, false));
+        try {
+            super.sendMessage(members, data, null);
+        }catch (ChannelException x) {
+            log.warn(sm.getString("tcpPingInterceptor.ping.failed"),x);
+        }
+    }
+
+    @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);
+    }//messageReceived
+
+    protected class PingThread extends Thread {
+        @Override
+        public void run() {
+            while (running) {
+                try {
+                    sleep(interval);
+                    sendPing();
+                }catch ( InterruptedException ix ) {
+                    // Ignore. Probably triggered by a call to stop().
+                    // In the highly unlikely event it was a different trigger,
+                    // simply ignore it and continue.
+                }catch ( Exception x )  {
+                    log.warn(sm.getString("tcpPingInterceptor.pingFailed.pingThread"),x);
+                }
+            }
+        }
+    }
+
+
+}

java/org/apache/catalina/tribes/group/interceptors/TcpPingInterceptorMBean.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.catalina.tribes.group.interceptors;
+
+public interface TcpPingInterceptorMBean {
+
+    public int getOptionFlag();
+
+    public long getInterval();
+
+    public boolean getUseThread();
+
+}

java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptor.java

@@ -0,0 +1,160 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.text.DecimalFormat;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.group.InterceptorPayload;
+import org.apache.catalina.tribes.io.ChannelData;
+import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.util.StringManager;
+import org.apache.juli.logging.Log;
+import org.apache.juli.logging.LogFactory;
+
+public class ThroughputInterceptor extends ChannelInterceptorBase
+        implements ThroughputInterceptorMBean {
+
+    private static final Log log = LogFactory.getLog(ThroughputInterceptor.class);
+    protected static final StringManager sm = StringManager.getManager(ThroughputInterceptor.class);
+
+    double mbTx = 0;
+    double mbAppTx = 0;
+    double mbRx = 0;
+    double timeTx = 0;
+    double lastCnt = 0;
+    final AtomicLong msgTxCnt = new AtomicLong(1);
+    final AtomicLong msgRxCnt = new AtomicLong(0);
+    final AtomicLong msgTxErr = new AtomicLong(0);
+    int interval = 10000;
+    final AtomicInteger access = new AtomicInteger(0);
+    long txStart = 0;
+    long rxStart = 0;
+    final DecimalFormat df = new DecimalFormat("#0.00");
+
+
+    @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() / (double) interval) >= lastCnt) {
+                lastCnt++;
+                report(timeTx);
+            }
+        }
+        msgTxCnt.addAndGet(1);
+    }
+
+    @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 report(double timeTx) {
+        if ( log.isInfoEnabled() )
+            log.info(sm.getString("throughputInterceptor.report",
+                    msgTxCnt, df.format(mbTx), df.format(mbAppTx), df.format(timeTx),
+                    df.format(mbTx/timeTx), df.format(mbAppTx/timeTx), msgTxErr, msgRxCnt,
+                    df.format(mbRx/((System.currentTimeMillis()-rxStart)/(double)1000)),
+                    df.format(mbRx)));
+    }
+
+    @Override
+    public void setInterval(int interval) {
+        this.interval = interval;
+    }
+
+    @Override
+    public int getInterval() {
+        return interval;
+    }
+
+    @Override
+    public double getLastCnt() {
+        return lastCnt;
+    }
+
+    @Override
+    public double getMbAppTx() {
+        return mbAppTx;
+    }
+
+    @Override
+    public double getMbRx() {
+        return mbRx;
+    }
+
+    @Override
+    public double getMbTx() {
+        return mbTx;
+    }
+
+    @Override
+    public AtomicLong getMsgRxCnt() {
+        return msgRxCnt;
+    }
+
+    @Override
+    public AtomicLong getMsgTxCnt() {
+        return msgTxCnt;
+    }
+
+    @Override
+    public AtomicLong getMsgTxErr() {
+        return msgTxErr;
+    }
+
+    @Override
+    public long getRxStart() {
+        return rxStart;
+    }
+
+    @Override
+    public double getTimeTx() {
+        return timeTx;
+    }
+
+    @Override
+    public long getTxStart() {
+        return txStart;
+    }
+
+}

java/org/apache/catalina/tribes/group/interceptors/ThroughputInterceptorMBean.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+public interface ThroughputInterceptorMBean {
+
+    public int getOptionFlag();
+
+    // Attributes
+    public int getInterval();
+
+    public void setInterval(int interval);
+
+    // stats
+    public double getLastCnt();
+
+    public double getMbAppTx();
+
+    public double getMbRx();
+
+    public double getMbTx();
+
+    public AtomicLong getMsgRxCnt();
+
+    public AtomicLong getMsgTxCnt();
+
+    public AtomicLong getMsgTxErr();
+
+    public long getRxStart();
+
+    public double getTimeTx();
+
+    public long getTxStart();
+
+    // Operations
+    public void report(double timeTx);
+
+}

java/org/apache/catalina/tribes/group/interceptors/TwoPhaseCommitInterceptor.java

@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.catalina.tribes.group.interceptors;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.UniqueId;
+import org.apache.catalina.tribes.group.ChannelInterceptorBase;
+import org.apache.catalina.tribes.group.InterceptorPayload;
+import org.apache.catalina.tribes.util.Arrays;
+import org.apache.catalina.tribes.util.StringManager;
+import org.apache.catalina.tribes.util.UUIDGenerator;
+import org.apache.juli.logging.Log;
+import org.apache.juli.logging.LogFactory;
+
+public class TwoPhaseCommitInterceptor extends ChannelInterceptorBase {
+
+    private static final byte[] START_DATA = new byte[] {113, 1, -58, 2, -34, -60, 75, -78, -101, -12, 32, -29, 32, 111, -40, 4};
+    private static final byte[] END_DATA = new byte[] {54, -13, 90, 110, 47, -31, 75, -24, -81, -29, 36, 52, -58, 77, -110, 56};
+    private static final Log log = LogFactory.getLog(TwoPhaseCommitInterceptor.class);
+    protected static final StringManager sm = StringManager.getManager(TwoPhaseCommitInterceptor.class);
+
+    protected final HashMap<UniqueId, MapEntry> messages = new HashMap<>();
+    protected long expire = 1000 * 60; //one minute expiration
+    protected boolean deepclone = true;
+
+    @Override
+    public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws
+        ChannelException {
+        //todo, optimize, if destination.length==1, then we can do
+        //msg.setOptions(msg.getOptions() & (~getOptionFlag())
+        //and just send one message
+        if (okToProcess(msg.getOptions()) ) {
+            super.sendMessage(destination, msg, null);
+            ChannelMessage confirmation = null;
+            if ( deepclone ) confirmation = (ChannelMessage)msg.deepclone();
+            else confirmation = (ChannelMessage)msg.clone();
+            confirmation.getMessage().reset();
+            UUIDGenerator.randomUUID(false,confirmation.getUniqueId(),0);
+            confirmation.getMessage().append(START_DATA,0,START_DATA.length);
+            confirmation.getMessage().append(msg.getUniqueId(),0,msg.getUniqueId().length);
+            confirmation.getMessage().append(END_DATA,0,END_DATA.length);
+            super.sendMessage(destination,confirmation,payload);
+        } else {
+            //turn off two phase commit
+            //this wont work if the interceptor has 0 as a flag
+            //since there is no flag to turn off
+            //msg.setOptions(msg.getOptions() & (~getOptionFlag()));
+            super.sendMessage(destination, msg, payload);
+        }
+    }
+
+    @Override
+    public void messageReceived(ChannelMessage msg) {
+        if (okToProcess(msg.getOptions())) {
+            if ( msg.getMessage().getLength() == (START_DATA.length+msg.getUniqueId().length+END_DATA.length) &&
+                 Arrays.contains(msg.getMessage().getBytesDirect(),0,START_DATA,0,START_DATA.length) &&
+                 Arrays.contains(msg.getMessage().getBytesDirect(),START_DATA.length+msg.getUniqueId().length,END_DATA,0,END_DATA.length) ) {
+                UniqueId id = new UniqueId(msg.getMessage().getBytesDirect(),START_DATA.length,msg.getUniqueId().length);
+                MapEntry original = messages.get(id);
+                if ( original != null ) {
+                    super.messageReceived(original.msg);
+                    messages.remove(id);
+                } else log.warn(sm.getString("twoPhaseCommitInterceptor.originalMessage.missing", Arrays.toString(id.getBytes())));
+            } else {
+                UniqueId id = new UniqueId(msg.getUniqueId());
+                MapEntry entry = new MapEntry((ChannelMessage)msg.deepclone(),id,System.currentTimeMillis());
+                messages.put(id,entry);
+            }
+        } else {
+            super.messageReceived(msg);
+        }
+    }
+
+    public boolean getDeepclone() {
+        return deepclone;
+    }
+
+    public long getExpire() {
+        return expire;
+    }
+
+    public void setDeepclone(boolean deepclone) {
+        this.deepclone = deepclone;
+    }
+
+    public void setExpire(long expire) {
+        this.expire = expire;
+    }
+
+    @Override
+    public void heartbeat() {
+        try {
+            long now = System.currentTimeMillis();
+            @SuppressWarnings("unchecked")
+            Map.Entry<UniqueId,MapEntry>[] entries = messages.entrySet().toArray(new Map.Entry[messages.size()]);
+            for (int i=0; i<entries.length; i++ ) {
+                MapEntry entry = entries[i].getValue();
+                if ( entry.expired(now,expire) ) {
+                    if(log.isInfoEnabled())
+                        log.info("Message ["+entry.id+"] has expired. Removing.");
+                    messages.remove(entry.id);
+                }//end if
+            }
+        } catch ( Exception x ) {
+            log.warn(sm.getString("twoPhaseCommitInterceptor.heartbeat.failed"),x);
+        } finally {
+            super.heartbeat();
+        }
+    }
+
+    public static class MapEntry {
+        public final ChannelMessage msg;
+        public final UniqueId id;
+        public final long timestamp;
+
+        public MapEntry(ChannelMessage msg, UniqueId id, long timestamp) {
+            this.msg = msg;
+            this.id = id;
+            this.timestamp = timestamp;
+        }
+        public boolean expired(long now, long expiration) {
+            return (now - timestamp ) > expiration;
+        }
+
+    }
+
+}