fhanik 2004/02/04 21:27:31
Modified: catalina/src/conf server.xml
catalina/src/share/org/apache/catalina/startup Catalina.java
HostRuleSet.java
modules/cluster build.xml
modules/cluster/src/share/org/apache/catalina/cluster
CatalinaCluster.java MembershipService.java
modules/cluster/src/share/org/apache/catalina/cluster/mcast
McastMember.java McastService.java
modules/cluster/src/share/org/apache/catalina/cluster/tcp
ReplicationListener.java
ReplicationTransmitter.java SimpleTcpCluster.java
Added: modules/cluster/src/share/org/apache/catalina/cluster
ClusterReceiver.java ClusterSender.java
Removed: modules/cluster/src/share/org/apache/catalina/cluster
MembershipFactory.java
Log:
Refactored a lot of the cluster code
1. server.xml - the cluster element has sub elements, makes the configuration is
modular
2. added in interfaces for all the modules inside the cluster, so that they can be
replaced, and the dependencies are not hard coded anymore
3. Added a ClusterRuleSet to support the new cluster configuration
Revision Changes Path
1.30 +30 -26 jakarta-tomcat-catalina/catalina/src/conf/server.xml
Index: server.xml
===================================================================
RCS file: /home/cvs/jakarta-tomcat-catalina/catalina/src/conf/server.xml,v
retrieving revision 1.29
retrieving revision 1.30
diff -u -r1.29 -r1.30
--- server.xml 21 Jan 2004 23:50:03 -0000 1.29
+++ server.xml 5 Feb 2004 05:27:31 -0000 1.30
@@ -271,29 +271,6 @@
in the queue instead of replicating two requests.
This almost never happens, unless there is a
large network delay.
-->
-
- <!-- When uncommenting the cluster, REMEMBER to uncomment the replication
Valve below as well
-
-
- <Cluster className="org.apache.catalina.cluster.tcp.SimpleTcpCluster"
- name="FilipsCluster"
- debug="10"
- serviceclass="org.apache.catalina.cluster.mcast.McastService"
- mcastAddr="228.0.0.4"
- mcastPort="45564"
- mcastFrequency="500"
- mcastDropTime="3000"
- tcpThreadCount="6"
- tcpListenAddress="auto"
- tcpListenPort="4001"
- tcpSelectorTimeout="100"
- printToScreen="false"
- expireSessionsOnShutdown="false"
- useDirtyFlag="true"
- replicationMode="pooled"
- />
-
- -->
<!--
When configuring for clustering, you also add in a valve to catch all
the requests
coming in, at the end of the request, the session may or may not be
replicated.
@@ -311,11 +288,38 @@
filter=".*\.gif;.*\.js;" means that we will not replicate the session
after requests with the URI
ending with .gif and .js are intercepted.
-->
+
<!--
- <Valve className="org.apache.catalina.cluster.tcp.ReplicationValve"
- filter=".*\.gif;.*\.js;.*\.jpg;.*\.htm;.*\.html;.*\.txt;"/>
+ <Cluster className="org.apache.catalina.cluster.tcp.SimpleTcpCluster"
+ managerClassName="org.apache.catalina.cluster.session.DeltaManager"
+ expireSessionsOnShutdown="false"
+ useDirtyFlag="true">
+
+ <Membership
+ className="org.apache.catalina.cluster.mcast.McastService"
+ mcastAddr="228.0.0.4"
+ mcastPort="45564"
+ mcastFrequency="500"
+ mcastDropTime="3000"/>
+
+ <Receiver
+ className="org.apache.catalina.cluster.tcp.ReplicationListener"
+ tcpListenAddress="auto"
+ tcpListenPort="4001"
+ tcpSelectorTimeout="100"
+ tcpThreadCount="6"/>
+
+ <Sender
+ className="org.apache.catalina.cluster.tcp.ReplicationTransmitter"
+ replicationMode="pooled"/>
+
+ <Valve className="org.apache.catalina.cluster.tcp.ReplicationValve"
+ filter=".*\.gif;.*\.js;.*\.jpg;.*\.htm;.*\.html;.*\.txt;"/>
+ </Cluster>
+ -->
+
+
- -->
<!-- Normally, users must authenticate themselves to each web app
individually. Uncomment the following entry if you would like
a user to be authenticated the first time they encounter a
1.25 +5 -4
jakarta-tomcat-catalina/catalina/src/share/org/apache/catalina/startup/Catalina.java
Index: Catalina.java
===================================================================
RCS file:
/home/cvs/jakarta-tomcat-catalina/catalina/src/share/org/apache/catalina/startup/Catalina.java,v
retrieving revision 1.24
retrieving revision 1.25
diff -u -r1.24 -r1.25
--- Catalina.java 25 Jan 2004 20:57:18 -0000 1.24
+++ Catalina.java 5 Feb 2004 05:27:31 -0000 1.25
@@ -386,6 +386,7 @@
digester.addRuleSet(new
ContextRuleSet("Server/Service/Engine/Host/Default"));
digester.addRuleSet(new
NamingRuleSet("Server/Service/Engine/Host/DefaultContext/"));
digester.addRuleSet(new ContextRuleSet("Server/Service/Engine/Host/"));
+ digester.addRuleSet(new
ClusterRuleSet("Server/Service/Engine/Host/Cluster/"));
digester.addRuleSet(new
NamingRuleSet("Server/Service/Engine/Host/Context/"));
// When the 'engine' is found, set the parentClassLoader.
1.5 +6 -4
jakarta-tomcat-catalina/catalina/src/share/org/apache/catalina/startup/HostRuleSet.java
Index: HostRuleSet.java
===================================================================
RCS file:
/home/cvs/jakarta-tomcat-catalina/catalina/src/share/org/apache/catalina/startup/HostRuleSet.java,v
retrieving revision 1.4
retrieving revision 1.5
diff -u -r1.4 -r1.5
--- HostRuleSet.java 25 Jan 2004 21:01:03 -0000 1.4
+++ HostRuleSet.java 5 Feb 2004 05:27:31 -0000 1.5
@@ -150,6 +150,7 @@
digester.addCallMethod(prefix + "Host/Alias",
"addAlias", 0);
+ //Cluster configuration start
digester.addObjectCreate(prefix + "Host/Cluster",
null, // MUST be specified in the element
"className");
@@ -157,6 +158,7 @@
digester.addSetNext(prefix + "Host/Cluster",
"setCluster",
"org.apache.catalina.Cluster");
+ //Cluster configuration end
digester.addObjectCreate(prefix + "Host/Listener",
null, // MUST be specified in the element
1.7 +1 -0 jakarta-tomcat-catalina/modules/cluster/build.xml
Index: build.xml
===================================================================
RCS file: /home/cvs/jakarta-tomcat-catalina/modules/cluster/build.xml,v
retrieving revision 1.6
retrieving revision 1.7
diff -u -r1.6 -r1.7
--- build.xml 6 Sep 2003 17:50:59 -0000 1.6
+++ build.xml 5 Feb 2004 05:27:31 -0000 1.7
@@ -20,6 +20,7 @@
<pathelement location="${commons-logging.jar}"/>
<pathelement location="${jmx.jar}"/>
<pathelement location="${catalina.build}/common/lib/servlet-api.jar"/>
+ <pathelement location="${catalina.build}/server/lib/commons-beanutils.jar"/>
</path>
<!-- Source path -->
1.2 +16 -6
jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/CatalinaCluster.java
Index: CatalinaCluster.java
===================================================================
RCS file:
/home/cvs/jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/CatalinaCluster.java,v
retrieving revision 1.1
retrieving revision 1.2
diff -u -r1.1 -r1.2
--- CatalinaCluster.java 5 Feb 2004 01:02:48 -0000 1.1
+++ CatalinaCluster.java 5 Feb 2004 05:27:31 -0000 1.2
@@ -64,9 +64,11 @@
package org.apache.catalina.cluster;
import org.apache.catalina.Cluster;
+import org.apache.catalina.cluster.io.ListenCallback;
import org.apache.catalina.LifecycleException;
import org.apache.catalina.LifecycleListener;
import org.apache.catalina.Logger;
+import org.apache.catalina.Valve;
import org.apache.commons.logging.Log;
/**
@@ -78,7 +80,7 @@
*/
public interface CatalinaCluster
- extends Cluster {
+ extends Cluster,ListenCallback {
// ----------------------------------------------------- Instance Variables
/**
@@ -122,5 +124,13 @@
* @return Member[]
*/
public Member[] getMembers();
-
+
+ public void setClusterSender(ClusterSender sender);
+
+ public void setClusterReceiver(ClusterReceiver receiver);
+
+ public void setMembershipService(MembershipService service);
+
+ public void addValve(Valve valve);
+
}
1.2 +11 -5
jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/MembershipService.java
Index: MembershipService.java
===================================================================
RCS file:
/home/cvs/jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/MembershipService.java,v
retrieving revision 1.1
retrieving revision 1.2
diff -u -r1.1 -r1.2
--- MembershipService.java 19 Feb 2003 20:57:17 -0000 1.1
+++ MembershipService.java 5 Feb 2004 05:27:31 -0000 1.2
@@ -105,6 +105,12 @@
* @return
*/
public Member getLocalMember();
+
+ /**
+ * Sets the local member properties for broadcasting
+ * @return
+ */
+ public void setLocalMemberProperties(String listenHost, int listenPort);
/**
* Sets the membership listener, only one listener can be added.
* If you call this method twice, the last listener will be used.
@@ -116,4 +122,4 @@
*/
public void removeMembershipListener();
-}
\ No newline at end of file
+}
1.1
jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/ClusterReceiver.java
Index: ClusterReceiver.java
===================================================================
/*
* $Header:
/home/cvs/jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/ClusterReceiver.java,v
1.1 2004/02/05 05:27:31 fhanik Exp $
* $Revision: 1.1 $
* $Date: 2004/02/05 05:27:31 $
*
* ====================================================================
*
* The Apache Software License, Version 1.1
*
* Copyright (c) 1999 The Apache Software Foundation. All rights
* reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
* 1. Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in
* the documentation and/or other materials provided with the
* distribution.
*
* 3. The end-user documentation included with the redistribution, if
* any, must include the following acknowlegement:
* "This product includes software developed by the
* Apache Software Foundation (http://www.apache.org/)."
* Alternately, this acknowlegement may appear in the software itself,
* if and wherever such third-party acknowlegements normally appear.
*
* 4. The names "The Jakarta Project", "Tomcat", and "Apache Software
* Foundation" must not be used to endorse or promote products derived
* from this software without prior written permission. For written
* permission, please contact [EMAIL PROTECTED]
*
* 5. Products derived from this software may not be called "Apache"
* nor may "Apache" appear in their names without prior written
* permission of the Apache Group.
*
* THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
* WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
* OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
* DISCLAIMED. IN NO EVENT SHALL THE APACHE SOFTWARE FOUNDATION OR
* ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
* USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
* OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
* OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
* SUCH DAMAGE.
* ====================================================================
*
* This software consists of voluntary contributions made by many
* individuals on behalf of the Apache Software Foundation. For more
* information on the Apache Software Foundation, please see
* <http://www.apache.org/>.
*
* [Additional notices, if required by prior licensing conditions]
*
*/
package org.apache.catalina.cluster;
public interface ClusterReceiver
{
public void start() throws java.io.IOException;
public void stop();
public void setCatalinaCluster(CatalinaCluster cluster);
public void setIsSenderSynchronized(boolean isSenderSynchronized);
public String getHost();
public int getPort();
}
1.1
jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/ClusterSender.java
Index: ClusterSender.java
===================================================================
/*
* $Header:
/home/cvs/jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/ClusterSender.java,v
1.1 2004/02/05 05:27:31 fhanik Exp $
* $Revision: 1.1 $
* $Date: 2004/02/05 05:27:31 $
*
* ====================================================================
*
* The Apache Software License, Version 1.1
*
* Copyright (c) 1999 The Apache Software Foundation. All rights
* reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
* are met:
*
* 1. Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in
* the documentation and/or other materials provided with the
* distribution.
*
* 3. The end-user documentation included with the redistribution, if
* any, must include the following acknowlegement:
* "This product includes software developed by the
* Apache Software Foundation (http://www.apache.org/)."
* Alternately, this acknowlegement may appear in the software itself,
* if and wherever such third-party acknowlegements normally appear.
*
* 4. The names "The Jakarta Project", "Tomcat", and "Apache Software
* Foundation" must not be used to endorse or promote products derived
* from this software without prior written permission. For written
* permission, please contact [EMAIL PROTECTED]
*
* 5. Products derived from this software may not be called "Apache"
* nor may "Apache" appear in their names without prior written
* permission of the Apache Group.
*
* THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
* WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
* OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
* DISCLAIMED. IN NO EVENT SHALL THE APACHE SOFTWARE FOUNDATION OR
* ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
* USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
* OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
* OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
* SUCH DAMAGE.
* ====================================================================
*
* This software consists of voluntary contributions made by many
* individuals on behalf of the Apache Software Foundation. For more
* information on the Apache Software Foundation, please see
* <http://www.apache.org/>.
*
* [Additional notices, if required by prior licensing conditions]
*
*/
package org.apache.catalina.cluster;
public interface ClusterSender
{
public void add(Member member);
public void remove(Member member);
public void start() throws java.io.IOException;
public void stop();
public void sendMessage(String messageId, byte[] indata, Member member) throws
java.io.IOException;
public void sendMessage(String messageId, byte[] indata) throws
java.io.IOException;
public boolean getIsSenderSynchronized();
}
1.3 +17 -5
jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/mcast/McastMember.java
Index: McastMember.java
===================================================================
RCS file:
/home/cvs/jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/mcast/McastMember.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- McastMember.java 16 Nov 2003 22:22:45 -0000 1.2
+++ McastMember.java 5 Feb 2004 05:27:31 -0000 1.3
@@ -328,4 +328,16 @@
}
return new String(buf, charPos, 15 - charPos);
}
-}
\ No newline at end of file
+ public void setHost(String host) {
+ this.host = host;
+ }
+ public void setMsgCount(int msgCount) {
+ this.msgCount = msgCount;
+ }
+ public void setName(String name) {
+ this.name = name;
+ }
+ public void setPort(int port) {
+ this.port = port;
+ }
+}
1.6 +48 -7
jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/mcast/McastService.java
Index: McastService.java
===================================================================
RCS file:
/home/cvs/jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/mcast/McastService.java,v
retrieving revision 1.5
retrieving revision 1.6
diff -u -r1.5 -r1.6
--- McastService.java 9 Jan 2004 02:50:54 -0000 1.5
+++ McastService.java 5 Feb 2004 05:27:31 -0000 1.6
@@ -87,7 +87,7 @@
/**
* The implementation specific properties
*/
- protected Properties properties;
+ protected Properties properties = new Properties();
/**
* A handle to the actual low level implementation
*/
@@ -99,7 +99,7 @@
/**
* The local member
*/
- protected McastMember localMember;
+ protected McastMember localMember ;
/**
* Create a membership service.
@@ -145,6 +145,37 @@
localMember.setMemberAliveTime(System.currentTimeMillis()-impl.getServiceStartTime());
return localMember;
}
+
+ /**
+ * Sets the local member properties for broadcasting
+ * @return
+ */
+ public void setLocalMemberProperties(String listenHost, int listenPort) {
+ properties.setProperty("tcpListenHost",listenHost);
+ properties.setProperty("tcpListenPort",String.valueOf(listenPort));
+ }
+
+ public void setMcastAddr(String addr) {
+ properties.setProperty("mcastAddress", addr);
+ }
+
+ public void setMcastBindAddress(String bindaddr) {
+ properties.setProperty("mcastBindAddress", bindaddr);
+ }
+
+ public void setMcastPort(int port) {
+ properties.setProperty("mcastPort", String.valueOf(port));
+ }
+
+ public void setMcastFrequency(long time) {
+ properties.setProperty("msgFrequency", String.valueOf(time));
+ }
+
+ public void setMcastDropTime(long time) {
+ properties.setProperty("memberDropTime", String.valueOf(time));
+ }
+
+
/**
@@ -165,7 +196,14 @@
String host = getProperties().getProperty("tcpListenHost");
int port = Integer.parseInt(getProperties().getProperty("tcpListenPort"));
String name = "tcp://"+host+":"+port;
- localMember = new McastMember(name,host,port,100);
+ if ( localMember == null ) {
+ localMember = new McastMember(name, host, port, 100);
+ } else {
+ localMember.setName(name);
+ localMember.setHost(host);
+ localMember.setPort(port);
+ localMember.setMemberAliveTime(100);
+ }
java.net.InetAddress bind = null;
if ( properties.getProperty("mcastBindAddress")!= null ) {
bind =
java.net.InetAddress.getByName(properties.getProperty("mcastBindAddress"));
@@ -178,6 +216,9 @@
this);
impl.start();
+ log.info("Sleeping for
"+(Long.parseLong(properties.getProperty("msgFrequency"))*4)+" secs to establish
cluster membership");
+
Thread.currentThread().sleep((Long.parseLong(properties.getProperty("msgFrequency"))*4));
+
}
/**
1.11 +77 -25
jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/tcp/ReplicationListener.java
Index: ReplicationListener.java
===================================================================
RCS file:
/home/cvs/jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/tcp/ReplicationListener.java,v
retrieving revision 1.10
retrieving revision 1.11
diff -u -r1.10 -r1.11
--- ReplicationListener.java 5 Feb 2004 01:02:48 -0000 1.10
+++ ReplicationListener.java 5 Feb 2004 05:27:31 -0000 1.11
@@ -79,9 +79,10 @@
import org.apache.catalina.cluster.io.ObjectReader;
import org.apache.catalina.cluster.io.XByteBuffer;
import org.apache.catalina.cluster.CatalinaCluster;
+import org.apache.catalina.cluster.ClusterReceiver;
/**
*/
-public class ReplicationListener extends Thread
+public class ReplicationListener implements Runnable,ClusterReceiver
{
private static org.apache.commons.logging.Log log =
@@ -90,27 +91,36 @@
private boolean doListen = false;
private ListenCallback callback;
private java.net.InetAddress bind;
- private int port;
- private long timeout = 0;
- private boolean synchronous = false;
- public ReplicationListener(ListenCallback callback,
- int poolSize,
- java.net.InetAddress bind,
- int port,
- long timeout,
- boolean synchronous)
- {
+ private String tcpListenAddress;
+ private int tcpThreadCount;
+ private long tcpSelectorTimeout;
+ private int tcpListenPort;
+ private boolean isSenderSynchronized;
+
+ public ReplicationListener() {
+ }
+
+ public void start() {
try {
- this.synchronous=synchronous;
- pool = new ThreadPool(poolSize, TcpReplicationThread.class);
- }catch ( Exception x ) {
- log.fatal("Unable to start thread pool for TCP listeners, session
replication will fail! msg="+x.getMessage());
+ pool = new ThreadPool(tcpThreadCount, TcpReplicationThread.class);
+ if ( "auto".equals(tcpListenAddress) ) {
+ tcpListenAddress = java.net.InetAddress.getLocalHost().
+ getHostAddress();
+ }
+ bind = java.net.InetAddress.getByName(tcpListenAddress);
+ Thread t = new Thread(this,"ClusterReceiver");
+ t.setDaemon(true);
+ t.start();
+ } catch ( Exception x ) {
+ log.fatal("Unable to start cluster receiver",x);
}
- this.callback = callback;
- this.bind = bind;
- this.port = port;
- this.timeout = timeout;
+
}
+
+ public void stop() {
+ stopListening();
+ }
+
public void run()
{
@@ -135,7 +145,7 @@
// create a new Selector for use below
Selector selector = Selector.open();
// set the port the server channel will listen to
- serverSocket.bind (new InetSocketAddress (bind,port));
+ serverSocket.bind (new InetSocketAddress (bind,tcpListenPort));
// set non-blocking mode for the listening socket
serverChannel.configureBlocking (false);
// register the ServerSocketChannel with the Selector
@@ -146,7 +156,7 @@
try {
//System.out.println("Selecting with timeout="+timeout);
- int n = selector.select(timeout);
+ int n = selector.select(tcpSelectorTimeout);
//System.out.println("select returned="+n);
if (n == 0) {
continue; // nothing to do
@@ -197,6 +207,10 @@
public void stopListening(){
doListen = false;
}
+
+ public void setCatalinaCluster(CatalinaCluster cluster) {
+ callback = cluster;
+ }
// ----------------------------------------------------------
@@ -239,9 +253,47 @@
return;
} else {
// invoking this wakes up the worker thread then returns
- worker.serviceChannel(key, synchronous);
+ worker.serviceChannel(key, isSenderSynchronized);
return;
}
+ }
+ public String getTcpListenAddress() {
+ return tcpListenAddress;
+ }
+ public void setTcpListenAddress(String tcpListenAddress) {
+ this.tcpListenAddress = tcpListenAddress;
+ }
+ public int getTcpListenPort() {
+ return tcpListenPort;
+ }
+ public void setTcpListenPort(int tcpListenPort) {
+ this.tcpListenPort = tcpListenPort;
+ }
+ public long getTcpSelectorTimeout() {
+ return tcpSelectorTimeout;
+ }
+ public void setTcpSelectorTimeout(long tcpSelectorTimeout) {
+ this.tcpSelectorTimeout = tcpSelectorTimeout;
+ }
+ public int getTcpThreadCount() {
+ return tcpThreadCount;
+ }
+ public void setTcpThreadCount(int tcpThreadCount) {
+ this.tcpThreadCount = tcpThreadCount;
+ }
+ public boolean getIsSenderSynchronized() {
+ return isSenderSynchronized;
+ }
+ public void setIsSenderSynchronized(boolean isSenderSynchronized) {
+ this.isSenderSynchronized = isSenderSynchronized;
+ }
+
+ public String getHost() {
+ return getTcpListenAddress();
+ }
+
+ public int getPort() {
+ return getTcpListenPort();
}
}
1.14 +49 -24
jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/tcp/ReplicationTransmitter.java
Index: ReplicationTransmitter.java
===================================================================
RCS file:
/home/cvs/jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/tcp/ReplicationTransmitter.java,v
retrieving revision 1.13
retrieving revision 1.14
diff -u -r1.13 -r1.14
--- ReplicationTransmitter.java 5 Feb 2004 01:02:48 -0000 1.13
+++ ReplicationTransmitter.java 5 Feb 2004 05:27:31 -0000 1.14
@@ -64,23 +64,23 @@
package org.apache.catalina.cluster.tcp;
import org.apache.catalina.cluster.io.XByteBuffer;
-import org.apache.catalina.cluster.CatalinaCluster;
+import org.apache.catalina.cluster.Member;
+import org.apache.catalina.cluster.ClusterSender;
-
-public class ReplicationTransmitter
+
+public class ReplicationTransmitter implements ClusterSender
{
private static org.apache.commons.logging.Log log =
org.apache.commons.logging.LogFactory.getLog( ReplicationTransmitter.class
);
private java.util.HashMap map = new java.util.HashMap();
- public ReplicationTransmitter(IDataSender[] senders)
+ public ReplicationTransmitter()
{
- for ( int i=0; i<senders.length; i++)
-
map.put(senders[i].getAddress().getHostAddress()+":"+senders[i].getPort(),senders[i]);
}
private static long nrOfRequests = 0;
private static long totalBytes = 0;
+ private String replicationMode;
private static synchronized void addStats(int length) {
nrOfRequests++;
totalBytes+=length;
@@ -89,20 +89,40 @@
}
}
+
+ public void setReplicationMode(String mode) {
+ String msg = IDataSenderFactory.validateMode(mode);
+ if (msg == null) {
+ log.debug("Setting replcation mode to " + mode);
+ this.replicationMode = mode;
+ }
+ else
+ throw new IllegalArgumentException(msg);
+
+ }
+
- public synchronized void add(IDataSender sender)
+ public synchronized void add(Member member)
{
- String key = sender.getAddress().getHostAddress()+":"+sender.getPort();
- if ( !map.containsKey(key) )
-
map.put(sender.getAddress().getHostAddress()+":"+sender.getPort(),sender);
+ try {
+ IDataSender sender = IDataSenderFactory.getIDataSender(
+ replicationMode, member);
+ String key = sender.getAddress().getHostAddress() + ":" +
+ sender.getPort();
+ if (!map.containsKey(key))
+ map.put(sender.getAddress().getHostAddress() + ":" +
+ sender.getPort(), sender);
+ }catch ( java.io.IOException x ) {
+ log.error("Unable to create and add a IDataSender object.",x);
+ }
}//add
- public synchronized void remove(java.net.InetAddress addr,int port)
+ public synchronized void remove(Member member)
{
- String key = addr.getHostAddress()+":"+port;
- IDataSender sender = (IDataSender)map.get(key);
- if ( sender == null ) return;
- sender.disconnect();
+ String key = member.getHost() + ":" + member.getPort();
+ IDataSender toberemoved = (IDataSender) map.get(key);
+ if (toberemoved == null)return;
+ toberemoved.disconnect();
map.remove(key);
}
@@ -117,10 +137,7 @@
while ( i.hasNext() )
{
IDataSender sender =
(IDataSender)((java.util.Map.Entry)i.next()).getValue();
- if ( sender.isConnected() )
- {
- try { sender.disconnect(); } catch ( Exception x ){}
- }//end if
+ try { sender.disconnect(); } catch ( Exception x ){}
}//while
}//stop
@@ -158,10 +175,10 @@
}
}
- public void sendMessage(String sessionId, byte[] indata, java.net.InetAddress
addr, int port) throws java.io.IOException
+ public void sendMessage(String sessionId, byte[] indata, Member member) throws
java.io.IOException
{
byte[] data = XByteBuffer.createDataPackage(indata);
- String key = addr.getHostAddress()+":"+port;
+ String key = member.getHost()+":"+member.getPort();
IDataSender sender = (IDataSender)map.get(key);
sendMessageData(sessionId,data,sender);
}
@@ -184,6 +201,14 @@
sender.setSuspect(true);
}
}//while
+ }
+ public String getReplicationMode() {
+ return replicationMode;
+ }
+
+ public boolean getIsSenderSynchronized() {
+ return IDataSenderFactory.SYNC_MODE.equals(replicationMode) ||
+ IDataSenderFactory.POOLED_SYNC_MODE.equals(replicationMode);
}
1.30 +88 -217
jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/tcp/SimpleTcpCluster.java
Index: SimpleTcpCluster.java
===================================================================
RCS file:
/home/cvs/jakarta-tomcat-catalina/modules/cluster/src/share/org/apache/catalina/cluster/tcp/SimpleTcpCluster.java,v
retrieving revision 1.29
retrieving revision 1.30
diff -u -r1.29 -r1.30
--- SimpleTcpCluster.java 5 Feb 2004 01:02:48 -0000 1.29
+++ SimpleTcpCluster.java 5 Feb 2004 05:27:31 -0000 1.30
@@ -83,13 +83,15 @@
import org.apache.catalina.Manager;
import org.apache.catalina.util.LifecycleSupport;
import org.apache.catalina.util.StringManager;
+import org.apache.catalina.Valve;
+import org.apache.catalina.Host;
import org.apache.catalina.cluster.Member;
import org.apache.catalina.cluster.CatalinaCluster;
import org.apache.catalina.cluster.MembershipFactory;
import org.apache.catalina.cluster.MembershipListener;
import org.apache.catalina.cluster.MembershipService;
-
+import org.apache.commons.beanutils.MethodUtils;
import org.apache.catalina.cluster.tcp.ReplicationListener;
import org.apache.catalina.cluster.tcp.ReplicationTransmitter;
import org.apache.catalina.cluster.tcp.SocketSender;
@@ -99,6 +101,9 @@
import org.apache.catalina.cluster.session.ReplicationStream;
import org.apache.catalina.cluster.ClusterManager;
import org.apache.catalina.cluster.Constants;
+import org.apache.catalina.cluster.ClusterReceiver;
+import org.apache.catalina.cluster.ClusterSender;
+
import org.apache.commons.logging.Log;
@@ -130,46 +135,14 @@
/**
* Descriptive information about this component implementation.
*/
- protected static final String info = "SimpleTcpCluster/1.0";
+ protected static final String info = "SimpleTcpCluster2/1.0";
/**
* the service that provides the membership
*/
- protected MembershipService service = null;
- /**
- * the class name of the membership service
- */
- protected String serviceclass = null;
- /**
- * The properties for the
- */
- protected java.util.Properties svcproperties = new java.util.Properties();
-
- /**
- * Tcp address to listen for incoming changes
- */
- protected java.net.InetAddress tcpAddress = null;
-
- /**
- * The tcp port this instance listens to for incoming changes
- */
- protected int tcpPort = 1234;
+ protected MembershipService membershipService = null;
- /**
- * number of thread we listen to tcp requests coming in on
- */
- protected int tcpThreadCount = 2;
-
- /**
- * ReplicationTransmitter to send data with
- */
- protected ReplicationTransmitter mReplicationTransmitter;
-
- /**
- * Name to register for the background thread.
- */
- protected String threadName = "SimpleTcpCluster";
/**
* Whether to expire sessions when shutting down
@@ -188,7 +161,7 @@
/**
* Name for logging purpose
*/
- protected String clusterImpName = "SimpleTcpCluster";
+ protected String clusterImpName = "SimpleTcpCluster2";
/**
@@ -196,13 +169,6 @@
*/
protected StringManager sm = StringManager.getManager(Constants.Package);
-
- /**
- * The background thread completion semaphore.
- */
- protected boolean threadDone = false;
-
-
/**
* The cluster name to join
*/
@@ -243,52 +209,30 @@
* The context name <-> manager association for distributed contexts.
*/
protected HashMap managers = new HashMap();
- /**
- * The context name <-> manager association for all contexts.
- */
- protected HashMap allmanagers = new HashMap();
/**
- * Nr of milliseconds between every heart beat
+ * Sending Stats
*/
- protected long msgFrequency = 500;
- /**
- * java.nio.Channels.Selector.select timeout in case the JDK has a
- * poor nio implementation
- */
- protected long tcpSelectorTimeout = 100;
+ private long nrOfMsgsReceived = 0;
+ private long msgSendTime = 0;
+ private long lastChecked = System.currentTimeMillis();
+
+ private String managerClassName =
"org.apache.catalina.cluster.session.DeltaManager";
/**
- * The channel configuration.
+ * Sender to send data with
*/
- protected String protocol = null;
+ private org.apache.catalina.cluster.ClusterSender clusterSender;
/**
- * The replication mode, can be either synchronous or asynchronous
- * defaults to synchronous
+ * Receiver to register call back with
*/
- protected String replicationMode="synchronous";
-
- private long nrOfMsgsReceived = 0;
- private long msgSendTime = 0;
- private long lastChecked = System.currentTimeMillis();
- private boolean isJdk13 = false;
- private String managerClassName =
"org.apache.catalina.cluster.session.DeltaManager";
+ private org.apache.catalina.cluster.ClusterReceiver clusterReceiver;
+ private org.apache.catalina.Valve valve;
// ------------------------------------------------------------- Properties
public SimpleTcpCluster() {
- try {
- tcpAddress = java.net.InetAddress.getLocalHost();
- }catch ( Exception x ) {
- log.error("In SimpleTcpCluster.constructor()",x);
- }
-
-// if ( ServerFactory.getServer() instanceof StandardServer ) {
-// StandardServer server = (StandardServer) ServerFactory.getServer();
-// server.addLifecycleListener(this);
-// }
-
}
/**
* Return descriptive information about this Cluster implementation and
@@ -319,15 +263,6 @@
return(this.debug);
}
- public void setReplicationMode(String mode) {
- String msg = IDataSenderFactory.validateMode(mode);
- if ( msg == null ) {
- log.debug("Setting replcation mode to "+mode);
- this.replicationMode = mode;
- } else
- throw new IllegalArgumentException(msg);
-
- }
/**
* Set the name of the cluster to join, if no cluster with
* this name is present create one.
@@ -335,11 +270,7 @@
* @param clusterName The clustername to join
*/
public void setClusterName(String clusterName) {
- String oldClusterName = this.clusterName;
this.clusterName = clusterName;
- support.firePropertyChange("clusterName",
- oldClusterName,
- this.clusterName);
}
@@ -350,7 +281,7 @@
* @return The name of the cluster associated with this server
*/
public String getClusterName() {
- return(this.clusterName);
+ return clusterName;
}
@@ -388,9 +319,6 @@
* @see <a href="www.javagroups.com">JavaGroups</a> for details
*/
public void setProtocol(String protocol) {
- String oldProtocol = this.protocol;
- this.protocol = protocol;
- support.firePropertyChange("protocol", oldProtocol, this.protocol);
}
@@ -398,12 +326,13 @@
* Returns the protocol.
*/
public String getProtocol() {
- return (this.protocol);
+ return null;
}
public Member[] getMembers() {
- return service.getMembers();
+ return membershipService.getMembers();
}
+
@@ -423,7 +352,6 @@
manager.setDistributable(true);
manager.setExpireSessionsOnShutdown(expireSessionsOnShutdown);
manager.setUseDirtyFlag(useDirtyFlag);
- allmanagers.put(name, manager);
managers.put(name,manager);
return manager;
}
@@ -483,59 +411,30 @@
(sm.getString("cluster.alreadyStarted"));
log.info("Cluster is about to start");
try {
- if ( isJdk13 ) {
- Jdk13ReplicationListener mReplicationListener =
- new Jdk13ReplicationListener(this,
- this.tcpThreadCount,
- this.tcpAddress,
- this.tcpPort,
- this.tcpSelectorTimeout,
- "synchronous".equals(this.
- replicationMode));
- Thread t = new Thread(mReplicationListener);
- t.setName("Cluster-TcpListener");
- t.setDaemon(true);
- t.start();
- } else {
- ReplicationListener mReplicationListener =
- new ReplicationListener(this,
- this.tcpThreadCount,
- this.tcpAddress,
- this.tcpPort,
- this.tcpSelectorTimeout,
-
IDataSenderFactory.SYNC_MODE.equals(replicationMode) ||
-
IDataSenderFactory.POOLED_SYNC_MODE.equals(replicationMode));
- mReplicationListener.setName("Cluster-ReplicationListener");
- mReplicationListener.setDaemon(true);
- mReplicationListener.start();
- }
-
- mReplicationTransmitter = new ReplicationTransmitter(new
IDataSender[0]);
- mReplicationTransmitter.start();
-
- //wait 5 seconds to establish the view membership
- log.info("Sleeping for "+(msgFrequency*4)+" secs to establish cluster
membership");
- service =
MembershipFactory.getMembershipService(serviceclass,svcproperties);
- service.addMembershipListener(this);
- service.start();
- Thread.currentThread().sleep((msgFrequency*4));
+ MethodUtils.invokeMethod(getContainer(), "addValve", valve);
+
clusterReceiver.setIsSenderSynchronized(clusterSender.getIsSenderSynchronized());
+ clusterReceiver.setCatalinaCluster(this);
+ clusterReceiver.start();
+ clusterSender.start();
+
membershipService.setLocalMemberProperties(clusterReceiver.getHost(),clusterReceiver.getPort());
+ membershipService.addMembershipListener(this);
+ membershipService.start();
this.started = true;
} catch ( Exception x ) {
log.error("Unable to start cluster.",x);
+ throw new LifecycleException(x);
}
-
-
}
public void send(SessionMessage msg, Member dest) {
try
{
- msg.setAddress(service.getLocalMember());
+ msg.setAddress(membershipService.getLocalMember());
Member destination = dest;
if ( (destination == null) && (msg.getEventType() ==
SessionMessage.EVT_GET_ALL_SESSIONS) ) {
- if (service.getMembers().length > 0)
- destination = service.getMembers()[0];
+ if (membershipService.getMembers().length > 0)
+ destination = membershipService.getMembers()[0];
}
msg.setTimestamp(System.currentTimeMillis());
java.io.ByteArrayOutputStream outs = new
java.io.ByteArrayOutputStream();
@@ -544,18 +443,15 @@
byte[] data = outs.toByteArray();
if(destination != null) {
Member tcpdest = dest;
- if ( (tcpdest != null) &&
(!service.getLocalMember().equals(tcpdest))) {
- mReplicationTransmitter.sendMessage(msg.getSessionID(),
- data,
-
InetAddress.getByName(tcpdest.getHost()),
- tcpdest.getPort());
+ if ( (tcpdest != null) &&
(!membershipService.getLocalMember().equals(tcpdest))) {
+ clusterSender.sendMessage(msg.getSessionID(), data, tcpdest);
}//end if
}
else {
- mReplicationTransmitter.sendMessage(msg.getSessionID(),data);
+ clusterSender.sendMessage(msg.getSessionID(),data);
}
} catch ( Exception x ) {
- log.error("Unable to send message through tcp channel",x);
+ log.error("Unable to send message through cluster sender.",x);
}
}
@@ -580,15 +476,28 @@
if (!started)
throw new IllegalStateException
(sm.getString("cluster.notStarted"));
-
+
+ membershipService.stop();
+ membershipService.removeMembershipListener();
+ try {
+ clusterSender.stop();
+ } catch (Exception x ) {
+ log.error("Unable to stop cluster sender.",x);
+ }
+ try {
+ clusterReceiver.stop();
+ clusterReceiver.setCatalinaCluster(null);
+ } catch (Exception x ) {
+ log.error("Unable to stop cluster receiver.",x);
+ }
+ started = false;
}
public void memberAdded(Member member) {
try {
log.info("Replication member added:" + member);
- Member mbr = member;
-
mReplicationTransmitter.add(IDataSenderFactory.getIDataSender(replicationMode,mbr));
+ clusterSender.add(member);
} catch ( Exception x ) {
log.error("Unable to connect to replication system.",x);
}
@@ -600,9 +509,7 @@
log.info("Received member disappeared:"+member);
try
{
- Member mbr = member;
- mReplicationTransmitter.remove(InetAddress.getByName(mbr.getHost()),
- mbr.getPort());
+ clusterSender.remove(member);
}
catch ( Exception x )
{
@@ -611,51 +518,7 @@
}
- public void setServiceclass(String clazz){
- this.serviceclass = clazz;
- }
- public void setMcastAddr(String addr) {
- svcproperties.setProperty("mcastAddress",addr);
- }
-
-
- public void setMcastBindAddress(String bindaddr) {
- svcproperties.setProperty("mcastBindAddress",bindaddr);
- }
-
- public void setMcastPort(int port) {
- svcproperties.setProperty("mcastPort",String.valueOf(port));
- }
-
- public void setMcastFrequency(long time) {
- svcproperties.setProperty("msgFrequency",String.valueOf(time));
- msgFrequency = time;
- }
-
- public void setMcastDropTime(long time) {
- svcproperties.setProperty("memberDropTime",String.valueOf(time));
- }
-
- public void setTcpThreadCount(int count) {
- this.tcpThreadCount = count;
- }
-
- public void setTcpListenAddress(String address) {
- try {
- if ("auto".equals(address) )
- {
- address = java.net.InetAddress.getLocalHost().getHostAddress();
- tcpAddress = java.net.InetAddress.getByName(address);
- }
- else
- {
- tcpAddress = java.net.InetAddress.getByName(address);
- }//end if
- svcproperties.setProperty("tcpListenHost",address);
- }catch ( Exception x ){
- log.error("Unable to set listen address",x);
- }
- }
+
public void setExpireSessionsOnShutdown(boolean expireSessionsOnShutdown){
this.expireSessionsOnShutdown = expireSessionsOnShutdown;
@@ -669,17 +532,6 @@
}
- public void setTcpListenPort(int port) {
- this.tcpPort = port;
- svcproperties.setProperty("tcpListenPort",String.valueOf(port));
- }
-
- public void setTcpSelectorTimeout(long timeout) {
- this.tcpSelectorTimeout = timeout;
- }
-
-
-
public void messageDataReceived(byte[] data) {
try {
ReplicationStream stream =
@@ -790,7 +642,7 @@
public void stop(String contextPath) throws IOException {
return;
}
-
+
public Log getLogger() {
return log;
}
@@ -808,17 +660,36 @@
log.debug("Calc msg send time total="+msgSendTime+"ms num
request="+nrOfMsgsReceived+" average per msg="+(msgSendTime/nrOfMsgsReceived)+"ms.");
}
}
- public boolean getIsJdk13() {
- return isJdk13;
- }
- public void setIsJdk13(boolean isJdk13) {
- this.isJdk13 = isJdk13;
- }
+
public String getManagerClassName() {
return managerClassName;
}
public void setManagerClassName(String managerClassName) {
this.managerClassName = managerClassName;
}
+ public org.apache.catalina.cluster.ClusterSender getClusterSender() {
+ return clusterSender;
+ }
+ public void setClusterSender(org.apache.catalina.cluster.ClusterSender
clusterSender) {
+ this.clusterSender = clusterSender;
+ }
+ public org.apache.catalina.cluster.ClusterReceiver getClusterReceiver() {
+ return clusterReceiver;
+ }
+ public void setClusterReceiver(org.apache.catalina.cluster.ClusterReceiver
clusterReceiver) {
+ this.clusterReceiver = clusterReceiver;
+ }
+ public MembershipService getMembershipService() {
+ return membershipService;
+ }
+ public void setMembershipService(MembershipService membershipService) {
+ this.membershipService = membershipService;
+ }
+
+ public void addValve(Valve valve) {
+ this.valve = valve;
+ }
+
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [EMAIL PROTECTED]
For additional commands, e-mail: [EMAIL PROTECTED]
