User: user57
Date: 01/07/31 22:19:48
Modified: src/main/org/jbossmq/server Tag: jboss_buildmagic
ClientConsumer.java JBossMQService.java
JBossMQServiceMBean.java JMSDestination.java
JMSQueue.java JMSServer.java JMSTopic.java
StateManager.java StateManagerMBean.java
Log:
o updated from HEAD
Revision Changes Path
No revision
No revision
1.10.2.1 +52 -27 jbossmq/src/main/org/jbossmq/server/ClientConsumer.java
Index: ClientConsumer.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jbossmq/server/ClientConsumer.java,v
retrieving revision 1.10
retrieving revision 1.10.2.1
diff -u -r1.10 -r1.10.2.1
--- ClientConsumer.java 2001/07/28 00:30:16 1.10
+++ ClientConsumer.java 2001/08/01 05:19:48 1.10.2.1
@@ -109,6 +109,9 @@
{
cat.debug("Queueing outbound message: "+message);
+ if( !enabled )
+ return;
+
LinkedList l = (LinkedList)destinationSubscriptions.get(
message.getJMSDestination() );
if( l == null )
throw new JMSException("No subscription found for that
destination.");
@@ -138,35 +141,29 @@
cat.debug("Adding subscription for: "+req);
req.dc = dc;
+ JMSDestination
queue=(JMSDestination)server.getJMSDestination(req.destination);
+ if (queue==null) throw new JMSException("This destination does not
exist !");
+
+ SubscriptionState subState = new SubscriptionState( req );
+ subState.dest = queue.addSubscriber(req);
+
synchronized (subscriptions ) {
- SubscriptionState subState = new SubscriptionState( req );
-
HashMap subscriptionsClone = (HashMap)subscriptions.clone();
subscriptionsClone.put(new Integer(req.subscriptionId),
subState );
subscriptions = subscriptionsClone;
LinkedList ll = (LinkedList)destinationSubscriptions.get(
req.destination );
if( ll == null ) {
-
- JMSDestination
queue=(JMSDestination)server.getJMSDestination(req.destination);
- if (queue==null) throw new JMSException("This
destination does not exist !");
-
ll = new LinkedList();
- ll.add( subState );
-
- HashMap destinationSubscriptionsClone =
(HashMap)destinationSubscriptions.clone();
- destinationSubscriptionsClone.put(req.destination, ll
);
- destinationSubscriptions =
destinationSubscriptionsClone;
- subState.dest = queue.addConsumer(req, this);
-
- } else {
- LinkedList llClone = (LinkedList)ll.clone();
- llClone.add( req );
-
- HashMap destinationSubscriptionsClone =
(HashMap)destinationSubscriptions.clone();
- destinationSubscriptionsClone.put(req.destination,
llClone);
- destinationSubscriptions =
destinationSubscriptionsClone;
+ queue.addConsumer(req,this);
}
+ else
+ ll = (LinkedList)ll.clone();
+ ll.add( subState );
+
+ HashMap destinationSubscriptionsClone =
(HashMap)destinationSubscriptions.clone();
+ destinationSubscriptionsClone.put(req.destination, ll );
+ destinationSubscriptions = destinationSubscriptionsClone;
}
}
@@ -243,15 +240,31 @@
JMSDestination queue =
server.getJMSDestination(req.subscription.destination);
if( queue == null )
throw new JMSException("The subscription's destination does
not exist");
+
+ if( enabled && req.subscription.actsLikeAQueue ) {
+ SpyMessage message = req.dest.receiveNoWait(req.subscription);
- // Is it a receiveNoWait()
- if( wait == -1 && req.subscription.actsLikeAQueue ) {
- return req.dest.receiveNoWait(req.subscription);
+ if( message != null ) {
+ AcknowledgementRequest ack = new
AcknowledgementRequest();
+ ack.destination = message.getJMSDestination();
+ ack.messageID = message.getJMSMessageID();
+ ack.subscriberId = subscriberId;
+ ack.isAck = false;
+
+ synchronized (unacknowledgedMessages) {
+ unacknowledgedMessages.put(ack, message);
+ }
+ return message;
+ }
+ }
+
+ // If not receiveNoWait()
+ if( wait != -1 ) {
+ // Notify the queue. It could be waiting for a consumer
+ req.subscription.receiving = true;
+ req.dest.notifyMessageAvailable();
}
- // Notify the queue. It could be waiting for a consumer
- req.subscription.receiving = true;
- req.dest.notifyMessageAvailable();
return null;
}
@@ -330,6 +343,9 @@
cat.debug(""+this+"->scanExclusiveQueue(queue="+queue+")");
+ if( !enabled )
+ return false;
+
Iterator i = queue.messages.iterator();
while( i.hasNext() ) {
@@ -381,6 +397,15 @@
public void setEnabled(boolean enabled) {
cat.debug(""+this+"->setEnabled(enabled="+enabled+")");
this.enabled = enabled;
+ if(enabled){
+ // queues might be waiting for messages.
+ for(Iterator it =
destinationSubscriptions.keySet().iterator();it.hasNext();){
+ SpyDestination destination = (SpyDestination)
it.next();
+ JMSDestination dest =
server.getJMSDestination(destination);
+ if(dest != null)
+ dest.notifyMessageAvailable();
+ }
+ }
}
public String toString() {
1.5.2.1 +181 -217 jbossmq/src/main/org/jbossmq/server/JBossMQService.java
Index: JBossMQService.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jbossmq/server/JBossMQService.java,v
retrieving revision 1.5
retrieving revision 1.5.2.1
diff -u -r1.5 -r1.5.2.1
--- JBossMQService.java 2001/07/28 00:30:16 1.5
+++ JBossMQService.java 2001/08/01 05:19:48 1.5.2.1
@@ -49,7 +49,7 @@
public String getName() {
- return "JBossMQ.Server";
+ return "JBossMQ";
}
@@ -80,50 +80,21 @@
jmsServer = new JMSServer();
}
- public SpyQueue createQueue(String name) throws JMSException
+ public void createQueue(String name) throws Exception
{
- /*
- cat.debug("new queue : "+name);
-
- SpyQueue newQueue=new SpyQueue(name);
- if (destinations.containsKey(newQueue)) throw new JMSException("This
queue already exists !");
-
- JMSDestination queue=new JMSDestination(newQueue,null,this);
-
- InitialContext ctx = null;
- Context subcontext = null;
-
- try{
- //Get an InitialContext
- ctx = new InitialContext();
-
- //get the queues subcontext
- subcontext = (Context)ctx.lookup("queue");
-
- //add this queue to the jndi queue context
- subcontext.rebind(name,newQueue);
+ try {
+
+ ObjectName objectName = new ObjectName(
"JBossMQ:serivce=Queue,name="+name );
+
getServer().createMBean("org.jbossmq.server.QueueManager",objectName);
+ getServer().invoke(objectName,"init", new Object[] {}, new
String[] {});
+ getServer().invoke(objectName,"start", new Object[] {}, new
String[] {});
+
+ } catch ( Exception e ) {
+ category.warn("Could not create the destination: ", e );
+ throw e;
}
- catch(Exception e){
- //remove the queue from the destinations
- synchronized (destinations) {
- HashMap newMap=(HashMap)destinations.clone();
- newMap.remove(newQueue);
- destinations=newMap;
- }
- JMSException newE = new JMSException("Exception unbinding Queue from
the JNDI queue context");
- newE.setLinkedException(e);
- throw newE;
- }
- //Add this new JMSServerQueue to the list
- synchronized (destinations) {
- HashMap newMap=(HashMap)destinations.clone();
- newMap.put(newQueue,queue);
- destinations=newMap;
- }
- */
-
/*
//check to see if this queue already exists in the config structure
boolean queueExists = false;
@@ -173,53 +144,23 @@
*/
//return newQueue;
- return null;
}
// Administration calls
- public SpyTopic createTopic(String name) throws JMSException
+ public void createTopic(String name) throws Exception
{
- /*
- cat.debug("new topic : "+name);
-
- SpyTopic newTopic=new SpyTopic(name);
- if (destinations.containsKey(newTopic)) throw new JMSException("This
topic already exists !");
-
- JMSDestination queue=new JMSDestination(newTopic,null,this);
-
- InitialContext ctx = null;
- Context subcontext = null;
-
- try{
- //Get an InitialContext
- ctx = new InitialContext();
-
- //get the topics subcontext
- subcontext = (Context)ctx.lookup("topic");
-
- //add this topic to the jndi topic context
- subcontext.rebind(name,newTopic);
-
- }
- catch(Exception e){
- //remove the topic from the destinations
- synchronized (destinations) {
- HashMap newMap=(HashMap)destinations.clone();
- newMap.remove(newTopic);
- destinations=newMap;
- }
- JMSException newE = new JMSException("Exception binding Topic to the
JNDI topic context");
- newE.setLinkedException(e);
- throw newE;
- }
-
- //Add this new JMSServerQueue to the list
- synchronized (destinations) {
- HashMap newMap=(HashMap)destinations.clone();
- newMap.put(newTopic,queue);
- destinations=newMap;
+ try {
+
+ ObjectName objectName = new ObjectName(
"JBossMQ:serivce=Topic,name="+name );
+
getServer().createMBean("org.jbossmq.server.TopicManager",objectName);
+ getServer().invoke(objectName,"init", new Object[] {}, new
String[] {});
+ getServer().invoke(objectName,"start", new Object[] {}, new
String[] {});
+
+ } catch ( Exception e ) {
+ category.warn("Could not create the destination: ", e );
+ throw e;
}
- */
+
/*
//check to see if this topic already exists in the config structure
@@ -268,145 +209,168 @@
}
return newTopic;
*/
- return null;
}
-
- public void destroyQueue(String name) throws JMSException
- {
- /*
- cat.debug("destroy queue : "+name);
-
- SpyQueue destroyQueue=new SpyQueue(name);
- if (!destinations.containsKey(destroyQueue)) throw new
JMSException("This queue doesn't exist!");
-
- JMSDestination queue=new JMSDestination(destroyQueue,null,this);
-
- InitialContext ctx = null;
- Context subcontext = null;
- try{
- //Get an InitialContext
- ctx = new InitialContext();
-
- //get the queues subcontext
- subcontext = (Context)ctx.lookup("queue");
-
- //remove this queue to the jndi queue context
- subcontext.unbind(name);
-
- }
- catch(Exception e){
- JMSException newE = new JMSException("Exception unbinding Queue from
the JNDI queue context");
- newE.setLinkedException(e);
- throw newE;
- }
-
- //remove this new JMSServerQueue from the list
- synchronized (destinations) {
- HashMap newMap=(HashMap)destinations.clone();
- newMap.remove(destroyQueue);
- destinations=newMap;
- }
- */
-
- /*
- //check to see if this queue already exists in the config structure
- //and remove it if it does
- XElement element = null;
+ public void destroyQueue(String name) throws Exception {
+ try {
+
+ ObjectName objectName = new ObjectName(
"JBossMQ:serivce=Queue,name="+name );
+ getServer().invoke(objectName,"stop", new Object[] {}, new
String[] {});
+ getServer().invoke(objectName,"destroy", new Object[] {}, new
String[] {});
+ getServer().unregisterMBean(objectName);
+
+ } catch ( Exception e ) {
+ category.warn("Could not destory the destination: ", e );
+ throw e;
+ }
+
+ /*
+ cat.debug("destroy queue : "+name);
+
+ SpyQueue destroyQueue=new SpyQueue(name);
+ if (!destinations.containsKey(destroyQueue)) throw new JMSException("This
queue doesn't exist!");
+
+ JMSDestination queue=new JMSDestination(destroyQueue,null,this);
+
+ InitialContext ctx = null;
+ Context subcontext = null;
+
+ try{
+ //Get an InitialContext
+ ctx = new InitialContext();
+
+ //get the queues subcontext
+ subcontext = (Context)ctx.lookup("queue");
+
+ //remove this queue to the jndi queue context
+ subcontext.unbind(name);
+
+ }
+ catch(Exception e){
+ JMSException newE = new JMSException("Exception unbinding Queue from the
JNDI queue context");
+ newE.setLinkedException(e);
+ throw newE;
+ }
+
+ //remove this new JMSServerQueue from the list
+ synchronized (destinations) {
+ HashMap newMap=(HashMap)destinations.clone();
+ newMap.remove(destroyQueue);
+ destinations=newMap;
+ }
+ */
+
+ /*
+ //check to see if this queue already exists in the config structure
+ //and remove it if it does
+ XElement element = null;
+
+ try{
+ Enumeration enum = serverConfig.getElementsNamed("Queue");
+
+ while( enum.hasMoreElements() ) {
+ element = (XElement)enum.nextElement();
+ if(name.equals(element.getField("Name"))){
+ //remove this queue element from the config structure
+ element.removeFromParent();
+ break;
+ }
+
+ }
+
+ //save the new state in the config file
+ //saveConfig();
+
+
+ }
+ catch(Exception ioe){
+ ResourceAllocationException newE = new
ResourceAllocationException("Exception saving updated configuration state");
+ newE.setLinkedException(ioe);
+ throw newE;
+ }
+ */
- try{
- Enumeration enum = serverConfig.getElementsNamed("Queue");
-
- while( enum.hasMoreElements() ) {
- element = (XElement)enum.nextElement();
- if(name.equals(element.getField("Name"))){
- //remove this queue element from the config structure
- element.removeFromParent();
- break;
- }
-
- }
-
- //save the new state in the config file
- //saveConfig();
-
-
- }
- catch(Exception ioe){
- ResourceAllocationException newE = new
ResourceAllocationException("Exception saving updated configuration state");
- newE.setLinkedException(ioe);
- throw newE;
- }
- */
-
}
-
- public void destroyTopic(String name) throws JMSException
- {
- /*
- cat.debug("destroy topic : "+name);
-
- SpyTopic destroyTopic=new SpyTopic(name);
- if (!destinations.containsKey(destroyTopic)) throw new
JMSException("This topic doesn't exist!");
-
- JMSDestination queue=new JMSDestination(destroyTopic,null,this);
-
- InitialContext ctx = null;
- Context subcontext = null;
- try{
- //Get an InitialContext
- ctx = new InitialContext();
-
- //get the topics subcontext
- subcontext = (Context)ctx.lookup("topic");
-
- //remove this topic to the jndi topic context
- subcontext.unbind(name);
+ public void destroyTopic(String name) throws Exception {
+ try {
+
+ ObjectName objectName = new ObjectName(
"JBossMQ:serivce=Topic,name="+name );
+ getServer().invoke(objectName,"stop", new Object[] {}, new
String[] {});
+ getServer().invoke(objectName,"destroy", new Object[] {}, new
String[] {});
+ getServer().unregisterMBean(objectName);
+
+ } catch ( Exception e ) {
+ category.warn("Could not destory the destination: ", e );
+ throw e;
+ }
+
+ /*
+ cat.debug("destroy queue : "+name);
+
+ SpyQueue destroyQueue=new SpyQueue(name);
+ if (!destinations.containsKey(destroyQueue)) throw new JMSException("This
queue doesn't exist!");
+
+ JMSDestination queue=new JMSDestination(destroyQueue,null,this);
+
+ InitialContext ctx = null;
+ Context subcontext = null;
+
+ try{
+ //Get an InitialContext
+ ctx = new InitialContext();
+
+ //get the queues subcontext
+ subcontext = (Context)ctx.lookup("queue");
+
+ //remove this queue to the jndi queue context
+ subcontext.unbind(name);
+
+ }
+ catch(Exception e){
+ JMSException newE = new JMSException("Exception unbinding Queue from the
JNDI queue context");
+ newE.setLinkedException(e);
+ throw newE;
+ }
+
+ //remove this new JMSServerQueue from the list
+ synchronized (destinations) {
+ HashMap newMap=(HashMap)destinations.clone();
+ newMap.remove(destroyQueue);
+ destinations=newMap;
+ }
+ */
+
+ /*
+ //check to see if this queue already exists in the config structure
+ //and remove it if it does
+ XElement element = null;
+
+ try{
+ Enumeration enum = serverConfig.getElementsNamed("Queue");
+
+ while( enum.hasMoreElements() ) {
+ element = (XElement)enum.nextElement();
+ if(name.equals(element.getField("Name"))){
+ //remove this queue element from the config structure
+ element.removeFromParent();
+ break;
+ }
+
+ }
+
+ //save the new state in the config file
+ //saveConfig();
+
+
+ }
+ catch(Exception ioe){
+ ResourceAllocationException newE = new
ResourceAllocationException("Exception saving updated configuration state");
+ newE.setLinkedException(ioe);
+ throw newE;
+ }
+ */
- }
- catch(Exception e){
- JMSException newE = new JMSException("Exception unbinding Topic from
the JNDI topic context");
- newE.setLinkedException(e);
- throw newE;
- }
-
- //remove this new JMSServerQueue from the list
- synchronized (destinations) {
- HashMap newMap=(HashMap)destinations.clone();
- newMap.remove(destroyTopic);
- destinations=newMap;
- }
- */
-
- /*
- //check to see if this topic already exists in the config structure
- //and remove it if it does
- XElement element = null;
-
- try{
- Enumeration enum = serverConfig.getElementsNamed("Topic");
-
- while( enum.hasMoreElements() ) {
- element = (XElement)enum.nextElement();
- if(name.equals(element.getField("Name"))){
- //remove this topic element from the config structure
- element.removeFromParent();
- break;
- }
-
- }
-
- //save the new state in the config file
- //saveConfig();
- }
- catch(Exception ioe){
- ResourceAllocationException newE = new
ResourceAllocationException("Exception saving updated configuration state");
- newE.setLinkedException(ioe);
- throw newE;
- }
- */
-
}
}
1.6.2.1 +53 -1 jbossmq/src/main/org/jbossmq/server/JBossMQServiceMBean.java
Index: JBossMQServiceMBean.java
===================================================================
RCS file:
/cvsroot/jboss/jbossmq/src/main/org/jbossmq/server/JBossMQServiceMBean.java,v
retrieving revision 1.6
retrieving revision 1.6.2.1
diff -u -r1.6 -r1.6.2.1
--- JBossMQServiceMBean.java 2001/07/28 00:30:16 1.6
+++ JBossMQServiceMBean.java 2001/08/01 05:19:48 1.6.2.1
@@ -75,4 +75,56 @@
// Public --------------------------------------------------------
JMSServer getJMSServer();
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+ public void createQueue(String name) throws Exception;
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+ public void createTopic(String name) throws Exception;
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+ public void destroyQueue(String name) throws Exception;
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+
+ // Public --------------------------------------------------------
+
+ public void destroyTopic(String name) throws Exception;
}
1.8.2.1 +4 -2 jbossmq/src/main/org/jbossmq/server/JMSDestination.java
Index: JMSDestination.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jbossmq/server/JMSDestination.java,v
retrieving revision 1.8
retrieving revision 1.8.2.1
diff -u -r1.8 -r1.8.2.1
--- JMSDestination.java 2001/07/28 00:30:16 1.8
+++ JMSDestination.java 2001/08/01 05:19:48 1.8.2.1
@@ -86,7 +86,9 @@
org.apache.log4j.Category cat;
// Package protected ---------------------------------------------
- abstract JMSDestination addConsumer(Subscription sub, ClientConsumer c) throws
JMSException;
+ abstract void addConsumer(Subscription sub,ClientConsumer c) throws
JMSException;
+
+ abstract JMSDestination addSubscriber(Subscription sub) throws JMSException;
abstract public void notifyMessageAvailable();
1.1.2.1 +7 -3 jbossmq/src/main/org/jbossmq/server/JMSQueue.java
Index: JMSQueue.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jbossmq/server/JMSQueue.java,v
retrieving revision 1.1
retrieving revision 1.1.2.1
diff -u -r1.1 -r1.1.2.1
--- JMSQueue.java 2001/07/28 00:33:38 1.1
+++ JMSQueue.java 2001/08/01 05:19:48 1.1.2.1
@@ -111,9 +111,13 @@
ExclusiveQueue exclusiveQueue;
// Package protected ---------------------------------------------
- JMSDestination addConsumer(Subscription sub,ClientConsumer c) throws
JMSException {
- cat.debug("Adding consumer: "+c+")");
+ void addConsumer(Subscription sub,ClientConsumer c) throws JMSException {
+ cat.debug("Adding consumer: "+c);
exclusiveQueue.addConsumer(c);
+ }
+
+ JMSDestination addSubscriber(Subscription sub) throws JMSException {
+ cat.debug("Adding subscriber: "+sub);
return this;
}
1.15.2.1 +17 -10 jbossmq/src/main/org/jbossmq/server/JMSServer.java
Index: JMSServer.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jbossmq/server/JMSServer.java,v
retrieving revision 1.15
retrieving revision 1.15.2.1
diff -u -r1.15 -r1.15.2.1
--- JMSServer.java 2001/07/28 00:30:16 1.15
+++ JMSServer.java 2001/08/01 05:19:48 1.15.2.1
@@ -46,8 +46,6 @@
*/
public class JMSServer
{
-
-
/////////////////////////////////////////////////////////////////////
// Attributes
/////////////////////////////////////////////////////////////////////
@@ -217,13 +215,13 @@
//A connection has sent a new message
public void addMessage(ConnectionToken dc, SpyMessage val, Long txId) throws
JMSException
{
-
cat.debug("INCOMING: (TX="+txId+")"+dc.getClientID()+" =>
"+val.getJMSDestination());
JMSDestination
queue=(JMSDestination)destinations.get(val.getJMSDestination());
if (queue==null) throw new JMSException("This destination does not
exist !");
+
//Add the message to the queue
+ val.setReadOnlyMode();
queue.addMessage(val, txId);
-
}
/**
@@ -376,11 +374,14 @@
//A connection object wants to subscribe to a Destination
public void subscribe(ConnectionToken dc, Subscription sub) throws JMSException
{
- cat.debug("Server:
subscribe(dest="+sub.destination+",idConnection="+dc.getClientID()+")");
-
- ClientConsumer ClientConsumer = getClientConsumer(dc);
-
- ClientConsumer.addSubscription(sub);
+ try {
+ cat.debug("Server:
subscribe(dest="+sub.destination+",idConnection="+dc.getClientID()+")");
+ ClientConsumer ClientConsumer = getClientConsumer(dc);
+ ClientConsumer.addSubscription(sub);
+ } catch ( JMSException e ) {
+ cat.debug("Exception:", e);
+ throw e;
+ }
}
@@ -515,11 +516,17 @@
stateManager = newStateManager;
}
+ public static final String JBOSS_VESION = "JBossMQ ver. 0.9b";
+
public void restoreMessage(SpyMessage message)
{
JMSDestination
queue=(JMSDestination)destinations.get(message.getJMSDestination());
if (queue==null) throw new RuntimeException("This destination does not
exist!");
//Add the message to the queue
queue.restoreMessage(message);
+ }
+
+ public String toString() {
+ return JBOSS_VESION;
}
}
1.1.2.1 +15 -5 jbossmq/src/main/org/jbossmq/server/JMSTopic.java
Index: JMSTopic.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jbossmq/server/JMSTopic.java,v
retrieving revision 1.1
retrieving revision 1.1.2.1
diff -u -r1.1 -r1.1.2.1
--- JMSTopic.java 2001/07/28 00:33:38 1.1
+++ JMSTopic.java 2001/08/01 05:19:48 1.1.2.1
@@ -112,8 +112,20 @@
// Package protected ---------------------------------------------
- JMSDestination addConsumer(Subscription sub,ClientConsumer c) throws
JMSException {
+ void addConsumer(Subscription sub,ClientConsumer c) throws JMSException {
+ cat.debug("Adding consumer: "+c);
+ SpyTopic topic = (SpyTopic)sub.destination;
+ DurableSubcriptionID id = topic.getDurableSubscriptionID();
+ if( id!=null ) {
+ JMSQueue queue = getDurableSubscription(id);
+ queue.addConsumer(sub,c);
+ }
+ else {
+ sharedQueue.addConsumer(c);
+ }
+ }
+ JMSDestination addSubscriber(Subscription sub) throws JMSException {
SpyTopic topic = (SpyTopic)sub.destination;
DurableSubcriptionID id = topic.getDurableSubscriptionID();
if( id!=null ) {
@@ -129,11 +141,9 @@
}
}
- return queue.addConsumer(sub, c);
+ return queue.addSubscriber(sub);
} else {
- cat.debug("Adding consumer: "+c);
- sharedQueue.addConsumer(c);
return this;
}
}
1.4.2.1 +93 -84 jbossmq/src/main/org/jbossmq/server/StateManager.java
Index: StateManager.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jbossmq/server/StateManager.java,v
retrieving revision 1.4
retrieving revision 1.4.2.1
diff -u -r1.4 -r1.4.2.1
--- StateManager.java 2001/07/28 00:30:16 1.4
+++ StateManager.java 2001/08/01 05:19:48 1.4.2.1
@@ -200,9 +200,14 @@
category.debug("Restarting Durable Subscription:
"+clientId+","+name+","+topicName);
SpyTopic topic=new SpyTopic(topicName);
-
JMSTopic dest =
(JMSTopic)server.getJMSDestination(topic);
- dest.createDurableSubscription(new
DurableSubcriptionID(clientId, name));
+ if( dest == null ) {
+ category.warn("Subscription topic of not
found: "+topicName);
+ category.warn("Subscription cannot be
initialized: "+clientId+","+name);
+ element.removeFromParent();
+ } else {
+ dest.createDurableSubscription(new
DurableSubcriptionID(clientId, name));
+ }
} catch (JMSException e ) {
category.error("Could not initialize a durable
subscription for : Client Id="+clientId+", Name="+name+", Topic Name="+topicName,e);
@@ -238,88 +243,92 @@
stateFile = newStateFile;
}
- public void setDurableSubscription(JMSServer server, DurableSubcriptionID sub,
SpyTopic topic) throws JMSException {
- category.debug("Checking durable subscription: "+sub+", on topic:
"+topic);
- try {
- //Set the known Ids
- Enumeration enum = stateConfig.getElementsNamed("User");
- while( enum.hasMoreElements() ) {
-
- // Match the User.Name
- XElement user = (XElement)enum.nextElement();
- if( !user.containsField("Id") ||
!user.getField("Id").equals(sub.getClientID()) )
- continue;
-
- category.debug("Found a matching ClientID
configuration section.");
-
-
- XElement subscription=null;
-
- // Match the User/DurableSubscription.Name
- Enumeration enum2 =
user.getElementsNamed("DurableSubscription");
- while( enum2.hasMoreElements() ) {
- XElement t = (XElement)enum2.nextElement();
- if(
t.getField("Name").equals(sub.getSubscriptionName())) {
- subscription = t;
- break;
- }
- }
+ public String displayStateConfig() throws Exception {
+ return stateConfig.toString();
+ }
- if( subscription == null ) {
- category.debug("The subscription was not
previously registered.");
- // it was not previously registered...
- if( topic == null )
- return;
-
- subscription = new
XElement("DurableSubscription");
- subscription.addField("Name",
sub.getSubscriptionName());
- subscription.addField("TopicName",
topic.getName());
- user.addElement(subscription);
-
- JMSTopic dest =
(JMSTopic)server.getJMSDestination(topic);
- dest.createDurableSubscription(sub);
-
- saveConfig();
-
- } else {
- category.debug("The subscription was
previously registered.");
- // it was previously registered...
- if(
subscription.getField("TopicName").equals(topic.getName()) ) {
- // and it is the same as before, do
nothing.
- return;
- } else {
- category.debug("But the topic was
different, changing the subscription.");
- // we have to change the
subscription...
- SpyTopic prevTopic = new SpyTopic(
subscription.getField("TopicName") );
- JMSTopic dest =
(JMSTopic)server.getJMSDestination(prevTopic);
- dest.destoryDurableSubscription(sub);
-
- if( topic == null ) {
-
subscription.removeFromParent();
- } else {
-
subscription.setField("TopicName", topic.getName());
- dest =
(JMSTopic)server.getJMSDestination(topic);
-
dest.createDurableSubscription(sub);
- }
-
- saveConfig();
- }
- // Subscription existed
- return;
- }
+ public void setDurableSubscription(JMSServer server, DurableSubcriptionID sub,
SpyTopic topic) throws JMSException {
+ category.debug("Checking durable subscription: " + sub + ", on topic: " +
topic);
+ try {
+ //Set the known Ids
+ Enumeration enum= stateConfig.getElementsNamed("User");
+ while (enum.hasMoreElements()) {
+
+ // Match the User.Name
+ XElement user= (XElement) enum.nextElement();
+ if (!user.containsField("Id") ||
!user.getField("Id").equals(sub.getClientID()))
+ continue;
+
+ category.debug("Found a matching ClientID configuration section.");
+
+ XElement subscription= null;
+
+ // Match the User/DurableSubscription.Name
+ Enumeration enum2= user.getElementsNamed("DurableSubscription");
+ while (enum2.hasMoreElements()) {
+ XElement t= (XElement) enum2.nextElement();
+ if (t.getField("Name").equals(sub.getSubscriptionName())) {
+ subscription= t;
+ break;
+ }
+ }
+
+ if (subscription == null) {
+ category.debug("The subscription was not previously
registered.");
+ // it was not previously registered...
+ if (topic == null)
+ return;
+
+ subscription= new XElement("DurableSubscription");
+ subscription.addField("Name", sub.getSubscriptionName());
+ subscription.addField("TopicName", topic.getName());
+ user.addElement(subscription);
+
+ JMSTopic dest= (JMSTopic) server.getJMSDestination(topic);
+ dest.createDurableSubscription(sub);
+
+ saveConfig();
+
+ } else {
+ category.debug("The subscription was previously registered.");
+ // it was previously registered...
+ if
(subscription.getField("TopicName").equals(topic.getName())) {
+ // and it is the same as before, do nothing.
+ return;
+ } else {
+ category.debug("But the topic was different, changing the
subscription.");
+ // we have to change the subscription...
+ SpyTopic prevTopic= new
SpyTopic(subscription.getField("TopicName"));
+ JMSTopic dest= (JMSTopic)
server.getJMSDestination(prevTopic);
+ dest.destoryDurableSubscription(sub);
+
+ if (topic == null) {
+ subscription.removeFromParent();
+ } else {
+ subscription.setField("TopicName", topic.getName());
+ dest= (JMSTopic) server.getJMSDestination(topic);
+ dest.createDurableSubscription(sub);
+ }
+
+ saveConfig();
+ }
+ // Subscription existed
+ }
+ return;
+ }
+
+ // Could not find that user..
+ throw new JMSException("ClientID '" + sub.getClientID() + "'
cannot create durable subscriptions.");
+
+ } catch (java.io.IOException e) {
+ JMSException newE= new JMSException("Could not setup the durable
subscription");
+ newE.setLinkedException(e);
+ throw newE;
+ } catch (org.jbossmq.xml.XElementException e) {
+ JMSException newE= new JMSException("Could not setup the durable
subscription");
+ newE.setLinkedException(e);
+ throw newE;
+ }
- // Could not find that user..
- throw new JMSException("ClientID
'"+sub.getClientID()+"' cannot create durable subscriptions.");
- }
- } catch ( java.io.IOException e ) {
- JMSException newE = new JMSException("Could not setup the
durable subscription");
- newE.setLinkedException(e);
- throw newE;
- } catch ( org.jbossmq.xml.XElementException e ) {
- JMSException newE = new JMSException("Could not setup the
durable subscription");
- newE.setLinkedException(e);
- throw newE;
- }
-
}
}
1.3.2.1 +3 -3 jbossmq/src/main/org/jbossmq/server/StateManagerMBean.java
Index: StateManagerMBean.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jbossmq/server/StateManagerMBean.java,v
retrieving revision 1.3
retrieving revision 1.3.2.1
diff -u -r1.3 -r1.3.2.1
--- StateManagerMBean.java 2001/07/28 00:30:16 1.3
+++ StateManagerMBean.java 2001/08/01 05:19:48 1.3.2.1
@@ -60,8 +60,6 @@
public interface StateManagerMBean
extends org.jboss.util.ServiceMBean
{
-
-
// Public --------------------------------------------------------
@@ -75,4 +73,6 @@
public java.lang.String getStateFile();
public void setStateFile(java.lang.String newStateFile);
+
+ public String displayStateConfig() throws Exception;
}
_______________________________________________
Jboss-development mailing list
[EMAIL PROTECTED]
http://lists.sourceforge.net/lists/listinfo/jboss-development
