mirror of https://github.com/apache/activemq.git
Make it easier for ActiveCluster to be used by other JMS implementations -
added patches supplied by Ben Kibler git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@382378 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
5fbce8982c
commit
3182029ef4
|
@ -19,6 +19,7 @@
|
|||
package org.apache.activecluster;
|
||||
|
||||
import javax.jms.Destination;
|
||||
import javax.jms.JMSException;
|
||||
|
||||
/**
|
||||
* A simple marshaller for Destinations
|
||||
|
@ -33,7 +34,7 @@ public interface DestinationMarshaller {
|
|||
*
|
||||
* @return the destination to send messages to all members of the cluster
|
||||
*/
|
||||
public Destination getDestination(String destinationName);
|
||||
public Destination getDestination(String destinationName) throws JMSException;
|
||||
|
||||
/**
|
||||
* Gets a destination's physical name
|
||||
|
|
|
@ -70,7 +70,7 @@ public class DefaultClusterFactory implements ClusterFactory {
|
|||
public Cluster createCluster(String name,Destination groupDestination) throws JMSException {
|
||||
Connection connection = getConnectionFactory().createConnection();
|
||||
Session session = createSession(connection);
|
||||
return createCluster(connection, session, name,groupDestination,new DefaultDestinationMarshaller());
|
||||
return createCluster(connection, session, name,groupDestination,new DefaultDestinationMarshaller(session));
|
||||
}
|
||||
|
||||
public Cluster createCluster(String name,Destination groupDestination,DestinationMarshaller marshaller) throws JMSException {
|
||||
|
@ -83,7 +83,7 @@ public class DefaultClusterFactory implements ClusterFactory {
|
|||
public Cluster createCluster(String name,String groupDestinationName) throws JMSException{
|
||||
Connection connection = getConnectionFactory().createConnection();
|
||||
Session session = createSession(connection);
|
||||
return createCluster(connection, session, name,session.createTopic(groupDestinationName),new DefaultDestinationMarshaller());
|
||||
return createCluster(connection, session, name,session.createTopic(groupDestinationName),new DefaultDestinationMarshaller(session));
|
||||
}
|
||||
|
||||
public Cluster createCluster(String name,String groupDestinationName,DestinationMarshaller marshaller) throws JMSException{
|
||||
|
@ -160,9 +160,7 @@ public class DefaultClusterFactory implements ClusterFactory {
|
|||
this.deliveryMode = deliveryMode;
|
||||
}
|
||||
|
||||
// Implementation methods
|
||||
//-------------------------------------------------------------------------
|
||||
protected Cluster createCluster(Connection connection,Session session,String name,Destination groupDestination,
|
||||
public Cluster createCluster(Connection connection,Session session,String name,Destination groupDestination,
|
||||
DestinationMarshaller marshaller) throws JMSException{
|
||||
String dataDestination = dataTopicPrefix + marshaller.getDestinationName(groupDestination);
|
||||
log.info("Creating cluster group producer on topic: "+groupDestination);
|
||||
|
|
|
@ -18,14 +18,19 @@
|
|||
|
||||
package org.apache.activecluster.impl;
|
||||
|
||||
|
||||
|
||||
import java.util.Map;
|
||||
import javax.jms.Destination;
|
||||
import javax.jms.JMSException;
|
||||
import javax.jms.Queue;
|
||||
import javax.jms.Session;
|
||||
import javax.jms.Topic;
|
||||
import org.apache.activemq.command.ActiveMQTopic;
|
||||
|
||||
import org.apache.activecluster.DestinationMarshaller;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
/**
|
||||
* A simple marshaller for Destinations
|
||||
|
@ -34,14 +39,35 @@ import org.apache.commons.logging.LogFactory;
|
|||
*/
|
||||
public class DefaultDestinationMarshaller implements DestinationMarshaller {
|
||||
private final static Log log = LogFactory.getLog(DefaultDestinationMarshaller.class);
|
||||
|
||||
/**
|
||||
* Keep a cache of name to destination mappings for fast lookup.
|
||||
*/
|
||||
private final Map destinations = new ConcurrentHashMap();
|
||||
/**
|
||||
* The active session used to create a new Destination from a name.
|
||||
*/
|
||||
private final Session session;
|
||||
|
||||
/**
|
||||
* Create a marshaller for this specific session.
|
||||
* @param session the session to use when mapping destinations.
|
||||
*/
|
||||
public DefaultDestinationMarshaller(Session session) {
|
||||
this.session = session;
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds a destination from a destinationName
|
||||
* @param destinationName
|
||||
*
|
||||
* @return the destination to send messages to all members of the cluster
|
||||
*/
|
||||
public Destination getDestination(String destinationName){
|
||||
return new ActiveMQTopic(destinationName);
|
||||
public Destination getDestination(String destinationName) throws JMSException {
|
||||
if (!destinations.containsKey(destinationName)) {
|
||||
destinations.put(destinationName, session.createTopic(destinationName));
|
||||
}
|
||||
return (Destination) destinations.get(destinationName);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
@ -23,6 +23,7 @@ import java.io.ObjectOutput;
|
|||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import javax.jms.Destination;
|
||||
import javax.jms.JMSException;
|
||||
|
||||
import org.apache.activecluster.DestinationMarshaller;
|
||||
import org.apache.activecluster.Node;
|
||||
|
@ -45,8 +46,9 @@ public class NodeImpl implements Node{
|
|||
* Construct an Node from a NodeState
|
||||
* @param nodeState
|
||||
* @param marshaller
|
||||
* @throws JMSException
|
||||
*/
|
||||
public NodeImpl(NodeState nodeState,DestinationMarshaller marshaller){
|
||||
public NodeImpl(NodeState nodeState,DestinationMarshaller marshaller) throws JMSException{
|
||||
this(nodeState.getName(),marshaller.getDestination(nodeState.getDestinationName()),nodeState.getState());
|
||||
}
|
||||
/**
|
||||
|
|
|
@ -0,0 +1,75 @@
|
|||
/**
|
||||
*
|
||||
* Copyright 2005-2006 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.activecluster.impl;
|
||||
|
||||
|
||||
|
||||
import javax.jms.Destination;
|
||||
import javax.jms.JMSException;
|
||||
import javax.jms.Queue;
|
||||
import javax.jms.Topic;
|
||||
import org.apache.activecluster.DestinationMarshaller;
|
||||
import org.apache.activemq.command.ActiveMQTopic;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
/**
|
||||
* A simple marshaller for Destinations
|
||||
*
|
||||
* @version $Revision: 1.5 $
|
||||
*/
|
||||
public class SimpleDestinationMarshaller implements DestinationMarshaller {
|
||||
private final static Log log = LogFactory.getLog(SimpleDestinationMarshaller.class);
|
||||
/**
|
||||
* Builds a destination from a destinationName
|
||||
* @param destinationName
|
||||
*
|
||||
* @return the destination to send messages to all members of the cluster
|
||||
*/
|
||||
public Destination getDestination(String destinationName){
|
||||
return new ActiveMQTopic(destinationName);
|
||||
}
|
||||
|
||||
/**
|
||||
* Gets a destination's physical name
|
||||
* @param destination
|
||||
* @return the destination's physical name
|
||||
*/
|
||||
public String getDestinationName(Destination destination){
|
||||
String result = null;
|
||||
if (destination != null){
|
||||
if (destination instanceof Topic){
|
||||
Topic topic = (Topic) destination;
|
||||
try{
|
||||
result = topic.getTopicName();
|
||||
}catch(JMSException e){
|
||||
log.error("Failed to get topic name for " + destination,e);
|
||||
}
|
||||
}else{
|
||||
Queue queue = (Queue) destination;
|
||||
try{
|
||||
result = queue.getQueueName();
|
||||
}catch(JMSException e){
|
||||
log.error("Failed to get queue name for " + destination,e);
|
||||
}
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
|
@ -17,6 +17,7 @@
|
|||
package org.apache.activecluster.group;
|
||||
|
||||
import java.util.List;
|
||||
import javax.jms.JMSException;
|
||||
|
||||
import org.apache.activecluster.group.BuddyGroupModel;
|
||||
import org.apache.activecluster.group.Group;
|
||||
|
@ -60,7 +61,7 @@ public class BuddyGroupModelTest extends GroupTestSupport {
|
|||
|
||||
}
|
||||
|
||||
public void testRemoveGroups() {
|
||||
public void testRemoveGroups() throws JMSException {
|
||||
String[] nodeNames = {"a", "b", "c"};
|
||||
addNodes(nodeNames);
|
||||
|
||||
|
|
|
@ -17,6 +17,7 @@
|
|||
package org.apache.activecluster.group;
|
||||
|
||||
import java.util.List;
|
||||
import javax.jms.JMSException;
|
||||
|
||||
import org.apache.activecluster.group.Group;
|
||||
|
||||
|
@ -50,7 +51,7 @@ public class GroupModelTest extends GroupTestSupport {
|
|||
assertIncomplete(group);
|
||||
}
|
||||
|
||||
public void testRemoveGroups() {
|
||||
public void testRemoveGroups() throws JMSException {
|
||||
String[] nodeNames = {"a", "b", "c"};
|
||||
addNodes(nodeNames);
|
||||
|
||||
|
|
|
@ -19,18 +19,15 @@ package org.apache.activecluster.group;
|
|||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import javax.jms.JMSException;
|
||||
import junit.framework.TestCase;
|
||||
|
||||
import org.apache.activecluster.Cluster;
|
||||
import org.apache.activecluster.ClusterEvent;
|
||||
import org.apache.activecluster.ClusterListener;
|
||||
import org.apache.activecluster.DestinationMarshaller;
|
||||
import org.apache.activecluster.Node;
|
||||
import org.apache.activecluster.group.Group;
|
||||
import org.apache.activecluster.group.GroupClusterListener;
|
||||
import org.apache.activecluster.group.GroupModel;
|
||||
import org.apache.activecluster.impl.DefaultDestinationMarshaller;
|
||||
import org.apache.activecluster.impl.NodeImpl;
|
||||
import org.apache.activecluster.impl.SimpleDestinationMarshaller;
|
||||
|
||||
/**
|
||||
* A base class for Group model testing
|
||||
|
@ -43,16 +40,16 @@ public abstract class GroupTestSupport extends TestCase {
|
|||
private ClusterListener listener;
|
||||
private Cluster cluster;
|
||||
private Map nodes = new HashMap();
|
||||
private DestinationMarshaller marshaller = new DefaultDestinationMarshaller();
|
||||
private DestinationMarshaller marshaller = new SimpleDestinationMarshaller();
|
||||
|
||||
protected void addNodes(String[] nodeNames) {
|
||||
protected void addNodes(String[] nodeNames) throws JMSException {
|
||||
for (int i = 0; i < nodeNames.length; i++) {
|
||||
String nodeName = nodeNames[i];
|
||||
addNode(nodeName);
|
||||
}
|
||||
}
|
||||
|
||||
protected void addNode(String nodeName) {
|
||||
protected void addNode(String nodeName) throws JMSException {
|
||||
|
||||
Node node = new NodeImpl(nodeName,marshaller.getDestination(nodeName));
|
||||
nodes.put(nodeName, node);
|
||||
|
|
Loading…
Reference in New Issue