svn commit: r587889 - in /incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client: AMQSession_0_10.java BasicMessageConsumer.java BasicMessageConsumer_0_10.java

[arnaudsimon]

Author: arnaudsimon
Date: Wed Oct 24 06:22:02 2007
New Revision: 587889

URL: http://svn.apache.org/viewvc?rev=587889&view=rev
Log:
Changed to handle sync receive when connection is started

Modified:
    
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
    
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
    
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java

Modified: 
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java?rev=587889&r1=587888&r2=587889&view=diff
==============================================================================
--- 
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
 (original)
+++ 
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
 Wed Oct 24 06:22:02 2007
@@ -556,11 +556,23 @@
     {
 
         super.suspendChannel(false);
-
+        for(BasicMessageConsumer  c:  _consumers.values())
+        {
+              c.start();
+        }
         // If the event dispatcher is not running then start it too.
         if (hasMessageListeners())
         {
             startDistpatcherIfNecessary();
+        }
+    }
+
+     void stop() throws AMQException
+    {
+        super.stop();
+           for(BasicMessageConsumer  c:  _consumers.values())
+        {
+              c.stop();
         }
     }
 

Modified: 
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java?rev=587889&r1=587888&r2=587889&view=diff
==============================================================================
--- 
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
 (original)
+++ 
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
 Wed Oct 24 06:22:02 2007
@@ -958,4 +958,16 @@
     {
         _synchronousQueue.clear();
     }
+
+
+    public void start()
+    {
+        // do nothing as this is a 0_10 feature
+    }
+
+
+    public void stop()
+    {
+        // do nothing as this is a 0_10 feature
+    }
 }

Modified: 
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java?rev=587889&r1=587888&r2=587889&view=diff
==============================================================================
--- 
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
 (original)
+++ 
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
 Wed Oct 24 06:22:02 2007
@@ -79,6 +79,11 @@
      */
     private boolean _preAcquire = true;
 
+    /**
+     * Indicate whether this consumer is started.
+     */
+    private boolean _isStarted = false;
+
     //--- constructor
     protected BasicMessageConsumer_0_10(int channelId, AMQConnection 
connection, AMQDestination destination,
                                         String messageSelector, boolean 
noLocal, MessageFactoryRegistry messageFactory,
@@ -105,6 +110,7 @@
                 _preAcquire = false;
             }
         }
+         _isStarted  = connection.started();
     }
 
     // ----- Interface org.apache.qpidity.client.util.MessageListener
@@ -449,6 +455,10 @@
 
     public Object getMessageFromQueue(long l) throws InterruptedException
     {
+        if( !_isStarted )
+        {
+            return null;
+        }
         Object o;
         _0_10session.getQpidSession().messageFlow(getConsumerTag().toString(),
                 
org.apache.qpidity.nclient.Session.MESSAGE_FLOW_UNIT_MESSAGE,1);
@@ -496,5 +506,15 @@
     private class NullTocken
     {
 
+    }
+        
+    public void start()
+    {
+       _isStarted = true;
+    }
+
+    public void stop()
+    {
+       _isStarted = false;
     }
 }