User: starksm
Date: 01/11/27 22:15:32
Modified: src/main/org/jboss/mq/server Tag: Branch_2_4 BasicQueue.java
ClientConsumer.java JMSDestination.java
JMSServer.java
Log:
Complete switch to org.jboss.logging.Logger and use trace to reduce default
logging
Add support for package version manifest headers
Revision Changes Path
No revision
No revision
1.4.2.3 +207 -128 jbossmq/src/main/org/jboss/mq/server/BasicQueue.java
Index: BasicQueue.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/BasicQueue.java,v
retrieving revision 1.4.2.2
retrieving revision 1.4.2.3
diff -u -r1.4.2.2 -r1.4.2.3
--- BasicQueue.java 2001/10/22 20:38:51 1.4.2.2
+++ BasicQueue.java 2001/11/28 06:15:32 1.4.2.3
@@ -18,6 +18,7 @@
import javax.jms.Destination;
import javax.jms.JMSException;
+import org.jboss.logging.Logger;
import org.jboss.mq.*;
import org.jboss.mq.pm.PersistenceManager;
import org.jboss.mq.selectors.Selector;
@@ -30,10 +31,12 @@
* @author Norbert Lataille ([EMAIL PROTECTED])
* @author David Maplesden ([EMAIL PROTECTED])
* @created August 16, 2001
- * @version $Revision: 1.4.2.2 $
+ * @version $Revision: 1.4.2.3 $
*/
//abstract public class BasicQueue implements Runnable {
-public class BasicQueue {
+public class BasicQueue
+{
+ static Logger log = Logger.getLogger( BasicQueue.class );
//List of messages waiting to be dispatched
SortedSet messages = new TreeSet();
//The JMSServer object
@@ -43,123 +46,157 @@
//List of messages that should be acked or else returned to thier
//owning exclusive queues.
HashMap unacknowledgedMessages = new HashMap();
-
+
HashMap removedSubscribers = new HashMap();
-
- static org.apache.log4j.Category cat = org.apache.log4j.Category.getInstance(
BasicQueue.class );
-
+
// Constructor ---------------------------------------------------
public BasicQueue( JMSServer server )
- throws JMSException {
+ throws JMSException
+ {
this.server = server;
}
-
- public void clientConsumerStopped( ClientConsumer clientConsumer ) {
+
+ public void clientConsumerStopped( ClientConsumer clientConsumer )
+ {
//remove all waiting subs for this clientConsumer and send to its blocked
list.
- synchronized ( receivers ) {
- for ( Iterator it = receivers.iterator(); it.hasNext(); ) {
+ synchronized ( receivers )
+ {
+ for ( Iterator it = receivers.iterator(); it.hasNext(); )
+ {
Subscription sub = ( Subscription )it.next();
- if ( sub.clientConsumer.equals( clientConsumer ) ) {
+ if ( sub.clientConsumer.equals( clientConsumer ) )
+ {
clientConsumer.addBlockedSubscription( sub );
it.remove();
}
}
}
}
-
+
//Used to put a message that was added previously to the queue, back in the queue
- public void restoreMessage( SpyMessage mes ) {
+ public void restoreMessage( SpyMessage mes )
+ {
internalAddMessage( mes );
}
-
-
+
+
public SpyMessage[] browse( String selector )
- throws JMSException {
-
- if ( selector == null ) {
+ throws JMSException
+ {
+
+ if ( selector == null )
+ {
SpyMessage list[];
- synchronized ( messages ) {
+ synchronized ( messages )
+ {
list = new SpyMessage[messages.size()];
list = ( SpyMessage[] )messages.toArray( list );
}
return list;
- } else {
+ }
+ else
+ {
Selector s = new Selector( selector );
LinkedList selection = new LinkedList();
-
- synchronized ( messages ) {
+
+ synchronized ( messages )
+ {
Iterator i = messages.iterator();
- while ( i.hasNext() ) {
+ while ( i.hasNext() )
+ {
SpyMessage m = ( SpyMessage )i.next();
- if ( s.test( m ) ) {
+ if ( s.test( m ) )
+ {
selection.add( m );
}
}
}
-
+
SpyMessage list[];
list = new SpyMessage[selection.size()];
list = ( SpyMessage[] )selection.toArray( list );
return list;
}
}
-
- public void addReceiver( Subscription sub ) {
- synchronized ( messages ) {
- if ( messages.size() != 0 ) {
- for ( Iterator it = messages.iterator(); it.hasNext(); ) {
+
+ public void addReceiver( Subscription sub )
+ {
+ synchronized ( messages )
+ {
+ if ( messages.size() != 0 )
+ {
+ for ( Iterator it = messages.iterator(); it.hasNext(); )
+ {
SpyMessage message = ( SpyMessage )it.next();
- try {
- if ( sub.accepts( message ) ) {
+ try
+ {
+ if ( sub.accepts( message ) )
+ {
//queue message for sending to this sub
queueMessageForSending( sub, message );
it.remove();
return;
}
- } catch ( JMSException ignore ) {
- cat.debug( "Caught unusual exception in addToReceivers.", ignore
);
+ } catch ( JMSException ignore )
+ {
+ log.debug( "Caught unusual exception in addToReceivers.", ignore
);
}
}
}
}
addToReceivers( sub );
}
-
- public void removeReceiver( Subscription sub ) {
- synchronized ( receivers ) {
+
+ public void removeReceiver( Subscription sub )
+ {
+ synchronized ( receivers )
+ {
receivers.remove( sub );
}
}
-
- public void removeSubscriber( Subscription sub ) {
+
+ public void removeSubscriber( Subscription sub )
+ {
removeReceiver( sub );
- if ( hasUnackedMessages( sub.subscriptionId ) ) {
- synchronized ( removedSubscribers ) {
+ if ( hasUnackedMessages( sub.subscriptionId ) )
+ {
+ synchronized ( removedSubscribers )
+ {
removedSubscribers.put( new Integer( sub.subscriptionId ), sub );
}
- } else {
+ } else
+ {
( ( ClientConsumer )sub.clientConsumer ).removeRemovedSubscription(
sub.subscriptionId );
}
}
-
+
public SpyMessage receive( Subscription sub, boolean wait )
- throws JMSException {
+ throws JMSException
+ {
SpyMessage message = null;
Selector selector = sub.getSelector();
- synchronized ( receivers ) {
- if ( selector == null ) {
- synchronized ( messages ) {
- if ( messages.size() != 0 ) {
+ synchronized ( receivers )
+ {
+ if ( selector == null )
+ {
+ synchronized ( messages )
+ {
+ if ( messages.size() != 0 )
+ {
message = ( SpyMessage )messages.first();
messages.remove( message );
}
}
- } else {
- synchronized ( messages ) {
+ } else
+ {
+ synchronized ( messages )
+ {
Iterator i = messages.iterator();
- while ( i.hasNext() ) {
+ while ( i.hasNext() )
+ {
SpyMessage m = ( SpyMessage )i.next();
- if ( selector.test( m ) ) {
+ if ( selector.test( m ) )
+ {
message = m;
messages.remove( message );
break;
@@ -168,175 +205,217 @@
}
}
- if ( message == null ) {
- if ( wait ) {
+ if ( message == null )
+ {
+ if ( wait )
+ {
addToReceivers( sub );
}
- } else {
+ } else
+ {
setupMessageAcknowledgement( sub, message );
}
}
-
+
return message;
}
-
+
public void destroy()
- throws JMSException {
- synchronized ( unacknowledgedMessages ) {
+ throws JMSException
+ {
+ synchronized ( unacknowledgedMessages )
+ {
Iterator i = ( ( HashMap )unacknowledgedMessages.clone()
).keySet().iterator();
- while ( i.hasNext() ) {
+ while ( i.hasNext() )
+ {
AcknowledgementRequest item = ( AcknowledgementRequest )i.next();
- try {
+ try
+ {
acknowledge( item, null );
- } catch ( JMSException ignore ) {
+ } catch ( JMSException ignore )
+ {
}
}
}
}
-
+
public void acknowledge( AcknowledgementRequest item, org.jboss.mq.pm.Tx txId )
- throws javax.jms.JMSException {
-
+ throws javax.jms.JMSException
+ {
+
// This task gets run to place the neg ack a messge (place it back on the
queue)
- class RestoreMessageTask implements Runnable {
-
+ class RestoreMessageTask implements Runnable
+ {
+
SpyMessage message;
int subscriptionId;
-
- RestoreMessageTask( SpyMessage m, int subscriptionId ) {
+
+ RestoreMessageTask( SpyMessage m, int subscriptionId )
+ {
message = m;
this.subscriptionId = subscriptionId;
}
-
- public void run() {
- cat.debug( "Restoring message: " + message.getJMSMessageID() );
+
+ public void run()
+ {
+ if( log.isTraceEnabled() )
+ log.trace( "Restoring message: " + message.getJMSMessageID() );
restoreMessage( message );
}
}
-
+
SpyMessage m;
- synchronized ( unacknowledgedMessages ) {
+ synchronized ( unacknowledgedMessages )
+ {
m = ( SpyMessage )unacknowledgedMessages.remove( item );
}
-
+
checkRemovedSubscribers( item.subscriberId );
-
- if ( m == null ) {
+
+ if ( m == null )
+ {
return;
}
-
+
// Was it a negative acknowledge??
- if ( !item.isAck ) {
-
+ if ( !item.isAck )
+ {
+
Runnable task = new RestoreMessageTask( m, item.subscriberId );
server.getPersistenceManager().getTxManager().addPostCommitTask( txId,
task );
-
- } else {
-
- if ( this instanceof PersistentQueue && m.getJMSDeliveryMode() ==
DeliveryMode.PERSISTENT ) {
+
+ } else
+ {
+
+ if ( this instanceof PersistentQueue && m.getJMSDeliveryMode() ==
DeliveryMode.PERSISTENT )
+ {
server.getPersistenceManager().remove( m, txId );
}
-
+
Runnable task = new RestoreMessageTask( m, item.subscriberId );
server.getPersistenceManager().getTxManager().addPostRollbackTask( txId,
task );
-
+
}
-
+
}
-
+
public void addMessage( SpyMessage mes, org.jboss.mq.pm.Tx txId )
- throws JMSException {
+ throws JMSException
+ {
// This task gets run to make the message visible in the queue.
- class AddMessagePostCommitTask implements Runnable {
-
+ class AddMessagePostCommitTask implements Runnable
+ {
+
SpyMessage message;
-
- AddMessagePostCommitTask( SpyMessage m ) {
+
+ AddMessagePostCommitTask( SpyMessage m )
+ {
message = m;
}
-
- public void run() {
+
+ public void run()
+ {
//restore a message to the message list...
internalAddMessage( message );
}
}
-
+
// The message gets added to the queue after the transaction commits
Runnable task = new AddMessagePostCommitTask( mes );
server.getPersistenceManager().getTxManager().addPostCommitTask( txId, task );
-
+
}
-
- protected void setupMessageAcknowledgement( Subscription sub, SpyMessage message
) {
+
+ protected void setupMessageAcknowledgement( Subscription sub, SpyMessage message
)
+ {
AcknowledgementRequest ack = new AcknowledgementRequest();
ack.destination = message.getJMSDestination();
ack.messageID = message.getJMSMessageID();
ack.subscriberId = sub.subscriptionId;
ack.isAck = false;
-
- synchronized ( unacknowledgedMessages ) {
+
+ synchronized ( unacknowledgedMessages )
+ {
unacknowledgedMessages.put( ack, message );
}
}
-
+
protected void queueMessageForSending( Subscription sub, SpyMessage message )
- throws JMSException {
+ throws JMSException
+ {
setupMessageAcknowledgement( sub, message );
ReceiveRequest r = new ReceiveRequest();
r.message = message;
r.subscriptionId = new Integer( sub.subscriptionId );
-
+
( ( ClientConsumer )sub.clientConsumer ).queueMessageForSending( r );
}
-
- protected void addToReceivers( Subscription sub ) {
- synchronized ( receivers ) {
+
+ protected void addToReceivers( Subscription sub )
+ {
+ synchronized ( receivers )
+ {
receivers.add( sub );
}
}
-
- protected boolean hasUnackedMessages( int sub ) {
- synchronized ( unacknowledgedMessages ) {
- for ( Iterator it = unacknowledgedMessages.keySet().iterator();
it.hasNext(); ) {
- if ( ( ( AcknowledgementRequest )it.next() ).subscriberId == sub ) {
+
+ protected boolean hasUnackedMessages( int sub )
+ {
+ synchronized ( unacknowledgedMessages )
+ {
+ for ( Iterator it = unacknowledgedMessages.keySet().iterator();
it.hasNext(); )
+ {
+ if ( ( ( AcknowledgementRequest )it.next() ).subscriberId == sub )
+ {
return true;
}
}
return false;
}
}
-
- protected void checkRemovedSubscribers( int subId ) {
+
+ protected void checkRemovedSubscribers( int subId )
+ {
Integer id = new Integer( subId );
- synchronized ( removedSubscribers ) {
- if ( removedSubscribers.containsKey( id ) && !hasUnackedMessages( subId )
) {
+ synchronized ( removedSubscribers )
+ {
+ if ( removedSubscribers.containsKey( id ) && !hasUnackedMessages( subId ) )
+ {
Subscription sub = ( Subscription )removedSubscribers.remove( id );
( ( ClientConsumer )sub.clientConsumer ).removeRemovedSubscription(
subId );
}
}
}
-
- private void internalAddMessage( SpyMessage message ) {
+
+ private void internalAddMessage( SpyMessage message )
+ {
//try waiting receivers
- synchronized ( receivers ) {
- if ( !receivers.isEmpty() ) {
- for ( Iterator it = receivers.iterator(); it.hasNext(); ) {
+ synchronized ( receivers )
+ {
+ if ( !receivers.isEmpty() )
+ {
+ for ( Iterator it = receivers.iterator(); it.hasNext(); )
+ {
Subscription sub = ( Subscription )it.next();
- try {
- if ( sub.accepts( message ) ) {
+ try
+ {
+ if ( sub.accepts( message ) )
+ {
//queue message for sending to this sub
queueMessageForSending( sub, message );
it.remove();
return;
}
- } catch ( JMSException ignore ) {
- cat.debug( "Caught unusual exception in internalAddMessage.",
ignore );
+ } catch ( JMSException ignore )
+ {
+ log.debug( "Caught unusual exception in internalAddMessage.",
ignore );
}
}
}
-
+
//else add to message list
- synchronized ( messages ) {
+ synchronized ( messages )
+ {
messages.add( message );
}
}
1.4.2.2 +156 -103 jbossmq/src/main/org/jboss/mq/server/ClientConsumer.java
Index: ClientConsumer.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/ClientConsumer.java,v
retrieving revision 1.4.2.1
retrieving revision 1.4.2.2
diff -u -r1.4.2.1 -r1.4.2.2
--- ClientConsumer.java 2001/08/23 03:57:12 1.4.2.1
+++ ClientConsumer.java 2001/11/28 06:15:32 1.4.2.2
@@ -15,8 +15,9 @@
import javax.jms.Destination;
import javax.jms.JMSException;
-import org.jboss.mq.*;
+import org.jboss.logging.Logger;
+import org.jboss.mq.*;
import org.jboss.mq.xml.XElement;
/**
@@ -25,9 +26,11 @@
*
* @author Hiram Chirino ([EMAIL PROTECTED])
* @created August 16, 2001
- * @version $Revision: 1.4.2.1 $
+ * @version $Revision: 1.4.2.2 $
*/
-public class ClientConsumer implements Runnable {
+public class ClientConsumer implements Runnable
+{
+ Logger log;
//The JMSServer object
JMSServer server;
//The connection this queue will send messages over
@@ -40,40 +43,45 @@
HashMap subscriptions = new HashMap();
//Maps a subscription id to a Subscription for subscriptions that have finished
receiving
HashMap removedSubscriptions = new HashMap();
-
+
LinkedList blockedSubscriptions = new LinkedList();
-
- org.apache.log4j.Category cat;
-
+
//List of messages waiting to be transmitted to the client
private LinkedList messages = new LinkedList();
//LinkedList of the the temporary destinations that this client created
-// public LinkedList temporaryDestinations = new LinkedList();
+ // public LinkedList temporaryDestinations = new LinkedList();
//The message push thread for this consumer connection
private Thread messagePushThread;
-
+
// Constructor ---------------------------------------------------
public ClientConsumer( JMSServer server, ConnectionToken dc )
- throws JMSException {
+ throws JMSException
+ {
this.server = server;
this.dc = dc;
- cat = org.apache.log4j.Category.getInstance( ClientConsumer.class.getName() +
":" + dc.getClientID() );
+ log = Logger.getLogger(ClientConsumer.class.getName() + ":" +
dc.getClientID() );
messagePushThread = new Thread( server.threadGroup, this, "Message Pusher " +
dc.getClientID() );
messagePushThread.setDaemon( true );
messagePushThread.start();
}
-
+
public void setEnabled( boolean enabled )
- throws JMSException {
- cat.debug( "" + this + "->setEnabled(enabled=" + enabled + ")" );
+ throws JMSException
+ {
+ if( log.isTraceEnabled() )
+ log.trace( "" + this + "->setEnabled(enabled=" + enabled + ")" );
this.enabled = enabled;
- if ( enabled ) {
+ if ( enabled )
+ {
// queues might be waiting for messages.
- synchronized ( blockedSubscriptions ) {
- for ( Iterator it = blockedSubscriptions.iterator(); it.hasNext(); ) {
+ synchronized ( blockedSubscriptions )
+ {
+ for ( Iterator it = blockedSubscriptions.iterator(); it.hasNext(); )
+ {
Subscription sub = ( Subscription )it.next();
JMSDestination dest = server.getJMSDestination( sub.destination );
- if ( dest != null ) {
+ if ( dest != null )
+ {
dest.addReceiver( sub );
}
}
@@ -81,182 +89,227 @@
}
}
}
-
- public void queueMessageForSending( ReceiveRequest r ) {
- synchronized ( messages ) {
+
+ public void queueMessageForSending( ReceiveRequest r )
+ {
+ synchronized ( messages )
+ {
messages.add( r );
messages.notify();
}
}
-
+
public void addSubscription( Subscription req )
- throws JMSException {
- cat.debug( "Adding subscription for: " + req );
+ throws JMSException
+ {
+ if( log.isTraceEnabled() )
+ log.trace( "Adding subscription for: " + req );
req.dc = dc;
req.clientConsumer = this;
-
+
JMSDestination jmsdest = ( JMSDestination )server.getJMSDestination(
req.destination );
- if ( jmsdest == null ) {
+ if ( jmsdest == null )
+ {
throw new JMSException( "This destination does not exist !" );
}
-
+
jmsdest.addSubscriber( req );
-
- synchronized ( subscriptions ) {
+
+ synchronized ( subscriptions )
+ {
HashMap subscriptionsClone = ( HashMap )subscriptions.clone();
subscriptionsClone.put( new Integer( req.subscriptionId ), req );
subscriptions = subscriptionsClone;
}
}
-
- public void close() {
-
- cat.debug( "" + this + "->close()" );
-
- synchronized ( messages ) {
+
+ public void close()
+ {
+ if( log.isTraceEnabled() )
+ log.trace( "" + this + "->close()" );
+
+ synchronized ( messages )
+ {
closed = true;
messages.notifyAll();
}
-
- synchronized ( subscriptions ) {
+
+ synchronized ( subscriptions )
+ {
Iterator i = subscriptions.keySet().iterator();
- while ( i.hasNext() ) {
+ while ( i.hasNext() )
+ {
Integer subscriptionId = ( Integer )i.next();
- try {
+ try
+ {
removeSubscription( subscriptionId.intValue() );
- } catch ( JMSException ignore ) {
+ } catch ( JMSException ignore )
+ {
}
}
}
-
+
}
-
+
public SpyMessage receive( int subscriberId, long wait )
- throws JMSException {
-
+ throws JMSException
+ {
+
Subscription req = ( Subscription )subscriptions.get( new Integer(
subscriberId ) );
- if ( req == null ) {
+ if ( req == null )
+ {
throw new JMSException( "The provided subscription does not exist" );
}
-
+
JMSDestination queue = server.getJMSDestination( req.destination );
- if ( queue == null ) {
+ if ( queue == null )
+ {
throw new JMSException( "The subscription's destination does not exist" );
}
-
- if ( enabled ) {
+
+ if ( enabled )
+ {
return queue.receive( req, ( wait != -1 ) );
- } else if ( wait != -1 ) {
+ } else if ( wait != -1 )
+ {
addBlockedSubscription( req );
}
-
+
return null;
}
-
-
+
+
public void removeSubscription( int subscriptionId )
- throws JMSException {
- cat.debug( "" + this + "->removeSubscription(subscriberId=" + subscriptionId
+ ")" );
-
+ throws JMSException
+ {
+ if( log.isTraceEnabled() )
+ log.trace( "" + this + "->removeSubscription(subscriberId=" +
subscriptionId + ")" );
+
Integer subId = new Integer( subscriptionId );
Subscription req;
- synchronized ( subscriptions ) {
+ synchronized ( subscriptions )
+ {
HashMap subscriptionsClone = ( HashMap )subscriptions.clone();
req = ( Subscription )subscriptionsClone.remove( subId );
subscriptions = subscriptionsClone;
- if ( req != null ) {
+ if ( req != null )
+ {
removedSubscriptions.put( subId, req );
}
}
-
- if ( req == null ) {
+
+ if ( req == null )
+ {
throw new JMSException( "The subscription had not been previously
registered" );
}
-
+
JMSDestination queue = ( JMSDestination )server.getJMSDestination(
req.destination );
- if ( queue == null ) {
+ if ( queue == null )
+ {
throw new JMSException( "The subscription was registed with a destination
that does not exist !" );
}
-
+
queue.removeSubscriber( req );
-
+
}
-
-
+
+
// Iterate over the consumers asking them to take messages until they stop
// consuming.
- public void run() {
-
- cat.debug( "" + this + "->run()" );
-
- while ( true ) {
-
+ public void run()
+ {
+ if( log.isTraceEnabled() )
+ log.trace( "" + this + "->run()" );
+ while ( true )
+ {
+
ReceiveRequest[] job;
- synchronized ( messages ) {
- while ( messages.size() == 0 ) {
- try {
+ synchronized ( messages )
+ {
+ while ( messages.size() == 0 )
+ {
+ try
+ {
messages.wait();
- } catch ( InterruptedException e ) {
+ }
+ catch ( InterruptedException e )
+ {
}
- if ( closed ) {
+ if ( closed )
+ {
return;
}
}
-
+
job = new ReceiveRequest[messages.size()];
job = ( ReceiveRequest[] )messages.toArray( job );
messages.clear();
}
-
- try {
+
+ try
+ {
dc.clientIL.receive( job );
- } catch ( Exception e ) {
- cat.warn( "Could not send messages to a receiver.", e );
- try {
+ }
+ catch ( Exception e )
+ {
+ log.warn( "Could not send messages to a receiver.", e );
+ try
+ {
server.connectionFailure( dc );
- } catch ( Throwable ignore ) {
- cat.warn( "Could not close the client connection..", ignore );
+ } catch ( Throwable ignore )
+ {
+ log.warn( "Could not close the client connection..", ignore );
}
}
}
-
+
}
- public String toString() {
+ public String toString()
+ {
return "ClientConsumer:" + dc.getClientID();
}
-
+
public void acknowledge( AcknowledgementRequest request, org.jboss.mq.pm.Tx txId
)
- throws JMSException {
+ throws JMSException
+ {
Subscription sub = ( Subscription )subscriptions.get( new Integer(
request.subscriberId ) );
-
- if ( sub == null ) {
+
+ if ( sub == null )
+ {
//might be in removed subscriptions
- synchronized ( subscriptions ) {
+ synchronized ( subscriptions )
+ {
sub = ( Subscription )removedSubscriptions.get( new Integer(
request.subscriberId ) );
}
}
-
- if ( sub == null ) {
+
+ if ( sub == null )
+ {
throw new JMSException( "The provided subscription does not exist" );
}
-
+
JMSDestination queue = server.getJMSDestination( sub.destination );
- if ( queue == null ) {
+ if ( queue == null )
+ {
throw new JMSException( "The subscription's destination does not exist" );
}
-
+
queue.acknowledge( request, sub, txId );
}
-
- void addBlockedSubscription( Subscription sub ) {
- synchronized ( blockedSubscriptions ) {
+
+ void addBlockedSubscription( Subscription sub )
+ {
+ synchronized ( blockedSubscriptions )
+ {
blockedSubscriptions.add( sub );
}
}
-
- void removeRemovedSubscription( int subId ) {
- synchronized ( subscriptions ) {
+
+ void removeRemovedSubscription( int subId )
+ {
+ synchronized ( subscriptions )
+ {
removedSubscriptions.remove( new Integer( subId ) );
}
}
1.3.2.2 +20 -19 jbossmq/src/main/org/jboss/mq/server/JMSDestination.java
Index: JMSDestination.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/JMSDestination.java,v
retrieving revision 1.3.2.1
retrieving revision 1.3.2.2
diff -u -r1.3.2.1 -r1.3.2.2
--- JMSDestination.java 2001/08/23 03:57:12 1.3.2.1
+++ JMSDestination.java 2001/11/28 06:15:32 1.3.2.2
@@ -16,6 +16,7 @@
import javax.jms.Destination;
import javax.jms.JMSException;
+import org.jboss.logging.Logger;
import org.jboss.mq.*;
/**
@@ -26,49 +27,49 @@
* @author Hiram Chirino ([EMAIL PROTECTED])
* @author David Maplesden ([EMAIL PROTECTED])
* @created August 16, 2001
- * @version $Revision: 1.3.2.1 $
+ * @version $Revision: 1.3.2.2 $
*/
-public abstract class JMSDestination {
-
+public abstract class JMSDestination
+{
//the Destination of this queue
SpyDestination destination;
//If this is a temporaryDestination, temporaryDestination=ClientConsumer of the
owner, otherwise it's null
ClientConsumer temporaryDestination;
//The JMSServer object
JMSServer server;
-
+
//Counter used to number incomming messages. (Used to order the messages.)
long messageIdCounter = Long.MIN_VALUE;
-
- org.apache.log4j.Category cat;
-
+ Logger log;
+
// Constructor ---------------------------------------------------
JMSDestination( SpyDestination dest, ClientConsumer temporary, JMSServer server )
- throws JMSException {
- cat = org.apache.log4j.Category.getInstance( this.getClass().getName() + ":"
+ dest );
+ throws JMSException
+ {
+ log = Logger.getLogger( this.getClass().getName() + ":" + dest );
destination = dest;
temporaryDestination = temporary;
this.server = server;
}
-
-
+
+
public abstract void addSubscriber( Subscription sub )
throws JMSException;
-
+
public abstract void removeSubscriber( Subscription sub )
throws JMSException;
-
+
public abstract SpyMessage receive( Subscription sub, boolean wait )
throws JMSException;
-
+
public abstract void addReceiver( Subscription sub );
-
+
public abstract void removeReceiver( Subscription sub );
-
+
public abstract void restoreMessage( SpyMessage message );
-
+
public abstract void clientConsumerStopped( ClientConsumer clientConsumer );
-
+
/**
* @param req org.jboss.mq.AcknowledgementRequest
* @param sub org.jboss.mq.Subscription
@@ -77,7 +78,7 @@
*/
public abstract void acknowledge( org.jboss.mq.AcknowledgementRequest req,
org.jboss.mq.Subscription sub, org.jboss.mq.pm.Tx txId )
throws javax.jms.JMSException;
-
+
/**
* @param mes org.jboss.mq.SpyMessage
* @param txId org.jboss.mq.pm.Tx
1.4.2.2 +270 -189 jbossmq/src/main/org/jboss/mq/server/JMSServer.java
Index: JMSServer.java
===================================================================
RCS file: /cvsroot/jboss/jbossmq/src/main/org/jboss/mq/server/JMSServer.java,v
retrieving revision 1.4.2.1
retrieving revision 1.4.2.2
diff -u -r1.4.2.1 -r1.4.2.2
--- JMSServer.java 2001/08/23 03:57:12 1.4.2.1
+++ JMSServer.java 2001/11/28 06:15:32 1.4.2.2
@@ -24,9 +24,10 @@
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
+
+import org.jboss.logging.Logger;
import org.jboss.mq.*;
import org.jboss.mq.pm.*;
-
import org.jboss.mq.xml.XElement;
/**
@@ -36,22 +37,27 @@
* @author Hiram Chirino ([EMAIL PROTECTED])
* @author David Maplesden ([EMAIL PROTECTED])
* @created August 16, 2001
- * @version $Revision: 1.4.2.1 $
+ * @version $Revision: 1.4.2.2 $
*/
-public class JMSServer {
-
+public class JMSServer
+{
+ private static Logger log = Logger.getLogger(JMSServer.class);
+ //Implement Singleton design pattern for JVMIL to work correctly
+ //It would be nice if there was another way of doing this.
+ protected static JMSServer theInstance = new JMSServer();
+
/////////////////////////////////////////////////////////////////////
// Attributes
/////////////////////////////////////////////////////////////////////
-
+
//messages pending for a Destination ( HashMap of JMSServerQueue objects )
public HashMap destinations = new HashMap();
//Thread group for server side threads.
public ThreadGroup threadGroup = new ThreadGroup( "JBossMQ Server Threads" );
-
+
//The list of ClientConsumers hased by ConnectionTokens
HashMap clientConsumers = new HashMap();
-
+
//last id given to a client
private int lastID = 1;
//last id given to a temporary topic
@@ -62,58 +68,58 @@
private StateManager stateManager;
//The persistence manager
private PersistenceManager persistenceManager;
-
+
private Object stateLock = new Object();
-
+
/**
* <code>true</code> when the server is running. <code>false</code> when the
* server should stop running.
*/
private boolean alive = true;
-
+
/**
* Because there can be a delay between killing the JMS service and the
* service actually dying, this field is used to tell external classes that
* that server has actually stopped.
*/
private boolean stopped = true;
-
- //Implement Singleton design pattern for JVMIL to work correctly
- //It would be nice if there was another way of doing this.
- protected static JMSServer theInstance = new JMSServer();
-
- static org.apache.log4j.Category cat = org.apache.log4j.Category.getInstance(
JMSServer.class );
-
+
/////////////////////////////////////////////////////////////////////
// Constructors
/////////////////////////////////////////////////////////////////////
- protected JMSServer() {
+ protected JMSServer()
+ {
}
-
+
public void setEnabled( ConnectionToken dc, boolean enabled )
- throws JMSException {
+ throws JMSException
+ {
ClientConsumer ClientConsumer = getClientConsumer( dc );
ClientConsumer.setEnabled( enabled );
- if ( !enabled ) {
- for ( Iterator it = destinations.values().iterator(); it.hasNext(); ) {
+ if ( !enabled )
+ {
+ for ( Iterator it = destinations.values().iterator(); it.hasNext(); )
+ {
( ( JMSDestination )it.next() ).clientConsumerStopped( ClientConsumer );
}
}
}
-
-
- public void setStateManager( StateManager newStateManager ) {
+
+
+ public void setStateManager( StateManager newStateManager )
+ {
stateManager = newStateManager;
}
-
- public void setPersistenceManager( org.jboss.mq.pm.PersistenceManager
newPersistenceManager ) {
+
+ public void setPersistenceManager( org.jboss.mq.pm.PersistenceManager
newPersistenceManager )
+ {
persistenceManager = newPersistenceManager;
}
-
+
/////////////////////////////////////////////////////////////////////
// Public Methods
/////////////////////////////////////////////////////////////////////
-
+
/**
* Returns <code>false</code> if the JMS server is currently running and
* handling requests, <code>true</code> otherwise.
@@ -121,135 +127,159 @@
* @return <code>false</code> if the JMS server is currently running and
* handling requests, <code>true</code> otherwise.
*/
- public boolean isStopped() {
- synchronized ( stateLock ) {
+ public boolean isStopped()
+ {
+ synchronized ( stateLock )
+ {
return this.stopped;
}
}
-
+
//Get a new ClientID for a connection
- public String getID() {
+ public String getID()
+ {
String ID = null;
-
- while ( true ) {
- try {
- synchronized ( this ) {
+
+ while ( true )
+ {
+ try
+ {
+ synchronized ( this )
+ {
ID = "ID" + ( new Integer( lastID++ ).toString() );
}
stateManager.addLoggedOnClientId( ID );
break;
- } catch ( Exception e ) {
+ } catch ( Exception e )
+ {
}
}
-
+
return ID;
}
-
+
public synchronized TemporaryTopic getTemporaryTopic( ConnectionToken dc )
- throws JMSException {
+ throws JMSException
+ {
SpyTemporaryTopic topic = new SpyTemporaryTopic( "JMS_TT" + ( new Integer(
lastTemporaryTopic++ ).toString() ), dc );
-
+
ClientConsumer ClientConsumer = getClientConsumer( dc );
- synchronized ( destinations ) {
-
+ synchronized ( destinations )
+ {
+
JMSDestination queue = new JMSTopic( topic, ClientConsumer, this );
HashMap newMap = ( HashMap )destinations.clone();
newMap.put( topic, queue );
destinations = newMap;
-
+
}
-
+
return topic;
}
-
+
public synchronized TemporaryQueue getTemporaryQueue( ConnectionToken dc )
- throws JMSException {
-
+ throws JMSException
+ {
+
SpyTemporaryQueue newQueue = new SpyTemporaryQueue( "JMS_TQ" + ( new Integer(
lastTemporaryQueue++ ).toString() ), dc );
-
+
ClientConsumer ClientConsumer = getClientConsumer( dc );
- synchronized ( destinations ) {
-
+ synchronized ( destinations )
+ {
+
JMSDestination queue = new JMSQueue( newQueue, ClientConsumer, this );
HashMap newMap = ( HashMap )destinations.clone();
newMap.put( newQueue, queue );
destinations = newMap;
-
+
}
-
+
return newQueue;
}
-
+
// Gets the ClientConsumers mapped the the connection
// If the connection is not mapped, a new ClientConsumer is created
public ClientConsumer getClientConsumer( ConnectionToken dc )
- throws JMSException {
+ throws JMSException
+ {
ClientConsumer cq = ( ClientConsumer )clientConsumers.get( dc );
- if ( cq == null ) {
+ if ( cq == null )
+ {
cq = new ClientConsumer( this, dc );
clientConsumers.put( dc, cq );
}
return cq;
}
-
- public JMSDestination getJMSDestination( SpyDestination dest ) {
+
+ public JMSDestination getJMSDestination( SpyDestination dest )
+ {
return ( JMSDestination )destinations.get( dest );
}
-
- public StateManager getStateManager() {
+
+ public StateManager getStateManager()
+ {
return stateManager;
}
-
- public org.jboss.mq.pm.PersistenceManager getPersistenceManager() {
+
+ public org.jboss.mq.pm.PersistenceManager getPersistenceManager()
+ {
return persistenceManager;
}
-
- public void startServer() {
- synchronized ( stateLock ) {
+
+ public void startServer()
+ {
+ synchronized ( stateLock )
+ {
this.stopped = false;
}
}
-
- public void stopServer() {
- synchronized ( stateLock ) {
+
+ public void stopServer()
+ {
+ synchronized ( stateLock )
+ {
this.stopped = true;
//Any work that needs doing should be done here
-
+
//At the moment there is nothing to do due to the fact that the individual
//parts of the JBossMQ (pm, ils etc) each have their own mbean service
//which starts and stops them separately
-
+
//We could wait in here for the client consumers to finish delivering any
//messages they have, but it is not neccessary as the client acks will be
//almost certainly be lost anyway.
this.alive = false;
}
}
-
+
public void checkID( String ID )
- throws JMSException {
+ throws JMSException
+ {
stateManager.addLoggedOnClientId( ID );
}
-
+
//A connection has sent a new message
public void addMessage( ConnectionToken dc, SpyMessage val )
- throws JMSException {
+ throws JMSException
+ {
addMessage( dc, val, null );
}
-
+
//A connection has sent a new message
public void addMessage( ConnectionToken dc, SpyMessage val, Tx txId )
- throws JMSException {
+ throws JMSException
+ {
JMSDestination queue = ( JMSDestination )destinations.get(
val.getJMSDestination() );
- if ( queue == null ) {
+ if ( queue == null )
+ {
throw new JMSException( "This destination does not exist !" );
}
-
+
//Add the message to the queue
val.setReadOnlyMode();
queue.addMessage( val, txId );
}
-
+
/**
* The following function performs a Unit Of Work.
*
@@ -258,245 +288,296 @@
* @exception JMSException Description of Exception
*/
public void transact( ConnectionToken dc, TransactionRequest t )
- throws JMSException {
-
+ throws JMSException
+ {
+
org.jboss.mq.pm.TxManager txManager = persistenceManager.getTxManager();
- if ( t.requestType == t.ONE_PHASE_COMMIT_REQUEST ) {
-
+ if ( t.requestType == t.ONE_PHASE_COMMIT_REQUEST )
+ {
+
Tx txId = txManager.createTx();
-
- try {
-
- if ( t.messages != null ) {
- for ( int i = 0; i < t.messages.length; i++ ) {
+
+ try
+ {
+
+ if ( t.messages != null )
+ {
+ for ( int i = 0; i < t.messages.length; i++ )
+ {
addMessage( dc, t.messages[i], txId );
}
}
-
- if ( t.acks != null ) {
- for ( int i = 0; i < t.acks.length; i++ ) {
+
+ if ( t.acks != null )
+ {
+ for ( int i = 0; i < t.acks.length; i++ )
+ {
acknowledge( dc, t.acks[i], txId );
}
}
-
+
txManager.commitTx( txId );
-
- } catch ( JMSException e ) {
+
+ } catch ( JMSException e )
+ {
txManager.rollbackTx( txId );
TransactionRolledBackException newE = new
TransactionRolledBackException( "Transaction was rolled back." );
newE.setLinkedException( e );
throw newE;
}
- } else if ( t.requestType == t.TWO_PHASE_COMMIT_PREPARE_REQUEST ) {
-
+ } else if ( t.requestType == t.TWO_PHASE_COMMIT_PREPARE_REQUEST )
+ {
+
Tx txId = txManager.createTx( dc, t.xid );
- try {
-
- if ( t.messages != null ) {
- for ( int i = 0; i < t.messages.length; i++ ) {
+ try
+ {
+
+ if ( t.messages != null )
+ {
+ for ( int i = 0; i < t.messages.length; i++ )
+ {
addMessage( dc, t.messages[i], txId );
}
}
-
- if ( t.acks != null ) {
- for ( int i = 0; i < t.acks.length; i++ ) {
+
+ if ( t.acks != null )
+ {
+ for ( int i = 0; i < t.acks.length; i++ )
+ {
acknowledge( dc, t.acks[i], txId );
}
}
- } catch ( JMSException e ) {
+ } catch ( JMSException e )
+ {
txManager.rollbackTx( txId );
TransactionRolledBackException newE = new
TransactionRolledBackException( "Transaction was rolled back." );
newE.setLinkedException( e );
throw newE;
}
- } else if ( t.requestType == t.TWO_PHASE_COMMIT_ROLLBACK_REQUEST ) {
-
+ } else if ( t.requestType == t.TWO_PHASE_COMMIT_ROLLBACK_REQUEST )
+ {
+
Tx txId = txManager.getPrepared( dc, t.xid );
txManager.rollbackTx( txId );
-
- } else if ( t.requestType == t.TWO_PHASE_COMMIT_COMMIT_REQUEST ) {
-
+
+ } else if ( t.requestType == t.TWO_PHASE_COMMIT_COMMIT_REQUEST )
+ {
+
Tx txId = txManager.getPrepared( dc, t.xid );
txManager.commitTx( txId );
-
+
}
}
-
+
//Sent by a client to Ack or Nack a message.
public void acknowledge( ConnectionToken dc, AcknowledgementRequest item )
- throws JMSException {
+ throws JMSException
+ {
acknowledge( dc, item, null );
}
-
+
//Sent by a client to Ack or Nack a message.
public void acknowledge( ConnectionToken dc, AcknowledgementRequest item, Tx
txId )
- throws JMSException {
-
+ throws JMSException
+ {
+
ClientConsumer cc = getClientConsumer( dc );
cc.acknowledge( item, txId );
-
+
}
-
+
//A connection is closing [error or notification]
public synchronized void connectionClosing( ConnectionToken dc )
- throws JMSException {
- cat.debug( "JMSServer->connectionClosing(dc=" + dc + ")" );
- if ( dc == null ) {
+ throws JMSException
+ {
+ if( log.isTraceEnabled() )
+ log.trace( "JMSServer->connectionClosing(dc=" + dc + ")" );
+ if ( dc == null )
+ {
return;
}
-
+
// Close it's ClientConsumer
ClientConsumer cq = ( ClientConsumer )clientConsumers.remove( dc );
- if ( cq != null ) {
+ if ( cq != null )
+ {
cq.close();
}
-
+
//unregister its clientID
- if ( dc.getClientID() != null ) {
+ if ( dc.getClientID() != null )
+ {
stateManager.removeLoggedOnClientId( dc.getClientID() );
}
-
+
//Remove any temporary destinations the consumer may have created.
- synchronized ( destinations ) {
-
+ synchronized ( destinations )
+ {
+
Iterator i = destinations.values().iterator();
- while ( i.hasNext() ) {
-
+ while ( i.hasNext() )
+ {
+
JMSDestination sq = ( JMSDestination )i.next();
- if ( dc.equals( sq.temporaryDestination ) ) {
+ if ( dc.equals( sq.temporaryDestination ) )
+ {
i.remove();
}
}
-
+
}
-
+
}
-
+
public void connectionFailure( ConnectionToken dc )
- throws JMSException {
- cat.debug( "JMSServer->connectionFailure(dc=" + dc + ")" );
-
+ throws JMSException
+ {
//We should try again :) This behavior should under control of a
Failure-Plugin
- cat.error( "The connection to client " + dc.getClientID() + " failed." );
-
+ log.error( "The connection to client " + dc.getClientID() + " failed." );
connectionClosing( dc );
}
-
+
//A connection object wants to subscribe to a Destination
public void subscribe( ConnectionToken dc, Subscription sub )
- throws JMSException {
- try {
- cat.debug( "Server: subscribe(dest=" + sub.destination + ",idConnection="
+ dc.getClientID() + ")" );
+ throws JMSException
+ {
+ try
+ {
+ if( log.isTraceEnabled() )
+ log.trace( "Server: subscribe(dest=" + sub.destination +
",idConnection=" + dc.getClientID() + ")" );
ClientConsumer ClientConsumer = getClientConsumer( dc );
ClientConsumer.addSubscription( sub );
- } catch ( JMSException e ) {
- cat.debug( "Exception:", e );
+ }
+ catch ( JMSException e )
+ {
+ log.debug( "Exception:", e );
throw e;
}
-
+
}
-
+
public void unsubscribe( ConnectionToken dc, int subscriptionId )
- throws JMSException {
- cat.debug( "Server: unsubscribe(idConnection=" + dc.getClientID() + ")" );
+ throws JMSException
+ {
+ if( log.isTraceEnabled() )
+ log.trace( "Server: unsubscribe(idConnection=" + dc.getClientID() + ")" );
ClientConsumer ClientConsumer = getClientConsumer( dc );
ClientConsumer.removeSubscription( subscriptionId );
}
-
+
public void destroySubscription( DurableSubcriptionID id )
- throws JMSException {
+ throws JMSException
+ {
getStateManager().setDurableSubscription( this, id, null );
}
-
+
public SpyMessage[] browse( ConnectionToken dc, Destination dest, String
selector )
- throws JMSException {
-
+ throws JMSException
+ {
+
JMSDestination queue = ( JMSDestination )destinations.get( dest );
- if ( queue == null ) {
+ if ( queue == null )
+ {
throw new JMSException( "That destination does not exist" );
}
- if ( !( queue instanceof JMSQueue ) ) {
+ if ( !( queue instanceof JMSQueue ) )
+ {
throw new JMSException( "That destination is not a queue" );
}
-
+
return ( ( JMSQueue )queue ).browse( selector );
}
-
+
public SpyMessage receive( ConnectionToken dc, int subscriberId, long wait )
- throws JMSException {
+ throws JMSException
+ {
ClientConsumer ClientConsumer = getClientConsumer( dc );
return ClientConsumer.receive( subscriberId, wait );
}
-
+
public synchronized Queue createQueue( ConnectionToken dc, String name )
- throws JMSException {
- cat.debug( "createQueue(" + name + ")" );
-
+ throws JMSException
+ {
+ log.debug( "createQueue(" + name + ")" );
+
SpyQueue newQueue = new SpyQueue( name );
- if ( !destinations.containsKey( newQueue ) ) {
+ if ( !destinations.containsKey( newQueue ) )
+ {
throw new JMSException( "This destination does not exist !" );
}
return newQueue;
}
-
+
public synchronized Topic createTopic( ConnectionToken dc, String name )
- throws JMSException {
- cat.debug( "createTopic(" + name + ")" );
-
+ throws JMSException
+ {
+ log.debug( "createTopic(" + name + ")" );
+
SpyTopic newTopic = new SpyTopic( name );
- if ( !destinations.containsKey( newTopic ) ) {
+ if ( !destinations.containsKey( newTopic ) )
+ {
throw new JMSException( "This destination does not exist !" );
}
return newTopic;
}
-
- public synchronized void deleteTemporaryDestination( ConnectionToken dc,
SpyDestination dest ) {
- cat.debug( "deleteDestination(dest=" + dest.toString() + ")" );
-
- synchronized ( destinations ) {
+
+ public synchronized void deleteTemporaryDestination( ConnectionToken dc,
SpyDestination dest )
+ {
+ log.debug( "deleteDestination(dest=" + dest.toString() + ")" );
+
+ synchronized ( destinations )
+ {
HashMap newMap = ( HashMap )destinations.clone();
newMap.remove( dest );
destinations = newMap;
}
-
+
}
public String checkUser( String userName, String password )
- throws JMSException {
+ throws JMSException
+ {
return stateManager.checkUser( userName, password );
}
-
+
// Administration calls
public void addDestination( SpyDestination topic, JMSDestination queue )
- throws JMSException {
- cat.debug( "addDestination: " + topic + ", " + queue );
- if ( destinations.containsKey( topic ) ) {
+ throws JMSException
+ {
+ log.debug( "addDestination: " + topic + ", " + queue );
+ if ( destinations.containsKey( topic ) )
+ {
throw new JMSException( "This destination already exists !" );
}
-
+
//Add this new destination to the list
- synchronized ( destinations ) {
+ synchronized ( destinations )
+ {
HashMap newMap = ( HashMap )destinations.clone();
newMap.put( topic, queue );
destinations = newMap;
}
}
-
- public void restoreMessage( SpyMessage message ) {
+
+ public void restoreMessage( SpyMessage message )
+ {
JMSDestination queue = ( JMSDestination )destinations.get(
message.getJMSDestination() );
- if ( queue == null ) {
+ if ( queue == null )
+ {
throw new RuntimeException( "This destination does not exist!" );
}
//Add the message to the queue
queue.restoreMessage( message );
}
-
- public String toString() {
+
+ public String toString()
+ {
SpyConnectionMetaData md = new SpyConnectionMetaData();
return md.getJMSProviderName()+" version "+md.getProviderVersion();
}
-
- public static JMSServer getInstance() {
+
+ public static JMSServer getInstance()
+ {
return theInstance;
}
}
_______________________________________________
Jboss-development mailing list
[EMAIL PROTECTED]
https://lists.sourceforge.net/lists/listinfo/jboss-development
