public int getPort();
/**
+ * Returns the secure listening port
+ * @return port, -1 if a secure port is not activated
+ */
+ public int getSecurePort();
+
+ /**
* Sets the message listener to receive notification of incoming
* @param listener MessageListener
* @see MessageListener
/**
* Returns the listen port for the ChannelReceiver implementation
- * @return IPv4 or IPv6 representation of the host address this member listens to incoming data
+ * @return the listen port for this member, -1 if its not listening on an unsecure port
* @see ChannelReceiver
*/
public int getPort();
+
+ /**
+ * Returns the secure listen port for the ChannelReceiver implementation.
+ * Returns -1 if its not listening to a secure port.
+ * @return the listen port for this member, -1 if its not listening on a secure port
+ * @see ChannelReceiver
+ */
+ public int getSecurePort();
+
/**
* Contains information on how long this member has been online.
-/*\r
- * Copyright 1999,2004-2005 The Apache Software Foundation.\r
- * \r
- * Licensed under the Apache License, Version 2.0 (the "License");\r
- * you may not use this file except in compliance with the License.\r
- * You may obtain a copy of the License at\r
- * \r
- * http://www.apache.org/licenses/LICENSE-2.0\r
- * \r
- * Unless required by applicable law or agreed to in writing, software\r
- * distributed under the License is distributed on an "AS IS" BASIS,\r
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
- * See the License for the specific language governing permissions and\r
- * limitations under the License.\r
- */\r
-\r
-package org.apache.catalina.tribes.membership;\r
-\r
-import java.io.IOException;\r
-import java.io.ObjectInput;\r
-import java.io.ObjectOutput;\r
-import java.util.Arrays;\r
-\r
-import org.apache.catalina.tribes.Member;\r
-import org.apache.catalina.tribes.io.XByteBuffer;\r
-import org.apache.catalina.tribes.transport.SenderState;\r
-\r
-/**\r
- * A <b>membership</b> implementation using simple multicast.\r
- * This is the representation of a multicast member.\r
- * Carries the host, and port of the this or other cluster nodes.\r
- *\r
- * @author Filip Hanik\r
- * @version $Revision: 304032 $, $Date: 2005-07-27 10:11:55 -0500 (Wed, 27 Jul 2005) $\r
- */\r
-public class MemberImpl implements Member, java.io.Externalizable {\r
-\r
- /**\r
- * Public properties specific to this implementation\r
- */\r
- public static final transient String TCP_LISTEN_PORT = "tcpListenPort";\r
- public static final transient String TCP_LISTEN_HOST = "tcpListenHost";\r
- public static final transient String MEMBER_NAME = "memberName";\r
- \r
- public static final transient byte[] TRIBES_MBR_BEGIN = new byte[] {84, 82, 73, 66, 69, 83, 45, 66};\r
- public static final transient byte[] TRIBES_MBR_END = new byte[] {84, 82, 73, 66, 69, 83, 45, 69};\r
- \r
- /**\r
- * The listen host for this member\r
- */\r
- protected byte[] host;\r
- protected transient String hostname;\r
- /**\r
- * The tcp listen port for this member\r
- */\r
- protected int port;\r
-\r
- /**\r
- * Counter for how many broadcast messages have been sent from this member\r
- */\r
- protected int msgCount = 0;\r
- /**\r
- * The number of milliseconds since this members was\r
- * created, is kept track of using the start time\r
- */\r
- protected long memberAliveTime = 0;\r
- \r
- /**\r
- * For the local member only\r
- */\r
- protected transient long serviceStartTime;\r
- \r
- /**\r
- * To avoid serialization over and over again, once the local dataPkg\r
- * has been set, we use that to transmit data\r
- */\r
- protected transient byte[] dataPkg = null;\r
-\r
- /**\r
- * Unique session Id for this member\r
- */\r
- protected byte[] uniqueId = new byte[16];\r
- \r
- /**\r
- * Custom payload that an app framework can broadcast\r
- * Also used to transport stop command.\r
- */\r
- protected byte[] payload = new byte[0];\r
- \r
- /**\r
- * Command, so that the custom payload doesn't have to be used\r
- * This is for internal tribes use, such as SHUTDOWN_COMMAND\r
- */\r
- protected byte[] command = new byte[0];\r
-\r
- /**\r
- * Domain if we want to filter based on domain.\r
- */\r
- protected byte[] domain = new byte[0];\r
- \r
- /**\r
- * Empty constructor for serialization\r
- */\r
- public MemberImpl() {\r
- \r
- }\r
-\r
- /**\r
- * Construct a new member object\r
- * @param name - the name of this member, cluster unique\r
- * @param domain - the cluster domain name of this member\r
- * @param host - the tcp listen host\r
- * @param port - the tcp listen port\r
- */\r
- public MemberImpl(String host,\r
- int port,\r
- long aliveTime) throws IOException {\r
- setHostname(host);\r
- this.port = port;\r
- this.memberAliveTime=aliveTime;\r
- }\r
- \r
- public MemberImpl(String host,\r
- int port,\r
- long aliveTime,\r
- byte[] payload) throws IOException {\r
- this(host,port,aliveTime);\r
- setPayload(payload);\r
- }\r
- \r
- public boolean isReady() {\r
- return SenderState.getSenderState(this).isReady();\r
- }\r
- public boolean isSuspect() {\r
- return SenderState.getSenderState(this).isSuspect();\r
- }\r
- public boolean isFailing() {\r
- return SenderState.getSenderState(this).isFailing();\r
- }\r
-\r
- /**\r
- * Increment the message count.\r
- */\r
- protected void inc() {\r
- msgCount++;\r
- }\r
-\r
- /**\r
- * Create a data package to send over the wire representing this member.\r
- * This is faster than serialization.\r
- * @return - the bytes for this member deserialized\r
- * @throws Exception\r
- */\r
- public byte[] getData() {\r
- return getData(true);\r
- }\r
- /**\r
- * Highly optimized version of serializing a member into a byte array\r
- * Returns a cached byte[] reference, do not modify this data\r
- * @param getalive boolean\r
- * @return byte[]\r
- */\r
- public byte[] getData(boolean getalive) {\r
- return getData(getalive,false);\r
- }\r
- \r
- \r
- public int getDataLength() {\r
- return TRIBES_MBR_BEGIN.length+ //start pkg\r
- 4+ //data length\r
- 8+ //alive time\r
- 4+ //port\r
- 1+ //host length\r
- host.length+ //host\r
- 4+ //command length\r
- command.length+ //command\r
- 4+ //domain length\r
- domain.length+ //domain\r
- 16+ //unique id\r
- 4+ //payload length\r
- payload.length+ //payload\r
- TRIBES_MBR_END.length; //end pkg\r
- }\r
- \r
- /**\r
- * \r
- * @param getalive boolean - calculate memberAlive time\r
- * @param reset boolean - reset the cached data package, and create a new one\r
- * @return byte[]\r
- */\r
- public byte[] getData(boolean getalive, boolean reset) {\r
- if ( reset ) dataPkg = null;\r
- //look in cache first\r
- if ( dataPkg!=null ) {\r
- if ( getalive ) {\r
- //you'd be surprised, but System.currentTimeMillis\r
- //shows up on the profiler\r
- long alive=System.currentTimeMillis()-getServiceStartTime();\r
- XByteBuffer.toBytes( (long) alive, dataPkg, TRIBES_MBR_BEGIN.length+4);\r
- }\r
- return dataPkg;\r
- }\r
- \r
- //package looks like\r
- //start package TRIBES_MBR_BEGIN.length\r
- //package length - 4 bytes\r
- //alive - 8 bytes\r
- //port - 4 bytes\r
- //host length - 1 byte\r
- //host - hl bytes\r
- //clen - 4 bytes\r
- //command - clen bytes\r
- //dlen - 4 bytes\r
- //domain - dlen bytes\r
- //uniqueId - 16 bytes\r
- //payload length - 4 bytes\r
- //payload plen bytes\r
- //end package TRIBES_MBR_END.length\r
- byte[] addr = host;\r
- long alive=System.currentTimeMillis()-getServiceStartTime();\r
- byte hl = (byte)addr.length;\r
- byte[] data = new byte[getDataLength()];\r
- \r
- int bodylength = (getDataLength() - TRIBES_MBR_BEGIN.length - TRIBES_MBR_END.length - 4);\r
- \r
- int pos = 0;\r
- \r
- //TRIBES_MBR_BEGIN\r
- System.arraycopy(TRIBES_MBR_BEGIN,0,data,pos,TRIBES_MBR_BEGIN.length);\r
- pos += TRIBES_MBR_BEGIN.length;\r
- \r
- //body length\r
- XByteBuffer.toBytes(bodylength,data,pos);\r
- pos += 4;\r
- \r
- //alive data\r
- XByteBuffer.toBytes((long)alive,data,pos);\r
- pos += 8;\r
- //port\r
- XByteBuffer.toBytes(port,data,pos);\r
- pos += 4;\r
- //host length\r
- data[pos++] = hl;\r
- //host\r
- System.arraycopy(addr,0,data,pos,addr.length);\r
- pos+=addr.length;\r
- //clen - 4 bytes\r
- XByteBuffer.toBytes(command.length,data,pos);\r
- pos+=4;\r
- //command - clen bytes\r
- System.arraycopy(command,0,data,pos,command.length);\r
- pos+=command.length;\r
- //dlen - 4 bytes\r
- XByteBuffer.toBytes(domain.length,data,pos);\r
- pos+=4;\r
- //domain - dlen bytes\r
- System.arraycopy(domain,0,data,pos,domain.length);\r
- pos+=domain.length;\r
- //unique Id\r
- System.arraycopy(uniqueId,0,data,pos,uniqueId.length);\r
- pos+=uniqueId.length;\r
- //payload\r
- XByteBuffer.toBytes(payload.length,data,pos);\r
- pos+=4;\r
- System.arraycopy(payload,0,data,pos,payload.length);\r
- pos+=payload.length;\r
- \r
- //TRIBES_MBR_END\r
- System.arraycopy(TRIBES_MBR_END,0,data,pos,TRIBES_MBR_END.length);\r
- pos += TRIBES_MBR_END.length;\r
-\r
- //create local data\r
- dataPkg = data;\r
- return data;\r
- }\r
- /**\r
- * Deserializes a member from data sent over the wire\r
- * @param data - the bytes received\r
- * @return a member object.\r
- */\r
- public static MemberImpl getMember(byte[] data, MemberImpl member) {\r
- return getMember(data,0,data.length,member);\r
- }\r
-\r
- public static MemberImpl getMember(byte[] data, int offset, int length, MemberImpl member) {\r
- //package looks like\r
- //start package TRIBES_MBR_BEGIN.length\r
- //package length - 4 bytes\r
- //alive - 8 bytes\r
- //port - 4 bytes\r
- //host length - 1 byte\r
- //host - hl bytes\r
- //clen - 4 bytes\r
- //command - clen bytes\r
- //dlen - 4 bytes\r
- //domain - dlen bytes\r
- //uniqueId - 16 bytes\r
- //payload length - 4 bytes\r
- //payload plen bytes\r
- //end package TRIBES_MBR_END.length\r
-\r
- int pos = offset;\r
- \r
- if (XByteBuffer.firstIndexOf(data,offset,TRIBES_MBR_BEGIN)!=pos) {\r
- throw new IllegalArgumentException("Invalid package, should start with:"+org.apache.catalina.tribes.util.Arrays.toString(TRIBES_MBR_BEGIN));\r
- }\r
-\r
- if ( length < (TRIBES_MBR_BEGIN.length+4) ) {\r
- throw new ArrayIndexOutOfBoundsException("Member package to small to validate.");\r
- }\r
- \r
- pos += TRIBES_MBR_BEGIN.length;\r
- \r
- int bodylength = XByteBuffer.toInt(data,pos);\r
- pos += 4;\r
- \r
- if ( length < (bodylength+4+TRIBES_MBR_BEGIN.length+TRIBES_MBR_END.length) ) {\r
- throw new ArrayIndexOutOfBoundsException("Not enough bytes in member package.");\r
- }\r
- \r
- int endpos = pos+bodylength;\r
- if (XByteBuffer.firstIndexOf(data,endpos,TRIBES_MBR_END)!=endpos) {\r
- throw new IllegalArgumentException("Invalid package, should end with:"+org.apache.catalina.tribes.util.Arrays.toString(TRIBES_MBR_END));\r
- }\r
-\r
-\r
- byte[] alived = new byte[8];\r
- System.arraycopy(data, pos, alived, 0, 8);\r
- pos += 8;\r
- byte[] portd = new byte[4];\r
- System.arraycopy(data, pos, portd, 0, 4);\r
- pos += 4;\r
- \r
- byte hl = data[pos++];\r
- byte[] addr = new byte[hl];\r
- System.arraycopy(data, pos, addr, 0, hl);\r
- pos += hl;\r
- \r
- int cl = XByteBuffer.toInt(data, pos);\r
- pos += 4;\r
- \r
- byte[] command = new byte[cl];\r
- System.arraycopy(data, pos, command, 0, command.length);\r
- pos += command.length;\r
- \r
- int dl = XByteBuffer.toInt(data, pos);\r
- pos += 4;\r
- \r
- byte[] domain = new byte[dl];\r
- System.arraycopy(data, pos, domain, 0, domain.length);\r
- pos += domain.length;\r
- \r
- byte[] uniqueId = new byte[16];\r
- System.arraycopy(data, pos, uniqueId, 0, 16);\r
- pos += 16;\r
- \r
- int pl = XByteBuffer.toInt(data, pos);\r
- pos += 4;\r
- \r
- byte[] payload = new byte[pl];\r
- System.arraycopy(data, pos, payload, 0, payload.length);\r
- pos += payload.length;\r
- \r
- member.setHost(addr);\r
- member.setPort(XByteBuffer.toInt(portd, 0));\r
- member.setMemberAliveTime(XByteBuffer.toLong(alived, 0));\r
- member.setUniqueId(uniqueId);\r
- member.payload = payload;\r
- member.domain = domain;\r
- member.command = command;\r
- \r
- member.dataPkg = new byte[length];\r
- System.arraycopy(data, offset, member.dataPkg, 0, length);\r
- \r
- return member;\r
- }\r
-\r
- public static MemberImpl getMember(byte[] data) {\r
- return getMember(data,new MemberImpl());\r
- }\r
-\r
- /**\r
- * Return the name of this object\r
- * @return a unique name to the cluster\r
- */\r
- public String getName() {\r
- return "tcp://"+getHostname()+":"+getPort();\r
- }\r
- \r
- /**\r
- * Return the listen port of this member\r
- * @return - tcp listen port\r
- */\r
- public int getPort() {\r
- return this.port;\r
- }\r
-\r
- /**\r
- * Return the TCP listen host for this member\r
- * @return IP address or host name\r
- */\r
- public byte[] getHost() {\r
- return host;\r
- }\r
- \r
- public String getHostname() {\r
- if ( this.hostname != null ) return hostname;\r
- else {\r
- try {\r
- this.hostname = java.net.InetAddress.getByAddress(host).getHostName();\r
- return this.hostname;\r
- }catch ( IOException x ) {\r
- throw new RuntimeException("Unable to parse hostname.",x);\r
- }\r
- }\r
- }\r
-\r
- /**\r
- * Contains information on how long this member has been online.\r
- * The result is the number of milli seconds this member has been\r
- * broadcasting its membership to the cluster.\r
- * @return nr of milliseconds since this member started.\r
- */\r
- public long getMemberAliveTime() {\r
- return memberAliveTime;\r
- }\r
-\r
- public long getServiceStartTime() {\r
- return serviceStartTime;\r
- }\r
-\r
- public byte[] getUniqueId() {\r
- return uniqueId;\r
- }\r
-\r
- public byte[] getPayload() {\r
- return payload;\r
- }\r
-\r
- public byte[] getCommand() {\r
- return command;\r
- }\r
-\r
- public byte[] getDomain() {\r
- return domain;\r
- }\r
-\r
- public void setMemberAliveTime(long time) {\r
- memberAliveTime=time;\r
- }\r
-\r
-\r
-\r
- /**\r
- * String representation of this object\r
- */\r
- public String toString() {\r
- StringBuffer buf = new StringBuffer("org.apache.catalina.tribes.membership.MemberImpl[");\r
- buf.append(getName()).append(",");\r
- buf.append(getHostname()).append(",");\r
- buf.append(port).append(", alive=");\r
- buf.append(memberAliveTime).append(",");\r
- buf.append("id=").append(bToS(this.uniqueId)).append(", ");\r
- buf.append("payload=").append(bToS(this.payload,8)).append(", ");\r
- buf.append("command=").append(bToS(this.command,8)).append(", ");\r
- buf.append("domain=").append(bToS(this.domain,8)).append(", ");\r
- buf.append("]");\r
- return buf.toString();\r
- }\r
- public static String bToS(byte[] data) {\r
- return bToS(data,data.length);\r
- }\r
- public static String bToS(byte[] data, int max) {\r
- StringBuffer buf = new StringBuffer(4*16);\r
- buf.append("{");\r
- for (int i=0; data!=null && i<data.length; i++ ) {\r
- buf.append(String.valueOf(data[i])).append(" ");\r
- if ( i==max ) {\r
- buf.append("...("+data.length+")");\r
- break;\r
- }\r
- }\r
- buf.append("}");\r
- return buf.toString();\r
- }\r
-\r
- /**\r
- * @see java.lang.Object#hashCode()\r
- * @return The hash code\r
- */\r
- public int hashCode() {\r
- return getHost()[0]+getHost()[1]+getHost()[2]+getHost()[3];\r
- }\r
-\r
- /**\r
- * Returns true if the param o is a McastMember with the same name\r
- * @param o\r
- */\r
- public boolean equals(Object o) {\r
- if ( o instanceof MemberImpl ) {\r
- return Arrays.equals(this.getHost(),((MemberImpl)o).getHost()) &&\r
- this.getPort() == ((MemberImpl)o).getPort() &&\r
- Arrays.equals(this.getUniqueId(),((MemberImpl)o).getUniqueId());\r
- }\r
- else\r
- return false;\r
- }\r
- \r
- public void setHost(byte[] host) {\r
- this.host = host;\r
- }\r
- \r
- public void setHostname(String host) throws IOException {\r
- hostname = host;\r
- this.host = java.net.InetAddress.getByName(host).getAddress();\r
- }\r
- \r
- public void setMsgCount(int msgCount) {\r
- this.msgCount = msgCount;\r
- }\r
-\r
- public void setPort(int port) {\r
- this.port = port;\r
- this.dataPkg = null;\r
- }\r
-\r
- public void setServiceStartTime(long serviceStartTime) {\r
- this.serviceStartTime = serviceStartTime;\r
- }\r
-\r
- public void setUniqueId(byte[] uniqueId) {\r
- this.uniqueId = uniqueId!=null?uniqueId:new byte[16];\r
- getData(true,true);\r
- }\r
-\r
- public void setPayload(byte[] payload) {\r
- byte[] oldpayload = this.payload;\r
- this.payload = payload!=null?payload:new byte[0];\r
- if ( this.getData(true,true).length > McastServiceImpl.MAX_PACKET_SIZE ) {\r
- this.payload = oldpayload;\r
- throw new IllegalArgumentException("Payload is to large for tribes to handle.");\r
- }\r
- \r
- }\r
-\r
- public void setCommand(byte[] command) {\r
- this.command = command!=null?command:new byte[0];\r
- getData(true,true);\r
- }\r
-\r
- public void setDomain(byte[] domain) {\r
- this.domain = domain!=null?domain:new byte[0];\r
- getData(true,true);\r
- }\r
-\r
- public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {\r
- int length = in.readInt();\r
- byte[] message = new byte[length];\r
- in.read(message);\r
- getMember(message,this);\r
- \r
- }\r
-\r
- public void writeExternal(ObjectOutput out) throws IOException {\r
- byte[] data = this.getData();\r
- out.writeInt(data.length);\r
- out.write(data);\r
- }\r
- \r
-}\r
+/*
+ * Copyright 1999,2004-2005 The Apache Software Foundation.
+ *
+ * Licensed 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.membership;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.Arrays;
+
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.transport.SenderState;
+
+/**
+ * A <b>membership</b> implementation using simple multicast.
+ * This is the representation of a multicast member.
+ * Carries the host, and port of the this or other cluster nodes.
+ *
+ * @author Filip Hanik
+ * @version $Revision: 304032 $, $Date: 2005-07-27 10:11:55 -0500 (Wed, 27 Jul 2005) $
+ */
+public class MemberImpl implements Member, java.io.Externalizable {
+
+ /**
+ * Public properties specific to this implementation
+ */
+ public static final transient String TCP_LISTEN_PORT = "tcpListenPort";
+ public static final transient String TCP_LISTEN_HOST = "tcpListenHost";
+ public static final transient String MEMBER_NAME = "memberName";
+
+ public static final transient byte[] TRIBES_MBR_BEGIN = new byte[] {84, 82, 73, 66, 69, 83, 45, 66};
+ public static final transient byte[] TRIBES_MBR_END = new byte[] {84, 82, 73, 66, 69, 83, 45, 69};
+
+ /**
+ * The listen host for this member
+ */
+ protected byte[] host;
+ protected transient String hostname;
+ /**
+ * The tcp listen port for this member
+ */
+ protected int port;
+
+ /**
+ * The tcp/SSL listen port for this member
+ */
+ protected int securePort = -1;
+
+ /**
+ * Counter for how many broadcast messages have been sent from this member
+ */
+ protected int msgCount = 0;
+ /**
+ * The number of milliseconds since this members was
+ * created, is kept track of using the start time
+ */
+ protected long memberAliveTime = 0;
+
+ /**
+ * For the local member only
+ */
+ protected transient long serviceStartTime;
+
+ /**
+ * To avoid serialization over and over again, once the local dataPkg
+ * has been set, we use that to transmit data
+ */
+ protected transient byte[] dataPkg = null;
+
+ /**
+ * Unique session Id for this member
+ */
+ protected byte[] uniqueId = new byte[16];
+
+ /**
+ * Custom payload that an app framework can broadcast
+ * Also used to transport stop command.
+ */
+ protected byte[] payload = new byte[0];
+
+ /**
+ * Command, so that the custom payload doesn't have to be used
+ * This is for internal tribes use, such as SHUTDOWN_COMMAND
+ */
+ protected byte[] command = new byte[0];
+
+ /**
+ * Domain if we want to filter based on domain.
+ */
+ protected byte[] domain = new byte[0];
+
+ /**
+ * Empty constructor for serialization
+ */
+ public MemberImpl() {
+
+ }
+
+ /**
+ * Construct a new member object
+ * @param name - the name of this member, cluster unique
+ * @param domain - the cluster domain name of this member
+ * @param host - the tcp listen host
+ * @param port - the tcp listen port
+ */
+ public MemberImpl(String host,
+ int port,
+ long aliveTime) throws IOException {
+ setHostname(host);
+ this.port = port;
+ this.memberAliveTime=aliveTime;
+ }
+
+ public MemberImpl(String host,
+ int port,
+ long aliveTime,
+ byte[] payload) throws IOException {
+ this(host,port,aliveTime);
+ setPayload(payload);
+ }
+
+ public boolean isReady() {
+ return SenderState.getSenderState(this).isReady();
+ }
+ public boolean isSuspect() {
+ return SenderState.getSenderState(this).isSuspect();
+ }
+ public boolean isFailing() {
+ return SenderState.getSenderState(this).isFailing();
+ }
+
+ /**
+ * Increment the message count.
+ */
+ protected void inc() {
+ msgCount++;
+ }
+
+ /**
+ * Create a data package to send over the wire representing this member.
+ * This is faster than serialization.
+ * @return - the bytes for this member deserialized
+ * @throws Exception
+ */
+ public byte[] getData() {
+ return getData(true);
+ }
+ /**
+ * Highly optimized version of serializing a member into a byte array
+ * Returns a cached byte[] reference, do not modify this data
+ * @param getalive boolean
+ * @return byte[]
+ */
+ public byte[] getData(boolean getalive) {
+ return getData(getalive,false);
+ }
+
+
+ public int getDataLength() {
+ return TRIBES_MBR_BEGIN.length+ //start pkg
+ 4+ //data length
+ 8+ //alive time
+ 4+ //port
+ 4+ //secure port
+ 1+ //host length
+ host.length+ //host
+ 4+ //command length
+ command.length+ //command
+ 4+ //domain length
+ domain.length+ //domain
+ 16+ //unique id
+ 4+ //payload length
+ payload.length+ //payload
+ TRIBES_MBR_END.length; //end pkg
+ }
+
+ /**
+ *
+ * @param getalive boolean - calculate memberAlive time
+ * @param reset boolean - reset the cached data package, and create a new one
+ * @return byte[]
+ */
+ public byte[] getData(boolean getalive, boolean reset) {
+ if ( reset ) dataPkg = null;
+ //look in cache first
+ if ( dataPkg!=null ) {
+ if ( getalive ) {
+ //you'd be surprised, but System.currentTimeMillis
+ //shows up on the profiler
+ long alive=System.currentTimeMillis()-getServiceStartTime();
+ XByteBuffer.toBytes( (long) alive, dataPkg, TRIBES_MBR_BEGIN.length+4);
+ }
+ return dataPkg;
+ }
+
+ //package looks like
+ //start package TRIBES_MBR_BEGIN.length
+ //package length - 4 bytes
+ //alive - 8 bytes
+ //port - 4 bytes
+ //secure port - 4 bytes
+ //host length - 1 byte
+ //host - hl bytes
+ //clen - 4 bytes
+ //command - clen bytes
+ //dlen - 4 bytes
+ //domain - dlen bytes
+ //uniqueId - 16 bytes
+ //payload length - 4 bytes
+ //payload plen bytes
+ //end package TRIBES_MBR_END.length
+ byte[] addr = host;
+ long alive=System.currentTimeMillis()-getServiceStartTime();
+ byte hl = (byte)addr.length;
+ byte[] data = new byte[getDataLength()];
+
+ int bodylength = (getDataLength() - TRIBES_MBR_BEGIN.length - TRIBES_MBR_END.length - 4);
+
+ int pos = 0;
+
+ //TRIBES_MBR_BEGIN
+ System.arraycopy(TRIBES_MBR_BEGIN,0,data,pos,TRIBES_MBR_BEGIN.length);
+ pos += TRIBES_MBR_BEGIN.length;
+
+ //body length
+ XByteBuffer.toBytes(bodylength,data,pos);
+ pos += 4;
+
+ //alive data
+ XByteBuffer.toBytes((long)alive,data,pos);
+ pos += 8;
+ //port
+ XByteBuffer.toBytes(port,data,pos);
+ pos += 4;
+ //secure port
+ XByteBuffer.toBytes(securePort,data,pos);
+ pos += 4;
+ //host length
+ data[pos++] = hl;
+ //host
+ System.arraycopy(addr,0,data,pos,addr.length);
+ pos+=addr.length;
+ //clen - 4 bytes
+ XByteBuffer.toBytes(command.length,data,pos);
+ pos+=4;
+ //command - clen bytes
+ System.arraycopy(command,0,data,pos,command.length);
+ pos+=command.length;
+ //dlen - 4 bytes
+ XByteBuffer.toBytes(domain.length,data,pos);
+ pos+=4;
+ //domain - dlen bytes
+ System.arraycopy(domain,0,data,pos,domain.length);
+ pos+=domain.length;
+ //unique Id
+ System.arraycopy(uniqueId,0,data,pos,uniqueId.length);
+ pos+=uniqueId.length;
+ //payload
+ XByteBuffer.toBytes(payload.length,data,pos);
+ pos+=4;
+ System.arraycopy(payload,0,data,pos,payload.length);
+ pos+=payload.length;
+
+ //TRIBES_MBR_END
+ System.arraycopy(TRIBES_MBR_END,0,data,pos,TRIBES_MBR_END.length);
+ pos += TRIBES_MBR_END.length;
+
+ //create local data
+ dataPkg = data;
+ return data;
+ }
+ /**
+ * Deserializes a member from data sent over the wire
+ * @param data - the bytes received
+ * @return a member object.
+ */
+ public static MemberImpl getMember(byte[] data, MemberImpl member) {
+ return getMember(data,0,data.length,member);
+ }
+
+ public static MemberImpl getMember(byte[] data, int offset, int length, MemberImpl member) {
+ //package looks like
+ //start package TRIBES_MBR_BEGIN.length
+ //package length - 4 bytes
+ //alive - 8 bytes
+ //port - 4 bytes
+ //secure port - 4 bytes
+ //host length - 1 byte
+ //host - hl bytes
+ //clen - 4 bytes
+ //command - clen bytes
+ //dlen - 4 bytes
+ //domain - dlen bytes
+ //uniqueId - 16 bytes
+ //payload length - 4 bytes
+ //payload plen bytes
+ //end package TRIBES_MBR_END.length
+
+ int pos = offset;
+
+ if (XByteBuffer.firstIndexOf(data,offset,TRIBES_MBR_BEGIN)!=pos) {
+ throw new IllegalArgumentException("Invalid package, should start with:"+org.apache.catalina.tribes.util.Arrays.toString(TRIBES_MBR_BEGIN));
+ }
+
+ if ( length < (TRIBES_MBR_BEGIN.length+4) ) {
+ throw new ArrayIndexOutOfBoundsException("Member package to small to validate.");
+ }
+
+ pos += TRIBES_MBR_BEGIN.length;
+
+ int bodylength = XByteBuffer.toInt(data,pos);
+ pos += 4;
+
+ if ( length < (bodylength+4+TRIBES_MBR_BEGIN.length+TRIBES_MBR_END.length) ) {
+ throw new ArrayIndexOutOfBoundsException("Not enough bytes in member package.");
+ }
+
+ int endpos = pos+bodylength;
+ if (XByteBuffer.firstIndexOf(data,endpos,TRIBES_MBR_END)!=endpos) {
+ throw new IllegalArgumentException("Invalid package, should end with:"+org.apache.catalina.tribes.util.Arrays.toString(TRIBES_MBR_END));
+ }
+
+
+ byte[] alived = new byte[8];
+ System.arraycopy(data, pos, alived, 0, 8);
+ pos += 8;
+ byte[] portd = new byte[4];
+ System.arraycopy(data, pos, portd, 0, 4);
+ pos += 4;
+
+ byte[] sportd = new byte[4];
+ System.arraycopy(data, pos, sportd, 0, 4);
+ pos += 4;
+
+
+
+ byte hl = data[pos++];
+ byte[] addr = new byte[hl];
+ System.arraycopy(data, pos, addr, 0, hl);
+ pos += hl;
+
+ int cl = XByteBuffer.toInt(data, pos);
+ pos += 4;
+
+ byte[] command = new byte[cl];
+ System.arraycopy(data, pos, command, 0, command.length);
+ pos += command.length;
+
+ int dl = XByteBuffer.toInt(data, pos);
+ pos += 4;
+
+ byte[] domain = new byte[dl];
+ System.arraycopy(data, pos, domain, 0, domain.length);
+ pos += domain.length;
+
+ byte[] uniqueId = new byte[16];
+ System.arraycopy(data, pos, uniqueId, 0, 16);
+ pos += 16;
+
+ int pl = XByteBuffer.toInt(data, pos);
+ pos += 4;
+
+ byte[] payload = new byte[pl];
+ System.arraycopy(data, pos, payload, 0, payload.length);
+ pos += payload.length;
+
+ member.setHost(addr);
+ member.setPort(XByteBuffer.toInt(portd, 0));
+ member.setSecurePort(XByteBuffer.toInt(sportd, 0));
+ member.setMemberAliveTime(XByteBuffer.toLong(alived, 0));
+ member.setUniqueId(uniqueId);
+ member.payload = payload;
+ member.domain = domain;
+ member.command = command;
+
+ member.dataPkg = new byte[length];
+ System.arraycopy(data, offset, member.dataPkg, 0, length);
+
+ return member;
+ }
+
+ public static MemberImpl getMember(byte[] data) {
+ return getMember(data,new MemberImpl());
+ }
+
+ /**
+ * Return the name of this object
+ * @return a unique name to the cluster
+ */
+ public String getName() {
+ return "tcp://"+getHostname()+":"+getPort();
+ }
+
+ /**
+ * Return the listen port of this member
+ * @return - tcp listen port
+ */
+ public int getPort() {
+ return this.port;
+ }
+
+ /**
+ * Return the TCP listen host for this member
+ * @return IP address or host name
+ */
+ public byte[] getHost() {
+ return host;
+ }
+
+ public String getHostname() {
+ if ( this.hostname != null ) return hostname;
+ else {
+ try {
+ this.hostname = java.net.InetAddress.getByAddress(host).getHostName();
+ return this.hostname;
+ }catch ( IOException x ) {
+ throw new RuntimeException("Unable to parse hostname.",x);
+ }
+ }
+ }
+
+ /**
+ * Contains information on how long this member has been online.
+ * The result is the number of milli seconds this member has been
+ * broadcasting its membership to the cluster.
+ * @return nr of milliseconds since this member started.
+ */
+ public long getMemberAliveTime() {
+ return memberAliveTime;
+ }
+
+ public long getServiceStartTime() {
+ return serviceStartTime;
+ }
+
+ public byte[] getUniqueId() {
+ return uniqueId;
+ }
+
+ public byte[] getPayload() {
+ return payload;
+ }
+
+ public byte[] getCommand() {
+ return command;
+ }
+
+ public byte[] getDomain() {
+ return domain;
+ }
+
+ public int getSecurePort() {
+ return securePort;
+ }
+
+ public void setMemberAliveTime(long time) {
+ memberAliveTime=time;
+ }
+
+
+
+ /**
+ * String representation of this object
+ */
+ public String toString() {
+ StringBuffer buf = new StringBuffer("org.apache.catalina.tribes.membership.MemberImpl[");
+ buf.append(getName()).append(",");
+ buf.append(getHostname()).append(",");
+ buf.append(port).append(", alive=");
+ buf.append(memberAliveTime).append(",");
+ buf.append("id=").append(bToS(this.uniqueId)).append(", ");
+ buf.append("payload=").append(bToS(this.payload,8)).append(", ");
+ buf.append("command=").append(bToS(this.command,8)).append(", ");
+ buf.append("domain=").append(bToS(this.domain,8)).append(", ");
+ buf.append("]");
+ return buf.toString();
+ }
+ public static String bToS(byte[] data) {
+ return bToS(data,data.length);
+ }
+ public static String bToS(byte[] data, int max) {
+ StringBuffer buf = new StringBuffer(4*16);
+ buf.append("{");
+ for (int i=0; data!=null && i<data.length; i++ ) {
+ buf.append(String.valueOf(data[i])).append(" ");
+ if ( i==max ) {
+ buf.append("...("+data.length+")");
+ break;
+ }
+ }
+ buf.append("}");
+ return buf.toString();
+ }
+
+ /**
+ * @see java.lang.Object#hashCode()
+ * @return The hash code
+ */
+ public int hashCode() {
+ return getHost()[0]+getHost()[1]+getHost()[2]+getHost()[3];
+ }
+
+ /**
+ * Returns true if the param o is a McastMember with the same name
+ * @param o
+ */
+ public boolean equals(Object o) {
+ if ( o instanceof MemberImpl ) {
+ return Arrays.equals(this.getHost(),((MemberImpl)o).getHost()) &&
+ this.getPort() == ((MemberImpl)o).getPort() &&
+ Arrays.equals(this.getUniqueId(),((MemberImpl)o).getUniqueId());
+ }
+ else
+ return false;
+ }
+
+ public void setHost(byte[] host) {
+ this.host = host;
+ }
+
+ public void setHostname(String host) throws IOException {
+ hostname = host;
+ this.host = java.net.InetAddress.getByName(host).getAddress();
+ }
+
+ public void setMsgCount(int msgCount) {
+ this.msgCount = msgCount;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ this.dataPkg = null;
+ }
+
+ public void setServiceStartTime(long serviceStartTime) {
+ this.serviceStartTime = serviceStartTime;
+ }
+
+ public void setUniqueId(byte[] uniqueId) {
+ this.uniqueId = uniqueId!=null?uniqueId:new byte[16];
+ getData(true,true);
+ }
+
+ public void setPayload(byte[] payload) {
+ byte[] oldpayload = this.payload;
+ this.payload = payload!=null?payload:new byte[0];
+ if ( this.getData(true,true).length > McastServiceImpl.MAX_PACKET_SIZE ) {
+ this.payload = oldpayload;
+ throw new IllegalArgumentException("Payload is to large for tribes to handle.");
+ }
+
+ }
+
+ public void setCommand(byte[] command) {
+ this.command = command!=null?command:new byte[0];
+ getData(true,true);
+ }
+
+ public void setDomain(byte[] domain) {
+ this.domain = domain!=null?domain:new byte[0];
+ getData(true,true);
+ }
+
+ public void setSecurePort(int securePort) {
+ this.securePort = securePort;
+ }
+
+ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ int length = in.readInt();
+ byte[] message = new byte[length];
+ in.read(message);
+ getMember(message,this);
+
+ }
+
+ public void writeExternal(ObjectOutput out) throws IOException {
+ byte[] data = this.getData();
+ out.writeInt(data.length);
+ out.write(data);
+ }
+
+}
private String host = "auto";
private InetAddress bind;
private int port = 4000;
+ private int securePort = -1;
private int rxBufSize = 43800;
private int txBufSize = 25188;
private boolean listen = false;
return useBufferPool;
}
+ public int getSecurePort() {
+ return securePort;
+ }
+
public void setTcpSelectorTimeout(long selTimeout) {
tcpSelectorTimeout = selTimeout;
}
this.useBufferPool = useBufferPool;
}
+ public void setSecurePort(int securePort) {
+ this.securePort = securePort;
+ }
+
public void heartbeat() {
//empty operation
}
--- /dev/null
+package org.apache.catalina.tribes.test;
+
+import java.io.IOException;
+import java.nio.channels.SelectionKey;
+import java.util.Iterator;
+import java.nio.channels.Selector;
+import org.apache.catalina.tribes.transport.nio.NioSender;
+import org.apache.catalina.tribes.membership.MemberImpl;
+import org.apache.catalina.tribes.io.ChannelData;
+import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.Channel;
+
+/**
+ * <p>Title: </p>
+ *
+ * <p>Description: </p>
+ *
+ * <p>Copyright: Copyright (c) 2005</p>
+ *
+ * <p>Company: </p>
+ *
+ * @author not attributable
+ * @version 1.0
+ */
+public class TestNioSender {
+ private Selector selector = null;
+ private int counter = 0;
+ MemberImpl mbr;
+ private static int testOptions = Channel.SEND_OPTIONS_DEFAULT;
+ public TestNioSender() {
+
+ }
+
+ public synchronized int inc() {
+ return ++counter;
+ }
+
+ public synchronized ChannelData getMessage(Member mbr) {
+ String msg = new String("Thread-"+Thread.currentThread().getName()+" Message:"+inc());
+ ChannelData data = new ChannelData(true);
+ data.setMessage(new XByteBuffer(msg.getBytes(),false));
+ data.setAddress(mbr);
+
+ return data;
+ }
+
+ public void init() throws Exception {
+ selector = Selector.open();
+ mbr = new MemberImpl("localhost",4444,0);
+ NioSender sender = new NioSender();
+ sender.setDestination(mbr);
+ sender.setDirectBuffer(true);
+ sender.setSelector(selector);
+ sender.setMessage(XByteBuffer.createDataPackage(getMessage(mbr)));
+ sender.connect();
+ }
+
+ public void run() {
+ while (true) {
+
+ int selectedKeys = 0;
+ try {
+ selectedKeys = selector.select(100);
+ // if ( selectedKeys == 0 ) {
+ // System.out.println("No registered interests. Sleeping for a second.");
+ // Thread.sleep(100);
+ } catch (Exception e) {
+ e.printStackTrace();
+ continue;
+ }
+
+ if (selectedKeys == 0) {
+ continue;
+ }
+
+ Iterator it = selector.selectedKeys().iterator();
+ while (it.hasNext()) {
+ SelectionKey sk = (SelectionKey) it.next();
+ it.remove();
+ try {
+ int readyOps = sk.readyOps();
+ sk.interestOps(sk.interestOps() & ~readyOps);
+ NioSender sender = (NioSender) sk.attachment();
+ if ( sender.process(sk, (testOptions&Channel.SEND_OPTIONS_USE_ACK)==Channel.SEND_OPTIONS_USE_ACK) ) {
+ System.out.println("Message completed for handler:"+sender);
+ Thread.currentThread().sleep(2000);
+ sender.reset();
+ sender.setMessage(XByteBuffer.createDataPackage(getMessage(mbr)));
+ }
+
+
+ } catch (Throwable t) {
+ t.printStackTrace();
+ return;
+ }
+ }
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ TestNioSender sender = new TestNioSender();
+ sender.init();
+ sender.run();
+ }
+}
--- /dev/null
+package org.apache.catalina.tribes.test;
+
+import junit.framework.Test;
+import junit.framework.TestCase;
+import junit.framework.TestSuite;
+
+public class TribesTestSuite
+ extends TestCase {
+
+ public TribesTestSuite(String s) {
+ super(s);
+ }
+
+ public static Test suite() {
+ TestSuite suite = new TestSuite();
+ suite.addTestSuite(org.apache.catalina.tribes.test.channel.ChannelStartStop.class);
+ suite.addTestSuite(org.apache.catalina.tribes.test.channel.TestChannelOptionFlag.class);
+ suite.addTestSuite(org.apache.catalina.tribes.test.membership.MemberSerialization.class);
+ suite.addTestSuite(org.apache.catalina.tribes.test.membership.TestMemberArrival.class);
+ suite.addTestSuite(org.apache.catalina.tribes.test.membership.TestTcpFailureDetector.class);
+ suite.addTestSuite(org.apache.catalina.tribes.test.channel.TestDataIntegrity.class);
+ return suite;
+ }
+}
--- /dev/null
+/*
+ * Copyright 1999,2004 The Apache Software Foundation.
+ *
+ * Licensed 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
+ */
+package org.apache.catalina.tribes.test.channel;
+
+import org.apache.catalina.tribes.group.GroupChannel;
+import junit.framework.TestCase;
+
+/**
+ * @author Filip Hanik
+ * @version 1.0
+ */
+public class ChannelStartStop extends TestCase {
+ GroupChannel channel = null;
+ protected void setUp() throws Exception {
+ super.setUp();
+ channel = new GroupChannel();
+ }
+
+ protected void tearDown() throws Exception {
+ super.tearDown();
+ try {channel.stop(channel.DEFAULT);}catch (Exception ignore){}
+ }
+
+ public void testDoubleFullStart() throws Exception {
+ int count = 0;
+ try {
+ channel.start(channel.DEFAULT);
+ count++;
+ } catch ( Exception x){x.printStackTrace();}
+ try {
+ channel.start(channel.DEFAULT);
+ count++;
+ } catch ( Exception x){x.printStackTrace();}
+ assertEquals(count,2);
+ channel.stop(channel.DEFAULT);
+ }
+
+ public void testDoublePartialStart() throws Exception {
+ //try to double start the RX
+ int count = 0;
+ try {
+ channel.start(channel.SND_RX_SEQ);
+ channel.start(channel.MBR_RX_SEQ);
+ count++;
+ } catch ( Exception x){x.printStackTrace();}
+ try {
+ channel.start(channel.MBR_RX_SEQ);
+ count++;
+ } catch ( Exception x){/*expected*/}
+ assertEquals(count,1);
+ channel.stop(channel.DEFAULT);
+ //double the membership sender
+ count = 0;
+ try {
+ channel.start(channel.SND_RX_SEQ);
+ channel.start(channel.MBR_TX_SEQ);
+ count++;
+ } catch ( Exception x){x.printStackTrace();}
+ try {
+ channel.start(channel.MBR_TX_SEQ);
+ count++;
+ } catch ( Exception x){/*expected*/}
+ assertEquals(count,1);
+ channel.stop(channel.DEFAULT);
+
+ count = 0;
+ try {
+ channel.start(channel.SND_RX_SEQ);
+ count++;
+ } catch ( Exception x){x.printStackTrace();}
+ try {
+ channel.start(channel.SND_RX_SEQ);
+ count++;
+ } catch ( Exception x){/*expected*/}
+ assertEquals(count,1);
+ channel.stop(channel.DEFAULT);
+
+ count = 0;
+ try {
+ channel.start(channel.SND_TX_SEQ);
+ count++;
+ } catch ( Exception x){x.printStackTrace();}
+ try {
+ channel.start(channel.SND_TX_SEQ);
+ count++;
+ } catch ( Exception x){/*expected*/}
+ assertEquals(count,1);
+ channel.stop(channel.DEFAULT);
+ }
+
+ public void testFalseOption() throws Exception {
+ int flag = 0xFFF0;//should get ignored by the underlying components
+ int count = 0;
+ try {
+ channel.start(flag);
+ count++;
+ } catch ( Exception x){x.printStackTrace();}
+ try {
+ channel.start(flag);
+ count++;
+ } catch ( Exception x){/*expected*/}
+ assertEquals(count,2);
+ channel.stop(channel.DEFAULT);
+ }
+
+}
--- /dev/null
+package org.apache.catalina.tribes.test.channel;
+
+import junit.framework.*;
+import org.apache.catalina.tribes.group.*;
+import org.apache.catalina.tribes.ChannelInterceptor;
+import org.apache.catalina.tribes.ChannelException;
+
+/**
+ * <p>Title: </p>
+ *
+ * <p>Description: </p>
+ *
+ * <p>Copyright: Copyright (c) 2005</p>
+ *
+ * <p>Company: </p>
+ *
+ * @author not attributable
+ * @version 1.0
+ */
+public class TestChannelOptionFlag extends TestCase {
+ GroupChannel channel = null;
+ protected void setUp() throws Exception {
+ super.setUp();
+ channel = new GroupChannel();
+ }
+
+ protected void tearDown() throws Exception {
+ super.tearDown();
+ if ( channel != null ) try {channel.stop(channel.DEFAULT);}catch ( Exception ignore) {}
+ channel = null;
+ }
+
+
+ public void testOptionConflict() throws Exception {
+ boolean error = false;
+ channel.setOptionCheck(true);
+ ChannelInterceptor i = new TestInterceptor();
+ i.setOptionFlag(128);
+ channel.addInterceptor(i);
+ i = new TestInterceptor();
+ i.setOptionFlag(128);
+ channel.addInterceptor(i);
+ try {
+ channel.start(channel.DEFAULT);
+ }catch ( ChannelException x ) {
+ if ( x.getMessage().indexOf("option flag conflict") >= 0 ) error = true;
+ }
+ assertEquals(true,error);
+ }
+
+ public void testOptionNoConflict() throws Exception {
+ boolean error = false;
+ channel.setOptionCheck(true);
+ ChannelInterceptor i = new TestInterceptor();
+ i.setOptionFlag(128);
+ channel.addInterceptor(i);
+ i = new TestInterceptor();
+ i.setOptionFlag(64);
+ channel.addInterceptor(i);
+ i = new TestInterceptor();
+ i.setOptionFlag(256);
+ channel.addInterceptor(i);
+ try {
+ channel.start(channel.DEFAULT);
+ }catch ( ChannelException x ) {
+ if ( x.getMessage().indexOf("option flag conflict") >= 0 ) error = true;
+ }
+ assertEquals(false,error);
+ }
+
+ public static class TestInterceptor extends ChannelInterceptorBase {
+
+ }
+
+
+}
--- /dev/null
+package org.apache.catalina.tribes.test.channel;
+
+import junit.framework.TestCase;
+import java.io.Serializable;
+import java.util.Random;
+import java.util.Arrays;
+import org.apache.catalina.tribes.ChannelListener;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.GroupChannel;
+import org.apache.catalina.tribes.test.channel.TestDataIntegrity.Listener;
+import org.apache.catalina.tribes.group.interceptors.MessageDispatchInterceptor;
+import org.apache.catalina.tribes.group.interceptors.MessageDispatch15Interceptor;
+
+/**
+ * <p>Title: </p>
+ *
+ * <p>Description: </p>
+ *
+ * <p>Copyright: Copyright (c) 2005</p>
+ *
+ * <p>Company: </p>
+ *
+ * @author not attributable
+ * @version 1.0
+ */
+public class TestDataIntegrity extends TestCase {
+ int msgCount = 500;
+ int threadCount = 20;
+ GroupChannel channel1;
+ GroupChannel channel2;
+ Listener listener1;
+ int threadCounter = 0;
+ protected void setUp() throws Exception {
+ super.setUp();
+ channel1 = new GroupChannel();
+ channel1.addInterceptor(new MessageDispatch15Interceptor());
+ channel2 = new GroupChannel();
+ channel2.addInterceptor(new MessageDispatch15Interceptor());
+ listener1 = new Listener();
+ channel2.addChannelListener(listener1);
+ channel1.start(GroupChannel.DEFAULT);
+ channel2.start(GroupChannel.DEFAULT);
+ }
+
+ protected void tearDown() throws Exception {
+ super.tearDown();
+ channel1.stop(GroupChannel.DEFAULT);
+ channel2.stop(GroupChannel.DEFAULT);
+ }
+
+ public void testDataSendNO_ACK() throws Exception {
+ System.err.println("Starting NO_ACK");
+ Thread[] threads = new Thread[threadCount];
+ for (int x=0; x<threads.length; x++ ) {
+ threads[x] = new Thread() {
+ public void run() {
+ try {
+ for (int i = 0; i < msgCount; i++) channel1.send(new Member[] {channel2.getLocalMember(false)}, Data.createRandomData(),0);
+ }catch ( Exception x ) {
+ x.printStackTrace();
+ return;
+ } finally {
+ threadCounter++;
+ }
+ }
+ };
+ }
+ for (int x=0; x<threads.length; x++ ) { threads[x].start();}
+ for (int x=0; x<threads.length; x++ ) { threads[x].join();}
+ //sleep for 50 sec, let the other messages in
+ long start = System.currentTimeMillis();
+ while ( (System.currentTimeMillis()-start)<30000 && msgCount*threadCount!=listener1.count) Thread.sleep(500);
+ System.err.println("Finished NO_ACK");
+ assertEquals("Checking success messages.",msgCount*threadCount,listener1.count);
+ }
+
+ public void testDataSendASYNCM() throws Exception {
+ System.err.println("Starting ASYNC MULTI THREAD");
+ Thread[] threads = new Thread[threadCount];
+ for (int x=0; x<threads.length; x++ ) {
+ threads[x] = new Thread() {
+ public void run() {
+ try {
+ for (int i = 0; i < msgCount; i++) channel1.send(new Member[] {channel2.getLocalMember(false)}, Data.createRandomData(),GroupChannel.SEND_OPTIONS_ASYNCHRONOUS);
+ }catch ( Exception x ) {
+ x.printStackTrace();
+ return;
+ } finally {
+ threadCounter++;
+ }
+ }
+ };
+ }
+ for (int x=0; x<threads.length; x++ ) { threads[x].start();}
+ for (int x=0; x<threads.length; x++ ) { threads[x].join();}
+ //sleep for 50 sec, let the other messages in
+ long start = System.currentTimeMillis();
+ while ( (System.currentTimeMillis()-start)<15000 && msgCount*threadCount!=listener1.count) Thread.sleep(500);
+ System.err.println("Finished ASYNC MULTI THREAD");
+ assertEquals("Checking success messages.",msgCount*threadCount,listener1.count);
+ }
+ public void testDataSendASYNC() throws Exception {
+ System.err.println("Starting ASYNC");
+ for (int i=0; i<msgCount; i++) channel1.send(new Member[] {channel2.getLocalMember(false)},Data.createRandomData(),GroupChannel.SEND_OPTIONS_ASYNCHRONOUS);
+ //sleep for 50 sec, let the other messages in
+ long start = System.currentTimeMillis();
+ while ( (System.currentTimeMillis()-start)<5000 && msgCount!=listener1.count) Thread.sleep(500);
+ System.err.println("Finished ASYNC");
+ assertEquals("Checking success messages.",msgCount,listener1.count);
+ }
+
+ public void testDataSendACK() throws Exception {
+ System.err.println("Starting ACK");
+ for (int i=0; i<msgCount; i++) channel1.send(new Member[] {channel2.getLocalMember(false)},Data.createRandomData(),GroupChannel.SEND_OPTIONS_USE_ACK);
+ Thread.sleep(250);
+ System.err.println("Finished ACK");
+ assertEquals("Checking success messages.",msgCount,listener1.count);
+ }
+
+ public void testDataSendSYNCACK() throws Exception {
+ System.err.println("Starting SYNC_ACK");
+ for (int i=0; i<msgCount; i++) channel1.send(new Member[] {channel2.getLocalMember(false)},Data.createRandomData(),GroupChannel.SEND_OPTIONS_SYNCHRONIZED_ACK|GroupChannel.SEND_OPTIONS_USE_ACK);
+ Thread.sleep(250);
+ System.err.println("Finished SYNC_ACK");
+ assertEquals("Checking success messages.",msgCount,listener1.count);
+ }
+
+ public static class Listener implements ChannelListener {
+ long count = 0;
+ public boolean accept(Serializable s, Member m) {
+ return (s instanceof Data);
+ }
+
+ public void messageReceived(Serializable s, Member m) {
+ Data d = (Data)s;
+ if ( !Data.verify(d) ) {
+ System.err.println("ERROR");
+ } else {
+ count++;
+ if ((count %1000) ==0 ) {
+ System.err.println("SUCCESS:"+count);
+ }
+ }
+ }
+ }
+
+ public static class Data implements Serializable {
+ public int length;
+ public byte[] data;
+ public byte key;
+ public static Random r = new Random(System.currentTimeMillis());
+ public static Data createRandomData() {
+ int i = r.nextInt();
+ i = ( i % 127 );
+ int length = Math.abs(r.nextInt() % 65555);
+ Data d = new Data();
+ d.length = length;
+ d.key = (byte)i;
+ d.data = new byte[length];
+ Arrays.fill(d.data,d.key);
+ return d;
+ }
+
+ public static boolean verify(Data d) {
+ boolean result = (d.length == d.data.length);
+ for ( int i=0; result && (i<d.data.length); i++ ) result = result && d.data[i] == d.key;
+ return result;
+ }
+ }
+
+
+
+}
--- /dev/null
+package org.apache.catalina.tribes.test.channel;\r
+\r
+import junit.framework.TestCase;\r
+import java.io.Serializable;\r
+import java.util.Random;\r
+import java.util.Arrays;\r
+import org.apache.catalina.tribes.ChannelListener;\r
+import org.apache.catalina.tribes.Member;\r
+import org.apache.catalina.tribes.group.GroupChannel;\r
+import java.io.PrintStream;\r
+\r
+/**\r
+ * <p>Title: </p> \r
+ * \r
+ * <p>Description: </p> \r
+ * \r
+ * <p>Copyright: Copyright (c) 2005</p> \r
+ * \r
+ * <p>Company: </p>\r
+ * \r
+ * @author not attributable\r
+ * @version 1.0\r
+ */\r
+public class TestRemoteProcessException extends TestCase {\r
+ int msgCount = 10000;\r
+ GroupChannel channel1;\r
+ GroupChannel channel2;\r
+ Listener listener1;\r
+ protected void setUp() throws Exception {\r
+ super.setUp();\r
+ channel1 = new GroupChannel();\r
+ channel2 = new GroupChannel();\r
+ listener1 = new Listener();\r
+ channel2.addChannelListener(listener1);\r
+ channel1.start(GroupChannel.DEFAULT);\r
+ channel2.start(GroupChannel.DEFAULT);\r
+ }\r
+\r
+ protected void tearDown() throws Exception {\r
+ super.tearDown();\r
+ channel1.stop(GroupChannel.DEFAULT);\r
+ channel2.stop(GroupChannel.DEFAULT);\r
+ }\r
+\r
+ public void testDataSendSYNCACK() throws Exception {\r
+ System.err.println("Starting SYNC_ACK");\r
+ int errC=0, nerrC=0;\r
+ for (int i=0; i<msgCount; i++) {\r
+ boolean error = Data.r.nextBoolean();\r
+ channel1.send(channel1.getMembers(),Data.createRandomData(error),GroupChannel.SEND_OPTIONS_SYNCHRONIZED_ACK|GroupChannel.SEND_OPTIONS_USE_ACK);\r
+ if ( error ) errC++; else nerrC++;\r
+ }\r
+ System.err.println("Finished SYNC_ACK");\r
+ assertEquals("Checking failure messages.",errC,listener1.errCnt);\r
+ assertEquals("Checking success messages.",nerrC,listener1.noErrCnt);\r
+ assertEquals("Checking all messages.",msgCount,listener1.noErrCnt+listener1.errCnt);\r
+ System.out.println("Listener 1 stats:");\r
+ listener1.printStats(System.out);\r
+ }\r
+\r
+ public static class Listener implements ChannelListener {\r
+ long noErrCnt = 0;\r
+ long errCnt = 0;\r
+ public boolean accept(Serializable s, Member m) {\r
+ return (s instanceof Data);\r
+ }\r
+\r
+ public void messageReceived(Serializable s, Member m) {\r
+ Data d = (Data)s;\r
+ if ( !Data.verify(d) ) {\r
+ System.err.println("ERROR");\r
+ } else {\r
+ if (d.error) {\r
+ errCnt++;\r
+ if ( (errCnt % 100) == 0) {\r
+ printStats(System.err);\r
+ }\r
+ throw new IllegalArgumentException();\r
+ } else {\r
+ noErrCnt++;\r
+ if ( (noErrCnt % 100) == 0) {\r
+ printStats(System.err);\r
+ }\r
+ }\r
+ }\r
+ }\r
+\r
+ public void printStats(PrintStream stream) {\r
+ stream.println("NORMAL:" + noErrCnt);\r
+ stream.println("FAILURES:" + errCnt);\r
+ stream.println("TOTAL:" + (errCnt+noErrCnt));\r
+ }\r
+ }\r
+\r
+ public static class Data implements Serializable {\r
+ public int length;\r
+ public byte[] data;\r
+ public byte key;\r
+ public boolean error = false;\r
+ public static Random r = new Random(System.currentTimeMillis());\r
+ public static Data createRandomData(boolean error) {\r
+ int i = r.nextInt();\r
+ i = ( i % 127 );\r
+ int length = Math.abs(r.nextInt() % 65555);\r
+ Data d = new Data();\r
+ d.length = length;\r
+ d.key = (byte)i;\r
+ d.data = new byte[length];\r
+ Arrays.fill(d.data,d.key);\r
+ d.error = error;\r
+ return d;\r
+ }\r
+\r
+ public static boolean verify(Data d) {\r
+ boolean result = (d.length == d.data.length);\r
+ for ( int i=0; result && (i<d.data.length); i++ ) result = result && d.data[i] == d.key;\r
+ return result;\r
+ }\r
+ }\r
+\r
+\r
+\r
+}\r
--- /dev/null
+package org.apache.catalina.tribes.test.interceptors;
+
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.group.GroupChannel;
+import org.apache.catalina.tribes.group.interceptors.NonBlockingCoordinator;
+import org.apache.catalina.tribes.group.interceptors.TcpFailureDetector;
+import junit.framework.TestCase;
+import junit.framework.TestResult;
+import junit.framework.TestSuite;
+
+public class TestNonBlockingCoordinator extends TestCase {
+
+ GroupChannel[] channels = null;
+ NonBlockingCoordinator[] coordinators = null;
+ int channelCount = 10;
+ Thread[] threads = null;
+ protected void setUp() throws Exception {
+ System.out.println("Setup");
+ super.setUp();
+ channels = new GroupChannel[channelCount];
+ coordinators = new NonBlockingCoordinator[channelCount];
+ threads = new Thread[channelCount];
+ for ( int i=0; i<channelCount; i++ ) {
+ channels[i] = new GroupChannel();
+ coordinators[i] = new NonBlockingCoordinator();
+ channels[i].addInterceptor(coordinators[i]);
+ channels[i].addInterceptor(new TcpFailureDetector());
+ final int j = i;
+ threads[i] = new Thread() {
+ public void run() {
+ try {
+ channels[j].start(Channel.DEFAULT);
+ Thread.sleep(50);
+ } catch (Exception x) {
+ x.printStackTrace();
+ }
+ }
+ };
+ }
+ for ( int i=0; i<channelCount; i++ ) threads[i].start();
+ for ( int i=0; i<channelCount; i++ ) threads[i].join();
+ Thread.sleep(1000);
+ }
+
+ public void testCoord1() throws Exception {
+ for (int i=1; i<channelCount; i++ )
+ assertEquals("Message count expected to be equal.",channels[i-1].getMembers().length,channels[i].getMembers().length);
+ Member member = coordinators[0].getCoordinator();
+ int cnt = 0;
+ while ( member == null && (cnt++ < 100 ) ) try {Thread.sleep(100); member = coordinators[0].getCoordinator();}catch ( Exception x){}
+ for (int i=0; i<channelCount; i++ ) super.assertEquals(member,coordinators[i].getCoordinator());
+ System.out.println("Coordinator[1] is:"+member);
+
+ }
+
+ public void testCoord2() throws Exception {
+ Member member = coordinators[1].getCoordinator();
+ System.out.println("Coordinator[2a] is:" + member);
+ int index = -1;
+ for ( int i=0; i<channelCount; i++ ) {
+ if ( channels[i].getLocalMember(false).equals(member) ) {
+ System.out.println("Shutting down:" + channels[i].getLocalMember(true).toString());
+ channels[i].stop(Channel.DEFAULT);
+ index = i;
+ }
+ }
+ int dead = index;
+ Thread.sleep(1000);
+ if ( index == 0 ) index = 1; else index = 0;
+ System.out.println("Member count:"+channels[index].getMembers().length);
+ member = coordinators[index].getCoordinator();
+ for (int i = 1; i < channelCount; i++) if ( i != dead ) super.assertEquals(member, coordinators[i].getCoordinator());
+ System.out.println("Coordinator[2b] is:" + member);
+ }
+
+ protected void tearDown() throws Exception {
+ System.out.println("tearDown");
+ super.tearDown();
+ for ( int i=0; i<channelCount; i++ ) {
+ channels[i].stop(Channel.DEFAULT);
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ TestSuite suite = new TestSuite();
+ suite.addTestSuite(TestNonBlockingCoordinator.class);
+ suite.run(new TestResult());
+ }
+
+
+
+
+
+}
--- /dev/null
+/*
+ * Copyright 1999,2004 The Apache Software Foundation.
+ *
+ * Licensed 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
+ */
+
+package org.apache.catalina.tribes.test.interceptors;
+
+import junit.framework.TestCase;
+
+/**
+ * <p>Title: </p>
+ *
+ * <p>Description: </p>
+ *
+ * <p>Copyright: Copyright (c) 2005</p>
+ *
+ * <p>Company: </p>
+ *
+ * @author not attributable
+ * @version 1.0
+ */
+public class TestTwoPhaseCommit extends TestCase {
+
+ protected void setUp() throws Exception {
+ super.setUp();
+ }
+
+ protected void tearDown() throws Exception {
+ super.tearDown();
+ }
+
+}
--- /dev/null
+package org.apache.catalina.tribes.test.io;
+
+import java.util.ArrayList;
+
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ManagedChannel;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.MembershipListener;
+import org.apache.catalina.tribes.group.GroupChannel;
+import junit.framework.TestCase;
+import org.apache.catalina.tribes.ChannelListener;
+import java.io.Serializable;
+import java.util.Random;
+import java.util.HashMap;
+import org.apache.catalina.tribes.transport.ReplicationTransmitter;
+
+public class TestSenderConnections extends TestCase {
+ private static int count = 2;
+ private ManagedChannel[] channels = new ManagedChannel[count];
+ private TestMsgListener[] listeners = new TestMsgListener[count];
+
+ protected void setUp() throws Exception {
+ super.setUp();
+ for (int i = 0; i < channels.length; i++) {
+ channels[i] = new GroupChannel();
+ channels[i].getMembershipService().setPayload( ("Channel-" + (i + 1)).getBytes("ASCII"));
+ listeners[i] = new TestMsgListener( ("Listener-" + (i + 1)));
+ channels[i].addChannelListener(listeners[i]);
+ channels[i].start(Channel.SND_RX_SEQ|Channel.SND_TX_SEQ);
+
+ }
+ }
+
+ public void clear() {
+ }
+
+ public void sendMessages(long delay, long sleep) throws Exception {
+ Member local = channels[0].getLocalMember(true);
+ Member dest = channels[1].getLocalMember(true);
+ int n = 3;
+ System.out.println("Sending " + n + " messages from [" + local.getName() + "] to [" + dest.getName() + "]");
+ for (int i = 0; i < n; i++) {
+ channels[0].send(new Member[] {dest}, new TestMsg(), 0);
+ if ( delay > 0 ) Thread.sleep(delay);
+ }
+ System.out.println("Messages sent. Sleeping for "+(sleep/1000)+" seconds to inspect connections");
+ if ( sleep > 0 ) Thread.sleep(sleep);
+
+ }
+
+ public void testConnectionLinger() throws Exception {
+ sendMessages(0,15000);
+ }
+
+ public void testKeepAliveCount() throws Exception {
+ System.out.println("Setting keep alive count to 0");
+ for (int i = 0; i < channels.length; i++) {
+ ReplicationTransmitter t = (ReplicationTransmitter)channels[0].getChannelSender();
+ t.getTransport().setKeepAliveCount(0);
+ }
+ sendMessages(1000,15000);
+ }
+
+ public void testKeepAliveTime() throws Exception {
+ System.out.println("Setting keep alive count to 1 second");
+ for (int i = 0; i < channels.length; i++) {
+ ReplicationTransmitter t = (ReplicationTransmitter)channels[0].getChannelSender();
+ t.getTransport().setKeepAliveTime(1000);
+ }
+ sendMessages(2000,15000);
+ }
+
+ protected void tearDown() throws Exception {
+ for (int i = 0; i < channels.length; i++) {
+ channels[i].stop(Channel.DEFAULT);
+ }
+
+ }
+
+ public static class TestMsg implements Serializable {
+ static Random r = new Random(System.currentTimeMillis());
+ HashMap map = new HashMap();
+ public TestMsg() {
+ int size = Math.abs(r.nextInt() % 200);
+ for (int i=0; i<size; i++ ) {
+ int length = Math.abs(r.nextInt() %65000);
+ ArrayList list = new ArrayList(length);
+ map.put(new Integer(i),list);
+ }
+ }
+ }
+
+ public class TestMsgListener implements ChannelListener {
+ public String name = null;
+ public TestMsgListener(String name) {
+ this.name = name;
+ }
+
+ public void messageReceived(Serializable msg, Member sender) {
+ System.out.println("["+name+"] Received message:"+msg+" from " + sender.getName());
+ }
+
+
+ public boolean accept(Serializable msg, Member sender) {
+ return true;
+ }
+
+
+
+ }
+
+}
--- /dev/null
+package org.apache.catalina.tribes.test.io;
+
+import org.apache.catalina.tribes.io.XByteBuffer;
+import junit.framework.TestCase;
+
+public class TestSerialization extends TestCase {
+ protected void setUp() throws Exception {
+ super.setUp();
+ }
+
+ public void testEmptyArray() throws Exception {
+
+ }
+
+ protected void tearDown() throws Exception {
+ super.tearDown();
+ }
+
+ public static void main(String[] args) throws Exception {
+ //XByteBuffer.deserialize(new byte[0]);
+ XByteBuffer.deserialize(new byte[] {-84, -19, 0, 5, 115, 114, 0, 17, 106});
+ }
+
+}
--- /dev/null
+/*
+ * Copyright 1999,2004 The Apache Software Foundation.
+ *
+ * Licensed 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
+ */
+package org.apache.catalina.tribes.test.membership;
+
+import junit.framework.TestCase;
+import org.apache.catalina.tribes.membership.MemberImpl;
+import java.util.Arrays;
+
+/**
+ * <p>Title: </p>
+ *
+ * <p>Description: </p>
+ *
+ * <p>Copyright: Copyright (c) 2005</p>
+ *
+ * <p>Company: </p>
+ *
+ * @author not attributable
+ * @version 1.0
+ */
+public class MemberSerialization extends TestCase {
+ MemberImpl m1, m2, p1,p2;
+ byte[] payload = null;
+ protected void setUp() throws Exception {
+ super.setUp();
+ payload = new byte[333];
+ Arrays.fill(payload,(byte)1);
+ m1 = new MemberImpl("localhost",3333,1,payload);
+ m2 = new MemberImpl("localhost",3333,1);
+ payload = new byte[333];
+ Arrays.fill(payload,(byte)2);
+ p1 = new MemberImpl("127.0.0.1",3333,1,payload);
+ p2 = new MemberImpl("localhost",3331,1,payload);
+ m1.setDomain(new byte[] {1,2,3,4,5,6,7,8,9});
+ m2.setDomain(new byte[] {1,2,3,4,5,6,7,8,9});
+ m1.setCommand(new byte[] {1,2,4,5,6,7,8,9});
+ m2.setCommand(new byte[] {1,2,4,5,6,7,8,9});
+ }
+
+ public void testCompare() throws Exception {
+ assertTrue(m1.equals(m2));
+ assertTrue(m2.equals(m1));
+ assertTrue(p1.equals(m2));
+ assertFalse(m1.equals(p2));
+ assertFalse(m1.equals(p2));
+ assertFalse(m2.equals(p2));
+ assertFalse(p1.equals(p2));
+ }
+
+ public void testSerializationOne() throws Exception {
+ MemberImpl m = m1;
+ byte[] md1 = m.getData(false,true);
+ byte[] mda1 = m.getData(false,false);
+ assertTrue(Arrays.equals(md1,mda1));
+ assertTrue(md1==mda1);
+ mda1 = m.getData(true,true);
+ MemberImpl ma1 = MemberImpl.getMember(mda1);
+ assertTrue(compareMembers(m,ma1));
+ mda1 = p1.getData(false);
+ assertFalse(Arrays.equals(md1,mda1));
+ ma1 = MemberImpl.getMember(mda1);
+ assertTrue(compareMembers(p1,ma1));
+
+ md1 = m.getData(true,true);
+ Thread.sleep(50);
+ mda1 = m.getData(true,true);
+ MemberImpl a1 = MemberImpl.getMember(md1);
+ MemberImpl a2 = MemberImpl.getMember(mda1);
+ assertTrue(a1.equals(a2));
+ assertFalse(Arrays.equals(md1,mda1));
+
+
+ }
+
+ public boolean compareMembers(MemberImpl impl1, MemberImpl impl2) {
+ boolean result = true;
+ result = result && Arrays.equals(impl1.getHost(),impl2.getHost());
+ result = result && Arrays.equals(impl1.getPayload(),impl2.getPayload());
+ result = result && Arrays.equals(impl1.getUniqueId(),impl2.getUniqueId());
+ result = result && impl1.getPort() == impl2.getPort();
+ return result;
+ }
+
+ protected void tearDown() throws Exception {
+ super.tearDown();
+ }
+
+}
--- /dev/null
+package org.apache.catalina.tribes.test.membership;\r
+\r
+import java.util.ArrayList;\r
+\r
+import org.apache.catalina.tribes.Channel;\r
+import org.apache.catalina.tribes.ManagedChannel;\r
+import org.apache.catalina.tribes.Member;\r
+import org.apache.catalina.tribes.MembershipListener;\r
+import org.apache.catalina.tribes.group.GroupChannel;\r
+import junit.framework.TestCase;\r
+import org.apache.catalina.tribes.group.interceptors.DomainFilterInterceptor;\r
+import org.apache.catalina.tribes.util.UUIDGenerator;\r
+\r
+public class TestDomainFilter\r
+ extends TestCase {\r
+ private static int count = 10;\r
+ private ManagedChannel[] channels = new ManagedChannel[count];\r
+ private TestMbrListener[] listeners = new TestMbrListener[count];\r
+\r
+ protected void setUp() throws Exception {\r
+ super.setUp();\r
+ for (int i = 0; i < channels.length; i++) {\r
+ channels[i] = new GroupChannel();\r
+ channels[i].getMembershipService().setPayload( ("Channel-" + (i + 1)).getBytes("ASCII"));\r
+ listeners[i] = new TestMbrListener( ("Listener-" + (i + 1)));\r
+ channels[i].addMembershipListener(listeners[i]);\r
+ DomainFilterInterceptor filter = new DomainFilterInterceptor();\r
+ filter.setDomain(UUIDGenerator.randomUUID(false));\r
+ channels[i].addInterceptor(filter);\r
+ }\r
+ }\r
+\r
+ public void clear() {\r
+ for (int i = 0; i < channels.length; i++) {\r
+ listeners[i].members.clear();\r
+ }\r
+ }\r
+\r
+ public void testMemberArrival() throws Exception {\r
+ //purpose of this test is to make sure that we have received all the members\r
+ //that we can expect before the start method returns\r
+ Thread[] threads = new Thread[channels.length];\r
+ for (int i=0; i<channels.length; i++ ) {\r
+ final Channel channel = channels[i];\r
+ Thread t = new Thread() {\r
+ public void run() {\r
+ try {\r
+ channel.start(Channel.DEFAULT);\r
+ }catch ( Exception x ) {\r
+ throw new RuntimeException(x);\r
+ }\r
+ }\r
+ };\r
+ threads[i] = t;\r
+ }\r
+ for (int i=0; i<threads.length; i++ ) threads[i].start();\r
+ for (int i=0; i<threads.length; i++ ) threads[i].join();\r
+ System.out.println("All channels started.");\r
+ for (int i=listeners.length-1; i>=0; i-- ) assertEquals("Checking member arrival length",0,listeners[i].members.size());\r
+ }\r
+\r
+ protected void tearDown() throws Exception {\r
+\r
+ for (int i = 0; i < channels.length; i++) {\r
+ try {\r
+ channels[i].stop(Channel.DEFAULT);\r
+ } catch (Exception ignore) {}\r
+ }\r
+ super.tearDown();\r
+ }\r
+\r
+ public class TestMbrListener\r
+ implements MembershipListener {\r
+ public String name = null;\r
+ public TestMbrListener(String name) {\r
+ this.name = name;\r
+ }\r
+\r
+ public ArrayList members = new ArrayList();\r
+ public void memberAdded(Member member) {\r
+ if (!members.contains(member)) {\r
+ members.add(member);\r
+ try {\r
+ System.out.println(name + ":member added[" + new String(member.getPayload(), "ASCII") + "; Thread:"+Thread.currentThread().getName()+"]");\r
+ } catch (Exception x) {\r
+ System.out.println(name + ":member added[unknown]");\r
+ }\r
+ }\r
+ }\r
+\r
+ public void memberDisappeared(Member member) {\r
+ if (members.contains(member)) {\r
+ members.remove(member);\r
+ try {\r
+ System.out.println(name + ":member disappeared[" + new String(member.getPayload(), "ASCII") + "; Thread:"+Thread.currentThread().getName()+"]");\r
+ } catch (Exception x) {\r
+ System.out.println(name + ":member disappeared[unknown]");\r
+ }\r
+ }\r
+ }\r
+\r
+ }\r
+\r
+}\r
--- /dev/null
+package org.apache.catalina.tribes.test.membership;
+
+import java.util.ArrayList;
+
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ManagedChannel;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.MembershipListener;
+import org.apache.catalina.tribes.group.GroupChannel;
+import junit.framework.TestCase;
+
+public class TestMemberArrival
+ extends TestCase {
+ private static int count = 10;
+ private ManagedChannel[] channels = new ManagedChannel[count];
+ private TestMbrListener[] listeners = new TestMbrListener[count];
+
+ protected void setUp() throws Exception {
+ super.setUp();
+ for (int i = 0; i < channels.length; i++) {
+ channels[i] = new GroupChannel();
+ channels[i].getMembershipService().setPayload( ("Channel-" + (i + 1)).getBytes("ASCII"));
+ listeners[i] = new TestMbrListener( ("Listener-" + (i + 1)));
+ channels[i].addMembershipListener(listeners[i]);
+
+ }
+ }
+
+ public void clear() {
+ for (int i = 0; i < channels.length; i++) {
+ listeners[i].members.clear();
+ }
+ }
+
+ public void testMemberArrival() throws Exception {
+ //purpose of this test is to make sure that we have received all the members
+ //that we can expect before the start method returns
+ Thread[] threads = new Thread[channels.length];
+ for (int i=0; i<channels.length; i++ ) {
+ final Channel channel = channels[i];
+ Thread t = new Thread() {
+ public void run() {
+ try {
+ channel.start(Channel.DEFAULT);
+ }catch ( Exception x ) {
+ throw new RuntimeException(x);
+ }
+ }
+ };
+ threads[i] = t;
+ }
+ for (int i=0; i<threads.length; i++ ) threads[i].start();
+ for (int i=0; i<threads.length; i++ ) threads[i].join();
+ System.out.println("All channels started.");
+ for (int i=listeners.length-1; i>=0; i-- ) assertEquals("Checking member arrival length",channels.length-1,listeners[i].members.size());
+ }
+
+ protected void tearDown() throws Exception {
+
+ for (int i = 0; i < channels.length; i++) {
+ try {
+ channels[i].stop(Channel.DEFAULT);
+ } catch (Exception ignore) {}
+ }
+ super.tearDown();
+ }
+
+ public class TestMbrListener
+ implements MembershipListener {
+ public String name = null;
+ public TestMbrListener(String name) {
+ this.name = name;
+ }
+
+ public ArrayList members = new ArrayList();
+ public void memberAdded(Member member) {
+ if (!members.contains(member)) {
+ members.add(member);
+ try {
+ System.out.println(name + ":member added[" + new String(member.getPayload(), "ASCII") + "; Thread:"+Thread.currentThread().getName()+"]");
+ } catch (Exception x) {
+ System.out.println(name + ":member added[unknown]");
+ }
+ }
+ }
+
+ public void memberDisappeared(Member member) {
+ if (members.contains(member)) {
+ members.remove(member);
+ try {
+ System.out.println(name + ":member disappeared[" + new String(member.getPayload(), "ASCII") + "; Thread:"+Thread.currentThread().getName()+"]");
+ } catch (Exception x) {
+ System.out.println(name + ":member disappeared[unknown]");
+ }
+ }
+ }
+
+ }
+
+}
--- /dev/null
+package org.apache.catalina.tribes.test.membership;
+
+import java.util.ArrayList;
+
+import org.apache.catalina.tribes.ByteMessage;
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.ManagedChannel;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.MembershipListener;
+import org.apache.catalina.tribes.group.GroupChannel;
+import org.apache.catalina.tribes.group.interceptors.TcpFailureDetector;
+import junit.framework.TestCase;
+
+/**
+ * <p>Title: </p>
+ *
+ * <p>Description: </p>
+ *
+ * <p>Copyright: Copyright (c) 2005</p>
+ *
+ * <p>Company: </p>
+ *
+ * @author not attributable
+ * @version 1.0
+ */
+public class TestTcpFailureDetector extends TestCase {
+ private TcpFailureDetector tcpFailureDetector1 = null;
+ private TcpFailureDetector tcpFailureDetector2 = null;
+ private ManagedChannel channel1 = null;
+ private ManagedChannel channel2 = null;
+ private TestMbrListener mbrlist1 = null;
+ private TestMbrListener mbrlist2 = null;
+ protected void setUp() throws Exception {
+ super.setUp();
+ channel1 = new GroupChannel();
+ channel2 = new GroupChannel();
+ channel1.getMembershipService().setPayload("Channel-1".getBytes("ASCII"));
+ channel2.getMembershipService().setPayload("Channel-2".getBytes("ASCII"));
+ mbrlist1 = new TestMbrListener("Channel-1");
+ mbrlist2 = new TestMbrListener("Channel-2");
+ tcpFailureDetector1 = new TcpFailureDetector();
+ tcpFailureDetector2 = new TcpFailureDetector();
+ channel1.addInterceptor(tcpFailureDetector1);
+ channel2.addInterceptor(tcpFailureDetector2);
+ channel1.addMembershipListener(mbrlist1);
+ channel2.addMembershipListener(mbrlist2);
+ }
+
+ public void clear() {
+ mbrlist1.members.clear();
+ mbrlist2.members.clear();
+ }
+
+ public void testTcpSendFailureMemberDrop() throws Exception {
+ System.out.println("testTcpSendFailureMemberDrop()");
+ clear();
+ channel1.start(channel1.DEFAULT);
+ channel2.start(channel2.DEFAULT);
+ //Thread.sleep(1000);
+ assertEquals("Expecting member count to be equal",mbrlist1.members.size(),mbrlist2.members.size());
+ channel2.stop(channel2.SND_RX_SEQ);
+ ByteMessage msg = new ByteMessage(new byte[1024]);
+ try {
+ channel1.send(channel1.getMembers(), msg, 0);
+ assertEquals("Message send should have failed.",true,false);
+ } catch ( ChannelException x ) {
+
+ }
+ assertEquals("Expecting member count to not be equal",mbrlist1.members.size()+1,mbrlist2.members.size());
+ channel1.stop(Channel.DEFAULT);
+ channel2.stop(Channel.DEFAULT);
+ }
+
+ public void testTcpFailureMemberAdd() throws Exception {
+ System.out.println("testTcpFailureMemberAdd()");
+ clear();
+ channel1.start(channel1.DEFAULT);
+ channel2.start(channel2.SND_RX_SEQ);
+ channel2.start(channel2.SND_TX_SEQ);
+ channel2.start(channel2.MBR_RX_SEQ);
+ channel2.stop(channel2.SND_RX_SEQ);
+ channel2.start(channel2.MBR_TX_SEQ);
+ //Thread.sleep(1000);
+ assertEquals("Expecting member count to not be equal",mbrlist1.members.size()+1,mbrlist2.members.size());
+ channel1.stop(Channel.DEFAULT);
+ channel2.stop(Channel.DEFAULT);
+ }
+
+ public void testTcpMcastFail() throws Exception {
+ System.out.println("testTcpMcastFail()");
+ clear();
+ channel1.start(channel1.DEFAULT);
+ channel2.start(channel2.DEFAULT);
+ //Thread.sleep(1000);
+ assertEquals("Expecting member count to be equal",mbrlist1.members.size(),mbrlist2.members.size());
+ channel2.stop(channel2.MBR_TX_SEQ);
+ ByteMessage msg = new ByteMessage(new byte[1024]);
+ try {
+ Thread.sleep(5000);
+ assertEquals("Expecting member count to be equal",mbrlist1.members.size(),mbrlist2.members.size());
+ channel1.send(channel1.getMembers(), msg, 0);
+ } catch ( ChannelException x ) {
+ assertEquals("Message send should have succeeded.",true,false);
+ }
+ channel1.stop(Channel.DEFAULT);
+ channel2.stop(Channel.DEFAULT);
+ }
+
+
+ protected void tearDown() throws Exception {
+ tcpFailureDetector1 = null;
+ tcpFailureDetector2 = null;
+ try { channel1.stop(Channel.DEFAULT);}catch (Exception ignore){}
+ channel1 = null;
+ try { channel2.stop(Channel.DEFAULT);}catch (Exception ignore){}
+ channel2 = null;
+ super.tearDown();
+ }
+
+ public class TestMbrListener implements MembershipListener {
+ public String name = null;
+ public TestMbrListener(String name) {
+ this.name = name;
+ }
+ public ArrayList members = new ArrayList();
+ public void memberAdded(Member member) {
+ if ( !members.contains(member) ) {
+ members.add(member);
+ try{
+ System.out.println(name + ":member added[" + new String(member.getPayload(), "ASCII") + "]");
+ }catch ( Exception x ) {
+ System.out.println(name + ":member added[unknown]");
+ }
+ }
+ }
+
+ public void memberDisappeared(Member member) {
+ if ( members.contains(member) ) {
+ members.remove(member);
+ try{
+ System.out.println(name + ":member disappeared[" + new String(member.getPayload(), "ASCII") + "]");
+ }catch ( Exception x ) {
+ System.out.println(name + ":member disappeared[unknown]");
+ }
+ }
+ }
+
+ }
+
+}
--- /dev/null
+package org.apache.catalina.tribes.test.transport;
+
+import java.text.DecimalFormat;
+
+import org.apache.catalina.tribes.ChannelMessage;
+import org.apache.catalina.tribes.Member;
+import org.apache.catalina.tribes.MessageListener;
+import org.apache.catalina.tribes.io.ChannelData;
+import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.membership.MemberImpl;
+import org.apache.catalina.tribes.transport.nio.NioReceiver;
+
+public class SocketNioReceive {
+ static int count = 0;
+ static int accept = 0;
+ static long start = 0;
+ static double mb = 0;
+ static int len = 0;
+ static DecimalFormat df = new DecimalFormat("##.00");
+ static double seconds = 0;
+
+ protected static Object mutex = new Object();
+ public static void main(String[] args) throws Exception {
+ Member mbr = new MemberImpl("localhost", 9999, 0);
+ ChannelData data = new ChannelData();
+ data.setAddress(mbr);
+ byte[] buf = new byte[8192 * 4];
+ data.setMessage(new XByteBuffer(buf, false));
+ buf = XByteBuffer.createDataPackage(data);
+ len = buf.length;
+ NioReceiver receiver = new NioReceiver();
+ receiver.setPort(9999);
+ receiver.setHost("localhost");
+ MyList list = new MyList();
+ receiver.setMessageListener(list);
+ receiver.start();
+ System.out.println("Listening on 9999");
+ while (true) {
+ try {
+ synchronized (mutex) {
+ mutex.wait(5000);
+ if ( start != 0 ) {
+ System.out.println("Throughput " + df.format(mb / seconds) + " MB/seconds, messages "+count+" accepts "+accept+", total "+mb+" MB.");
+ }
+ }
+ }catch (Throwable x) {
+ x.printStackTrace();
+ }
+ }
+ }
+
+ public static class MyList implements MessageListener {
+ boolean first = true;
+
+
+ public void messageReceived(ChannelMessage msg) {
+ if (first) {
+ first = false;
+ start = System.currentTimeMillis();
+ }
+ mb += ( (double) len) / 1024 / 1024;
+ synchronized (this) {count++;}
+ if ( ( (count) % 10000) == 0) {
+ long time = System.currentTimeMillis();
+ seconds = ( (double) (time - start)) / 1000;
+ System.out.println("Throughput " + df.format(mb / seconds) + " MB/seconds, messages "+count+", total "+mb+" MB.");
+ }
+ }
+
+ public boolean accept(ChannelMessage msg) {
+ synchronized (this) {accept++;}
+ return true;
+ }
+
+ }
+}
--- /dev/null
+package org.apache.catalina.tribes.test.transport;
+
+import java.io.OutputStream;
+import java.net.Socket;
+import java.text.DecimalFormat;
+import org.apache.catalina.tribes.transport.nio.NioSender;
+import org.apache.catalina.tribes.membership.MemberImpl;
+import java.nio.channels.Selector;
+import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.Member;
+import java.nio.channels.SelectionKey;
+import java.util.Iterator;
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.io.ChannelData;
+import java.math.BigDecimal;
+
+public class SocketNioSend {
+
+ public static void main(String[] args) throws Exception {
+ Selector selector = Selector.open();
+ Member mbr = new MemberImpl("localhost", 9999, 0);
+ ChannelData data = new ChannelData();
+ data.setOptions(Channel.SEND_OPTIONS_BYTE_MESSAGE);
+ data.setAddress(mbr);
+ byte[] buf = new byte[8192 * 4];
+ data.setMessage(new XByteBuffer(buf,false));
+ buf = XByteBuffer.createDataPackage(data);
+ int len = buf.length;
+ BigDecimal total = new BigDecimal((double)0);
+ BigDecimal bytes = new BigDecimal((double)len);
+ NioSender sender = new NioSender();
+ sender.setDestination(mbr);
+ sender.setDirectBuffer(true);
+ sender.setSelector(selector);
+ sender.setTxBufSize(1024*1024);
+ sender.connect();
+ sender.setMessage(buf);
+ System.out.println("Writing to 9999");
+ long start = 0;
+ double mb = 0;
+ boolean first = true;
+ int count = 0;
+ DecimalFormat df = new DecimalFormat("##.00");
+ while (count<100000) {
+ if (first) {
+ first = false;
+ start = System.currentTimeMillis();
+ }
+ sender.setMessage(buf);
+ int selectedKeys = 0;
+ try {
+ selectedKeys = selector.select(0);
+ } catch (Exception e) {
+ e.printStackTrace();
+ continue;
+ }
+
+ if (selectedKeys == 0) {
+ continue;
+ }
+
+ Iterator it = selector.selectedKeys().iterator();
+ while (it.hasNext()) {
+ SelectionKey sk = (SelectionKey) it.next();
+ it.remove();
+ try {
+ int readyOps = sk.readyOps();
+ sk.interestOps(sk.interestOps() & ~readyOps);
+ if (sender.process(sk, false)) {
+ total = total.add(bytes);
+ sender.reset();
+ sender.setMessage(buf);
+ mb += ( (double) len) / 1024 / 1024;
+ if ( ( (++count) % 10000) == 0) {
+ long time = System.currentTimeMillis();
+ double seconds = ( (double) (time - start)) / 1000;
+ System.out.println("Throughput " + df.format(mb / seconds) + " MB/seconds, total "+mb+" MB, total "+total+" bytes.");
+ }
+ }
+
+ } catch (Throwable t) {
+ t.printStackTrace();
+ return;
+ }
+ }
+ selector.selectedKeys().clear();
+ }
+ System.out.println("Complete, sleeping 15 seconds");
+ Thread.sleep(15000);
+ }
+}
--- /dev/null
+package org.apache.catalina.tribes.test.transport;
+
+import java.io.OutputStream;
+import java.net.Socket;
+import java.text.DecimalFormat;
+import org.apache.catalina.tribes.transport.nio.NioSender;
+import org.apache.catalina.tribes.membership.MemberImpl;
+import java.nio.channels.Selector;
+import org.apache.catalina.tribes.io.XByteBuffer;
+import org.apache.catalina.tribes.Member;
+import java.nio.channels.SelectionKey;
+import java.util.Iterator;
+import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.io.ChannelData;
+import java.math.BigDecimal;
+import java.util.Arrays;
+
+public class SocketNioValidateSend {
+
+ public static void main(String[] args) throws Exception {
+ Selector selector = Selector.open();
+ Member mbr = new MemberImpl("localhost", 9999, 0);
+ byte seq = 0;
+ byte[] buf = new byte[50000];
+ Arrays.fill(buf,seq);
+ int len = buf.length;
+ BigDecimal total = new BigDecimal((double)0);
+ BigDecimal bytes = new BigDecimal((double)len);
+ NioSender sender = new NioSender();
+ sender.setDestination(mbr);
+ sender.setDirectBuffer(true);
+ sender.setSelector(selector);
+ sender.connect();
+ sender.setMessage(buf);
+ System.out.println("Writing to 9999");
+ long start = 0;
+ double mb = 0;
+ boolean first = true;
+ int count = 0;
+
+ DecimalFormat df = new DecimalFormat("##.00");
+ while (count<100000) {
+ if (first) {
+ first = false;
+ start = System.currentTimeMillis();
+ }
+ sender.setMessage(buf);
+ int selectedKeys = 0;
+ try {
+ selectedKeys = selector.select(0);
+ } catch (Exception e) {
+ e.printStackTrace();
+ continue;
+ }
+
+ if (selectedKeys == 0) {
+ continue;
+ }
+
+ Iterator it = selector.selectedKeys().iterator();
+ while (it.hasNext()) {
+ SelectionKey sk = (SelectionKey) it.next();
+ it.remove();
+ try {
+ int readyOps = sk.readyOps();
+ sk.interestOps(sk.interestOps() & ~readyOps);
+ if (sender.process(sk, false)) {
+ total = total.add(bytes);
+ sender.reset();
+ seq++;
+ Arrays.fill(buf,seq);
+ sender.setMessage(buf);
+ mb += ( (double) len) / 1024 / 1024;
+ if ( ( (++count) % 10000) == 0) {
+ long time = System.currentTimeMillis();
+ double seconds = ( (double) (time - start)) / 1000;
+ System.out.println("Throughput " + df.format(mb / seconds) + " MB/seconds, total "+mb+" MB, total "+total+" bytes.");
+ }
+ }
+
+ } catch (Throwable t) {
+ t.printStackTrace();
+ return;
+ }
+ }
+ }
+ System.out.println("Complete, sleeping 15 seconds");
+ Thread.sleep(15000);
+ }
+}
--- /dev/null
+package org.apache.catalina.tribes.test.transport;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.io.InputStream;
+import java.text.DecimalFormat;
+import java.math.BigDecimal;
+
+public class SocketReceive {
+ static long start = 0;
+ static double mb = 0;
+ static byte[] buf = new byte[8192 * 4];
+ static boolean first = true;
+ static int count = 0;
+ static DecimalFormat df = new DecimalFormat("##.00");
+ static BigDecimal total = new BigDecimal(0);
+ static BigDecimal bytes = new BigDecimal(32871);
+
+
+ public static void main(String[] args) throws Exception {
+
+ ServerSocket srvSocket = new ServerSocket(9999);
+ System.out.println("Listening on 9999");
+ Socket socket = srvSocket.accept();
+ socket.setReceiveBufferSize(43800);
+ InputStream in = socket.getInputStream();
+ Thread t = new Thread() {
+ public void run() {
+ while ( true ) {
+ try {
+ Thread.sleep(1000);
+ printStats(start, mb, count, df, total);
+ }catch ( Exception x ) {}
+ }
+ }
+ };
+ t.setDaemon(true);
+ t.start();
+
+ while ( true ) {
+ if ( first ) { first = false; start = System.currentTimeMillis();}
+ int len = in.read(buf);
+ if ( len == -1 ) {
+ printStats(start, mb, count, df, total);
+ System.exit(1);
+ }
+ if ( bytes.intValue() != len ) bytes = new BigDecimal((double)len);
+ total = total.add(bytes);
+ mb += ( (double) len) / 1024 / 1024;
+ if ( ((++count) % 10000) == 0 ) {
+ printStats(start, mb, count, df, total);
+ }
+ }
+
+ }
+
+ private static void printStats(long start, double mb, int count, DecimalFormat df, BigDecimal total) {
+ long time = System.currentTimeMillis();
+ double seconds = ((double)(time-start))/1000;
+ System.out.println("Throughput "+df.format(mb/seconds)+" MB/seconds messages "+count+", total "+mb+" MB, total "+total+" bytes.");
+ }
+}
\ No newline at end of file
--- /dev/null
+package org.apache.catalina.tribes.test.transport;\r
+\r
+import java.io.OutputStream;\r
+import java.net.Socket;\r
+import java.text.DecimalFormat;\r
+import org.apache.catalina.tribes.membership.MemberImpl;\r
+import org.apache.catalina.tribes.io.XByteBuffer;\r
+import org.apache.catalina.tribes.Member;\r
+import org.apache.catalina.tribes.io.ChannelData;\r
+import org.apache.catalina.tribes.Channel;\r
+import java.math.BigDecimal;\r
+\r
+public class SocketSend {\r
+\r
+ public static void main(String[] args) throws Exception {\r
+ \r
+ \r
+ Member mbr = new MemberImpl("localhost", 9999, 0);\r
+ ChannelData data = new ChannelData();\r
+ data.setOptions(Channel.SEND_OPTIONS_BYTE_MESSAGE);\r
+ data.setAddress(mbr);\r
+ byte[] buf = new byte[8192 * 4];\r
+ data.setMessage(new XByteBuffer(buf,false));\r
+ buf = XByteBuffer.createDataPackage(data);\r
+ int len = buf.length;\r
+ System.out.println("Message size:"+len+" bytes");\r
+ BigDecimal total = new BigDecimal((double)0);\r
+ BigDecimal bytes = new BigDecimal((double)len);\r
+ Socket socket = new Socket("localhost",9999);\r
+ System.out.println("Writing to 9999");\r
+ OutputStream out = socket.getOutputStream();\r
+ long start = 0;\r
+ double mb = 0;\r
+ boolean first = true;\r
+ int count = 0;\r
+ DecimalFormat df = new DecimalFormat("##.00");\r
+ while ( count<100000 ) {\r
+ if ( first ) { first = false; start = System.currentTimeMillis();}\r
+ out.write(buf,0,buf.length);\r
+ mb += ( (double) buf.length) / 1024 / 1024;\r
+ total = total.add(bytes);\r
+ if ( ((++count) % 10000) == 0 ) {\r
+ long time = System.currentTimeMillis();\r
+ double seconds = ((double)(time-start))/1000;\r
+ System.out.println("Throughput "+df.format(mb/seconds)+" MB/seconds messages "+count+", total "+mb+" MB, total "+total+" bytes.");\r
+ }\r
+ }\r
+ out.flush(); \r
+ System.out.println("Complete, sleeping 5 seconds");\r
+ Thread.sleep(5000);\r
+\r
+ }\r
+}\r
--- /dev/null
+package org.apache.catalina.tribes.test.transport;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.io.InputStream;
+import java.text.DecimalFormat;
+import java.math.BigDecimal;
+import org.apache.catalina.tribes.io.XByteBuffer;
+
+public class SocketTribesReceive {
+ static long start = 0;
+ static double mb = 0;
+ //static byte[] buf = new byte[32871];
+ static byte[] buf = new byte[32871];
+ static boolean first = true;
+ static int count = 0;
+ static DecimalFormat df = new DecimalFormat("##.00");
+ static BigDecimal total = new BigDecimal((double)0);
+ static BigDecimal bytes = new BigDecimal((double)32871);
+
+
+ public static void main(String[] args) throws Exception {
+ int size = 43800;
+ if (args.length > 0 ) try {size=Integer.parseInt(args[0]);}catch(Exception x){}
+ XByteBuffer xbuf = new XByteBuffer(43800,true);
+ ServerSocket srvSocket = new ServerSocket(9999);
+ System.out.println("Listening on 9999");
+ Socket socket = srvSocket.accept();
+ socket.setReceiveBufferSize(size);
+ InputStream in = socket.getInputStream();
+ Thread t = new Thread() {
+ public void run() {
+ while ( true ) {
+ try {
+ Thread.sleep(1000);
+ printStats(start, mb, count, df, total);
+ }catch ( Exception x ) {}
+ }
+ }
+ };
+ t.setDaemon(true);
+ t.start();
+
+ while ( true ) {
+ if ( first ) { first = false; start = System.currentTimeMillis();}
+ int len = in.read(buf);
+ if ( len == -1 ) {
+ printStats(start, mb, count, df, total);
+ System.exit(1);
+ }
+ xbuf.append(buf,0,len);
+ if ( bytes.intValue() != len ) bytes = new BigDecimal((double)len);
+ total = total.add(bytes);
+ while ( xbuf.countPackages(true) > 0 ) {
+ xbuf.extractPackage(true);
+ count++;
+ }
+ mb += ( (double) len) / 1024 / 1024;
+ if ( ((count) % 10000) == 0 ) {
+ printStats(start, mb, count, df, total);
+ }
+ }
+
+ }
+
+ private static void printStats(long start, double mb, int count, DecimalFormat df, BigDecimal total) {
+ long time = System.currentTimeMillis();
+ double seconds = ((double)(time-start))/1000;
+ System.out.println("Throughput "+df.format(mb/seconds)+" MB/seconds messages "+count+", total "+mb+" MB, total "+total+" bytes.");
+ }
+}
\ No newline at end of file
--- /dev/null
+package org.apache.catalina.tribes.test.transport;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.io.InputStream;
+import java.text.DecimalFormat;
+import java.math.BigDecimal;
+
+public class SocketValidateReceive {
+ static long start = 0;
+ static double mb = 0;
+ static byte[] buf = new byte[8192 * 4];
+ static boolean first = true;
+ static int count = 0;
+ static DecimalFormat df = new DecimalFormat("##.00");
+ static BigDecimal total = new BigDecimal(0);
+ static BigDecimal bytes = new BigDecimal(32871);
+
+
+ public static void main(String[] args) throws Exception {
+ int size = 43800;
+ if (args.length > 0 ) try {size=Integer.parseInt(args[0]);}catch(Exception x){}
+
+ ServerSocket srvSocket = new ServerSocket(9999);
+ System.out.println("Listening on 9999");
+ Socket socket = srvSocket.accept();
+ socket.setReceiveBufferSize(size);
+ InputStream in = socket.getInputStream();
+ MyDataReader reader = new MyDataReader(50000);
+ Thread t = new Thread() {
+ public void run() {
+ while ( true ) {
+ try {
+ Thread.sleep(1000);
+ printStats(start, mb, count, df, total);
+ }catch ( Exception x ) {}
+ }
+ }
+ };
+ t.setDaemon(true);
+ t.start();
+
+ while ( true ) {
+ if ( first ) { first = false; start = System.currentTimeMillis();}
+ int len = in.read(buf);
+ if ( len == -1 ) {
+ printStats(start, mb, count, df, total);
+ System.exit(1);
+ }
+ count += reader.append(buf,0,len);
+
+ if ( bytes.intValue() != len ) bytes = new BigDecimal((double)len);
+ total = total.add(bytes);
+ mb += ( (double) len) / 1024 / 1024;
+ if ( ((count) % 10000) == 0 ) {
+ printStats(start, mb, count, df, total);
+ }
+ }
+
+ }
+
+ private static void printStats(long start, double mb, int count, DecimalFormat df, BigDecimal total) {
+ long time = System.currentTimeMillis();
+ double seconds = ((double)(time-start))/1000;
+ System.out.println("Throughput "+df.format(mb/seconds)+" MB/seconds messages "+count+", total "+mb+" MB, total "+total+" bytes.");
+ }
+
+ public static class MyDataReader {
+ byte[] data = new byte[43800];
+ int length = 10;
+ int cur = 0;
+ byte seq = 0;
+ public MyDataReader(int len) {
+ length = len;
+ }
+
+ public int append(byte[] b, int off, int len) throws Exception {
+ int packages = 0;
+ for ( int i=off; i<len; i++ ) {
+ if ( cur == length ) {
+ cur = 0;
+ seq++;
+ packages++;
+ }
+ if ( b[i] != seq ) throw new Exception("mismatch on seq:"+seq+" and byte nr:"+cur+" count:"+count+" packages:"+packages);
+ cur++;
+ }
+ return packages;
+ }
+ }
+}
\ No newline at end of file
projects / tomcat7.0 / commitdiff