You are viewing a plain text version of this content. The canonical link for it is here.
Posted to axis-cvs@ws.apache.org by az...@apache.org on 2008/07/02 12:08:01 UTC
svn commit: r673345 - in
/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering:
./ control/wka/ tribes/
Author: azeez
Date: Wed Jul 2 03:08:00 2008
New Revision: 673345
URL: http://svn.apache.org/viewvc?rev=673345&view=rev
Log:
1. Introduced separate domain specific membership & intialization RPC channels
2. Handling WKA mode with load balancers
Modified:
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/MembershipScheme.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/MemberJoinedCommand.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/MemberListCommand.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/RpcMembershipRequestHandler.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ApplicationMode.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/LoadBalancerMode.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/MembershipManager.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/Mode.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesClusterManager.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesMembershipListener.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesUtil.java
webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/WkaBasedMembershipScheme.java
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/MembershipScheme.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/MembershipScheme.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/MembershipScheme.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/MembershipScheme.java Wed Jul 2 03:08:00 2008
@@ -34,4 +34,5 @@
* @throws ClusteringFault If an error occurs while joining the group
*/
void joinGroup() throws ClusteringFault;
+
}
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/MemberJoinedCommand.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/MemberJoinedCommand.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/MemberJoinedCommand.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/MemberJoinedCommand.java Wed Jul 2 03:08:00 2008
@@ -41,6 +41,10 @@
this.member = member;
}
+ public Member getMember() {
+ return member;
+ }
+
public void execute(ConfigurationContext configurationContext) throws ClusteringFault {
Member localMember = membershipManager.getLocalMember();
if (localMember == null || !(Arrays.equals(localMember.getHost(), member.getHost()) &&
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/MemberListCommand.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/MemberListCommand.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/MemberListCommand.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/MemberListCommand.java Wed Jul 2 03:08:00 2008
@@ -48,6 +48,9 @@
public void execute(ConfigurationContext configurationContext) throws ClusteringFault {
log.info("Received MEMBER_LIST message");
+ if(log.isDebugEnabled()){
+ log.debug("MembershipManager#domain: " + new String(membershipManager.getDomain()));
+ }
Member localMember = membershipManager.getLocalMember();
for (Member member : members) {
addMember(localMember, member);
@@ -56,8 +59,9 @@
private void addMember(Member localMember, Member member) {
log.info("Trying to add member " + TribesUtil.getName(member) + "...");
- if (!(Arrays.equals(localMember.getHost(), member.getHost()) &&
- localMember.getPort() == member.getPort())) {
+ if (localMember == null ||
+ (!(Arrays.equals(localMember.getHost(), member.getHost()) &&
+ localMember.getPort() == member.getPort()))) {
log.info("Added member " + TribesUtil.getName(member));
membershipManager.memberAdded(member);
}
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/RpcMembershipRequestHandler.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/RpcMembershipRequestHandler.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/RpcMembershipRequestHandler.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/control/wka/RpcMembershipRequestHandler.java Wed Jul 2 03:08:00 2008
@@ -20,8 +20,10 @@
package org.apache.axis2.clustering.control.wka;
import org.apache.axis2.clustering.ClusteringFault;
+import org.apache.axis2.clustering.MembershipScheme;
import org.apache.axis2.clustering.tribes.MembershipManager;
import org.apache.axis2.clustering.tribes.TribesUtil;
+import org.apache.axis2.clustering.tribes.WkaBasedMembershipScheme;
import org.apache.catalina.tribes.Member;
import org.apache.catalina.tribes.RemoteProcessException;
import org.apache.catalina.tribes.group.RpcCallback;
@@ -38,32 +40,47 @@
private static Log log = LogFactory.getLog(RpcMembershipRequestHandler.class);
private MembershipManager membershipManager;
+ private WkaBasedMembershipScheme membershipScheme;
- public RpcMembershipRequestHandler(MembershipManager membershipManager) {
+ public RpcMembershipRequestHandler(MembershipManager membershipManager,
+ WkaBasedMembershipScheme membershipScheme) {
this.membershipManager = membershipManager;
+ this.membershipScheme = membershipScheme;
}
public Serializable replyRequest(Serializable msg, Member sender) {
String domain = new String(sender.getDomain());
+
if (log.isDebugEnabled()) {
log.debug("Membership request received by RpcMembershipRequestHandler for domain " +
domain);
+ log.debug("local domain: " + new String(membershipManager.getDomain()));
}
if (msg instanceof JoinGroupCommand) {
- log.info("Received JOIN message from " + TribesUtil.getName(sender) +
- TribesUtil.getName(sender) + " in domain " + domain);
+ log.info("Received JOIN message from " + TribesUtil.getName(sender));
membershipManager.memberAdded(sender);
+ // do something specific for the membership scheme
+ membershipScheme.processJoin(sender);
+
// Return the list of current members to the caller
MemberListCommand memListCmd = new MemberListCommand();
memListCmd.setMembers(membershipManager.getMembers());
+ if(log.isDebugEnabled()){
+ log.debug("Sent MEMBER_LIST to " + TribesUtil.getName(sender));
+ }
return memListCmd;
} else if (msg instanceof MemberJoinedCommand) {
- log.info("Received MEMBER_JOINED message from " + TribesUtil.getName(sender) +
- TribesUtil.getName(sender) + " in domain " + domain);
+ log.info("Received MEMBER_JOINED message from " + TribesUtil.getName(sender));
+ MemberJoinedCommand command = (MemberJoinedCommand) msg;
+
+ // do something specific for the membership scheme
+ if (sender.equals(command.getMember())) { // only if the sender is the member who joine, we need to do some special processing
+ membershipScheme.processJoin(sender); //TODO: This may not be necessary
+ }
+
try {
- MemberJoinedCommand command = (MemberJoinedCommand) msg;
command.setMembershipManager(membershipManager);
command.execute(null);
} catch (ClusteringFault e) {
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ApplicationMode.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ApplicationMode.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ApplicationMode.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/ApplicationMode.java Wed Jul 2 03:08:00 2008
@@ -16,6 +16,7 @@
package org.apache.axis2.clustering.tribes;
import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.Member;
import org.apache.catalina.tribes.group.interceptors.DomainFilterInterceptor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -31,9 +32,11 @@
private static final Log log = LogFactory.getLog(LoadBalancerMode.class);
private byte[] loadBalancerDomain;
+ private MembershipManager membershipManager;
- public ApplicationMode(byte[] loadBalancerDomain) {
+ public ApplicationMode(byte[] loadBalancerDomain, MembershipManager membershipManager) {
this.loadBalancerDomain = loadBalancerDomain;
+ this.membershipManager = membershipManager;
}
public void addInterceptors(Channel channel) {
@@ -53,4 +56,13 @@
public List<MembershipManager> getMembershipManagers() {
return new ArrayList<MembershipManager>();
}
+
+ public void notifyMemberJoin(final Member member) {
+ Thread th = new Thread(){
+ public void run() {
+ membershipManager.sendMemberJoinedToAll(member);
+ }
+ };
+ th.start();
+ }
}
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/LoadBalancerMode.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/LoadBalancerMode.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/LoadBalancerMode.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/LoadBalancerMode.java Wed Jul 2 03:08:00 2008
@@ -16,12 +16,18 @@
package org.apache.axis2.clustering.tribes;
import org.apache.axis2.clustering.LoadBalanceEventHandler;
+import org.apache.axis2.clustering.control.wka.MemberJoinedCommand;
import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.ChannelException;
+import org.apache.catalina.tribes.Member;
import org.apache.catalina.tribes.MembershipListener;
+import org.apache.catalina.tribes.RemoteProcessException;
+import org.apache.catalina.tribes.group.RpcChannel;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
@@ -35,11 +41,14 @@
private byte[] loadBalancerDomain;
private Map<String, LoadBalanceEventHandler> lbEventHandlers;
private List<MembershipManager> membershipManagers = new ArrayList<MembershipManager>();
+ private MembershipManager primaryMembershipManager;
public LoadBalancerMode(byte[] loadBalancerDomain,
- Map<String, LoadBalanceEventHandler> lbEventHandlers) {
+ Map<String, LoadBalanceEventHandler> lbEventHandlers,
+ MembershipManager primaryMembershipManager) {
this.loadBalancerDomain = loadBalancerDomain;
this.lbEventHandlers = lbEventHandlers;
+ this.primaryMembershipManager = primaryMembershipManager;
}
public void addInterceptors(Channel channel) {
@@ -78,4 +87,40 @@
public List<MembershipManager> getMembershipManagers() {
return membershipManagers;
}
+
+ public void notifyMemberJoin(final Member member) { //TODO: ### rename this method
+
+ if (Arrays.equals(loadBalancerDomain, member.getDomain())) { // A peer load balancer has joined. Need to send it the entire member lists
+ for (MembershipManager manager : membershipManagers) {
+ manager.sendMemberList(member);
+ }
+ } else { // An application member has joined.
+
+ // Need to notify all members in the group of the new app member
+ Thread th = new Thread() {
+ public void run() {
+ for (MembershipManager manager : membershipManagers) {
+ if (Arrays.equals(manager.getDomain(), member.getDomain())) {
+ manager.sendMemberJoinedToAll(member);
+ MemberJoinedCommand cmd = new MemberJoinedCommand();
+ cmd.setMember(member);
+ try {
+ manager.getRpcMembershipChannel().send(primaryMembershipManager.getMembers(),
+ cmd,
+ RpcChannel.ALL_REPLY, Channel.SEND_OPTIONS_ASYNCHRONOUS,
+ 10000);
+ } catch (ChannelException e) {
+ String errMsg = "Could not send MEMBER_JOINED[" + TribesUtil.getName(member) +
+ "] to all load balancer members ";
+ log.error(errMsg, e);
+ throw new RemoteProcessException(errMsg, e);
+ }
+ break;
+ }
+ }
+ }
+ };
+ th.start();
+ }
+ }
}
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/MembershipManager.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/MembershipManager.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/MembershipManager.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/MembershipManager.java Wed Jul 2 03:08:00 2008
@@ -21,6 +21,7 @@
import org.apache.axis2.clustering.ClusteringConstants;
import org.apache.axis2.clustering.LoadBalanceEventHandler;
+import org.apache.axis2.clustering.control.wka.MemberJoinedCommand;
import org.apache.axis2.clustering.control.wka.MemberListCommand;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.catalina.tribes.Channel;
@@ -68,6 +69,10 @@
this.rpcMembershipChannel = rpcMembershipChannel;
}
+ public RpcChannel getRpcMembershipChannel() {
+ return rpcMembershipChannel;
+ }
+
public void setStaticMembershipInterceptor(
StaticMembershipInterceptor staticMembershipInterceptor) {
this.staticMembershipInterceptor = staticMembershipInterceptor;
@@ -127,15 +132,14 @@
// If this member already exists or if the member belongs to another domain,
// there is no need to add it
- if(members.contains(member) || !Arrays.equals(domain, member.getDomain())){
+ if (members.contains(member) || !Arrays.equals(domain, member.getDomain())) {
return false;
}
if (staticMembershipInterceptor != null) { // this interceptor is null when multicast based scheme is used
staticMembershipInterceptor.addStaticMember(member);
if (log.isDebugEnabled()) {
- log.debug("Added static member " + TribesUtil.getName(member) +
- " from domain " + new String(member.getDomain()));
+ log.debug("Added static member " + TribesUtil.getName(member));
}
}
@@ -163,7 +167,7 @@
members.add(localMember); // Need to set the local member too
memListCmd.setMembers(members.toArray(new Member[members.size()]));
rpcMembershipChannel.send(new Member[]{member}, memListCmd, RpcChannel.ALL_REPLY,
- Channel.SEND_OPTIONS_ASYNCHRONOUS, 10000);
+ Channel.SEND_OPTIONS_ASYNCHRONOUS, 10000);
} catch (Exception e) {
String errMsg = "Could not send MEMBER_LIST to well-known member " +
TribesUtil.getName(member);
@@ -181,6 +185,58 @@
return false;
}
+ /**
+ * Send the list of members to the <code>member</code>
+ *
+ * @param member The member to whom the member list has to be sent
+ */
+ public void sendMemberList(Member member) {
+ try {
+ MemberListCommand memListCmd = new MemberListCommand();
+ List<Member> members = new ArrayList<Member>(this.members);
+ memListCmd.setMembers(members.toArray(new Member[members.size()]));
+ rpcMembershipChannel.send(new Member[]{member}, memListCmd, RpcChannel.ALL_REPLY,
+ Channel.SEND_OPTIONS_ASYNCHRONOUS, 10000);
+ if (log.isDebugEnabled()) {
+ log.debug("Sent MEMBER_LIST to " + TribesUtil.getName(member));
+ }
+ } catch (Exception e) {
+ String errMsg = "Could not send MEMBER_LIST to member " + TribesUtil.getName(member);
+ log.error(errMsg, e);
+ throw new RemoteProcessException(errMsg, e);
+ }
+ }
+
+ /**
+ * Inform all members that a particular member just joined
+ *
+ * @param member The member who just joined
+ */
+ public void sendMemberJoinedToAll(Member member) {
+ try {
+
+ MemberJoinedCommand cmd = new MemberJoinedCommand();
+ cmd.setMember(member);
+ ArrayList<Member> membersToSend = (ArrayList<Member>) (((ArrayList) members).clone());
+ membersToSend.remove(member); // Do not send MEMBER_JOINED to the new member who just joined
+
+ if (membersToSend.size() > 0) {
+ rpcMembershipChannel.send(membersToSend.toArray(new Member[membersToSend.size()]), cmd,
+ RpcChannel.ALL_REPLY, Channel.SEND_OPTIONS_ASYNCHRONOUS,
+ 10000);
+ if (log.isDebugEnabled()) {
+ log.debug("Sent MEMBER_JOINED[" + TribesUtil.getName(member) +
+ "] to all members in domain " + new String(domain));
+ }
+ }
+ } catch (Exception e) {
+ String errMsg = "Could not send MEMBER_JOINED[" + TribesUtil.getName(member) +
+ "] to all members ";
+ log.error(errMsg, e);
+ throw new RemoteProcessException(errMsg, e);
+ }
+ }
+
private org.apache.axis2.clustering.Member toAxis2Member(Member member) {
org.apache.axis2.clustering.Member axis2Member =
new org.apache.axis2.clustering.Member(TribesUtil.getHost(member),
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/Mode.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/Mode.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/Mode.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/Mode.java Wed Jul 2 03:08:00 2008
@@ -16,6 +16,7 @@
package org.apache.axis2.clustering.tribes;
import org.apache.catalina.tribes.Channel;
+import org.apache.catalina.tribes.Member;
import java.util.List;
@@ -38,5 +39,17 @@
*/
void init(Channel channel);
+ /**
+ * Get all the membership managers associated with a particular mode
+ *
+ * @return membership managers associated with a particular mode
+ */
List<MembershipManager> getMembershipManagers();
+
+ /**
+ * Notify to the relevant parties that a member <code>member</code> joined
+ *
+ * @param member The member to whom this member lists have to be sent
+ */
+ void notifyMemberJoin(Member member);
}
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesClusterManager.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesClusterManager.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesClusterManager.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesClusterManager.java Wed Jul 2 03:08:00 2008
@@ -80,7 +80,7 @@
private ConfigurationContext configurationContext;
private ChannelListener channelListener;
private ChannelSender channelSender;
- private MembershipManager membershipManager;
+ private MembershipManager primaryMembershipManager;
private RpcInitializationRequestHandler rpcInitRequestHandler;
private MembershipScheme membershipScheme;
@@ -138,17 +138,17 @@
public void init() throws ClusteringFault {
log.info("Initializing cluster...");
addRequestBlockingHandlerToInFlows();
- membershipManager = new MembershipManager(configurationContext);
+ primaryMembershipManager = new MembershipManager(configurationContext);
channel = new GroupChannel();
- channelSender = new ChannelSender(channel, membershipManager, synchronizeAllMembers());
+ channelSender = new ChannelSender(channel, primaryMembershipManager, synchronizeAllMembers());
channelListener =
new ChannelListener(configurationContext, configurationManager, contextManager);
channel.addChannelListener(channelListener);
byte[] domain = getClusterDomain();
log.info("Cluster domain: " + new String(domain));
- membershipManager.setDomain(domain);
+ primaryMembershipManager.setDomain(domain);
// RpcChannel is a ChannelListener. When the reply to a particular request comes back, it
// picks it up. Each RPC is given a UUID, hence can correlate the request-response pair
@@ -165,7 +165,7 @@
configureMembershipScheme(domain, mode.getMembershipManagers());
setMemberTransportInfo();
- TribesMembershipListener membershipListener = new TribesMembershipListener(membershipManager);
+ TribesMembershipListener membershipListener = new TribesMembershipListener(primaryMembershipManager);
channel.addMembershipListener(membershipListener);
try {
channel.start(Channel.DEFAULT); // At this point, this member joins the group
@@ -183,7 +183,7 @@
}
log.info("Local Member " + TribesUtil.getLocalHost(channel));
- TribesUtil.printMembers(membershipManager);
+ TribesUtil.printMembers(primaryMembershipManager);
membershipScheme.joinGroup();
@@ -343,9 +343,9 @@
private void configureMode(byte[] domain) {
if (loadBalanceMode) {
- mode = new LoadBalancerMode(domain, lbEventHandlers);
+ mode = new LoadBalancerMode(domain, lbEventHandlers, primaryMembershipManager);
} else {
- mode = new ApplicationMode(domain);
+ mode = new ApplicationMode(domain, primaryMembershipManager);
}
mode.init(channel);
}
@@ -366,7 +366,7 @@
if (scheme.equals(ClusteringConstants.MembershipScheme.WKA_BASED)) {
membershipScheme = new WkaBasedMembershipScheme(channel, mode,
membershipManagers,
- membershipManager,
+ primaryMembershipManager,
parameters, localDomain, members);
} else if (scheme.equals(ClusteringConstants.MembershipScheme.MULTICAST_BASED)) {
membershipScheme = new MulticastBasedMembershipScheme(channel, mode, parameters, localDomain);
@@ -397,15 +397,15 @@
// Do not send another request to these members
List<String> sentMembersList = new ArrayList<String>();
sentMembersList.add(TribesUtil.getLocalHost(channel));
- Member[] members = membershipManager.getMembers();
+ Member[] members = primaryMembershipManager.getMembers();
if (members.length == 0) {
return;
}
while (members.length > 0 && numberOfTries < 5) {
Member member = (numberOfTries == 0) ?
- membershipManager.getLongestLivingMember() : // First try to get from the longest member alive
- membershipManager.getRandomMember(); // Else get from a random member
+ primaryMembershipManager.getLongestLivingMember() : // First try to get from the longest member alive
+ primaryMembershipManager.getRandomMember(); // Else get from a random member
String memberHost = TribesUtil.getName(member);
log.info("Trying to send intialization request to " + memberHost);
try {
@@ -440,7 +440,7 @@
}
}
numberOfTries++;
- members = membershipManager.getMembers();
+ members = primaryMembershipManager.getMembers();
if (numberOfTries >= members.length) {
break;
}
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesMembershipListener.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesMembershipListener.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesMembershipListener.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesMembershipListener.java Wed Jul 2 03:08:00 2008
@@ -24,7 +24,7 @@
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-/**
+/**
* Membership changes are notified using this class
*/
public class TribesMembershipListener implements MembershipListener {
@@ -40,7 +40,7 @@
if (membershipManager.memberAdded(member)) {
log.info("New member " + TribesUtil.getName(member) + " joined cluster.");
}
- // System.err.println("++++++ IS COORD="+TribesClusterManager.nbc.isCoordinator());
+ // System.err.println("++++++ IS COORD="+TribesClusterManager.nbc.isCoordinator());
}
public void memberDisappeared(Member member) {
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesUtil.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesUtil.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesUtil.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/TribesUtil.java Wed Jul 2 03:08:00 2008
@@ -44,7 +44,7 @@
}
public static String getName(Member member) {
- return getHost(member) + ":" + member.getPort();
+ return getHost(member) + ":" + member.getPort() + "(" + new String(member.getDomain())+ ")";
}
public static String getHost(Member member) {
Modified: webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/WkaBasedMembershipScheme.java
URL: http://svn.apache.org/viewvc/webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/WkaBasedMembershipScheme.java?rev=673345&r1=673344&r2=673345&view=diff
==============================================================================
--- webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/WkaBasedMembershipScheme.java (original)
+++ webservices/axis2/trunk/java/modules/clustering/src/org/apache/axis2/clustering/tribes/WkaBasedMembershipScheme.java Wed Jul 2 03:08:00 2008
@@ -19,13 +19,11 @@
import org.apache.axis2.clustering.Member;
import org.apache.axis2.clustering.MembershipScheme;
import org.apache.axis2.clustering.control.wka.JoinGroupCommand;
-import org.apache.axis2.clustering.control.wka.MemberJoinedCommand;
import org.apache.axis2.clustering.control.wka.MemberListCommand;
import org.apache.axis2.clustering.control.wka.RpcMembershipRequestHandler;
import org.apache.axis2.description.Parameter;
import org.apache.axis2.util.Utils;
import org.apache.catalina.tribes.Channel;
-import org.apache.catalina.tribes.ChannelException;
import org.apache.catalina.tribes.ManagedChannel;
import org.apache.catalina.tribes.group.Response;
import org.apache.catalina.tribes.group.RpcChannel;
@@ -354,9 +352,10 @@
// Create an RpcChannel for each localDomain
String domain = new String(appDomainMembershipManager.getDomain());
RpcChannel rpcMembershipChannel =
- new RpcChannel(TribesUtil.getRpcMembershipChannelId(appDomainMembershipManager.getDomain()),
+ new RpcChannel(TribesUtil.getRpcMembershipChannelId(appDomainMembershipManager.getDomain()),
channel,
- new RpcMembershipRequestHandler(appDomainMembershipManager));
+ new RpcMembershipRequestHandler(appDomainMembershipManager,
+ this));
appDomainMembershipManager.setRpcMembershipChannel(rpcMembershipChannel);
if (log.isDebugEnabled()) {
log.debug("Created RPC Channel for application domain " + domain);
@@ -366,7 +365,8 @@
// Create a Membership channel for handling membership requests
RpcChannel rpcMembershipChannel =
new RpcChannel(TribesUtil.getRpcMembershipChannelId(localDomain),
- channel, new RpcMembershipRequestHandler(primaryMembershipManager));
+ channel, new RpcMembershipRequestHandler(primaryMembershipManager,
+ this));
primaryMembershipManager.setRpcMembershipChannel(rpcMembershipChannel);
// Send JOIN message to a WKA member
@@ -388,7 +388,7 @@
10000);
if (responses.length == 0) {
try {
- if(log.isDebugEnabled()){
+ if (log.isDebugEnabled()) {
log.debug("No responses received");
}
Thread.sleep(500);
@@ -409,8 +409,6 @@
}
while (responses == null || responses.length == 0); // Wait until we've received at least one response
- //TODO: ######## If this node is a LB, it needs to get the entire localDomain to member-list map
-
for (Response response : responses) {
MemberListCommand command = (MemberListCommand) response.getMessage();
command.setMembershipManager(primaryMembershipManager);
@@ -427,28 +425,20 @@
}
}
}
-
- // Send MEMBER_JOINED to the group
- if (primaryMembershipManager.getMembers().length > 0) {
- log.info("Sending MEMBER_JOINED to group...");
- MemberJoinedCommand memberJoinedCommand = new MemberJoinedCommand();
- memberJoinedCommand.setMember(primaryMembershipManager.getLocalMember());
- try {
- rpcMembershipChannel.send(primaryMembershipManager.getMembers(),
- memberJoinedCommand,
- RpcChannel.ALL_REPLY,
- Channel.SEND_OPTIONS_ASYNCHRONOUS |
- TribesConstants.MEMBERSHIP_MSG_OPTION,
- 10000);
- } catch (ChannelException e) {
- String msg = "Could not send MEMBER_JOINED message to group";
- log.error(msg, e);
- throw new ClusteringFault(msg, e);
- }
- }
}
}
+ /**
+ * When a JOIN message is received from some other member, it is notified using this method,
+ * so that membership scheme specific processing can be carried out
+ *
+ * @param member The member who just joined
+ */
+ public void processJoin(org.apache.catalina.tribes.Member member) {
+ mode.notifyMemberJoin(member);
+ }
+
+
public Parameter getParameter(String name) {
return parameters.get(name);
}