svn commit: r1367163 - in /activemq/activemq-apollo/trunk: apollo-itests/src/test/java/org/apache/activemq/apollo/JmsQueueBrowserTest.java apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala

[ch...@apache.org]

Author: chirino
Date: Mon Jul 30 16:46:08 2012
New Revision: 1367163

URL: http://svn.apache.org/viewvc?rev=1367163&view=rev
Log:
Fixes APLO-226: Tests fail in JmsQueueBrowserTest

Modified:
    activemq/activemq-apollo/trunk/apollo-itests/src/test/java/org/apache/activemq/apollo/JmsQueueBrowserTest.java
    activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala

Modified: activemq/activemq-apollo/trunk/apollo-itests/src/test/java/org/apache/activemq/apollo/JmsQueueBrowserTest.java
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-itests/src/test/java/org/apache/activemq/apollo/JmsQueueBrowserTest.java?rev=1367163&r1=1367162&r2=1367163&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-itests/src/test/java/org/apache/activemq/apollo/JmsQueueBrowserTest.java (original)
+++ activemq/activemq-apollo/trunk/apollo-itests/src/test/java/org/apache/activemq/apollo/JmsQueueBrowserTest.java Mon Jul 30 16:46:08 2012
@@ -48,7 +48,7 @@ public class JmsQueueBrowserTest extends
      *
      * @throws Exception
      */
-    public void ignoreReceiveBrowseReceive() throws Exception {
+    public void testReceiveBrowseReceive() throws Exception {
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         ActiveMQQueue destination = new ActiveMQQueue("TEST");
         MessageProducer producer = session.createProducer(destination);
@@ -136,7 +136,7 @@ public class JmsQueueBrowserTest extends
     }
 
     // Re-enable once https://issues.apache.org/jira/browse/APLO-226 is fixed.
-    public void ignoreQueueBrowserWith2Consumers() throws Exception {
+    public void testQueueBrowserWith2Consumers() throws Exception {
         final int numMessages = 1000;
 //        connection.setAlwaysSyncSend(false);
         Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);

Modified: activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala
URL: http://svn.apache.org/viewvc/activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala?rev=1367163&r1=1367162&r2=1367163&view=diff
==============================================================================
--- activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala (original)
+++ activemq/activemq-apollo/trunk/apollo-openwire/src/main/scala/org/apache/activemq/apollo/openwire/OpenwireProtocolHandler.scala Mon Jul 30 16:46:08 2012
@@ -202,7 +202,7 @@ class OpenwireProtocolHandler extends Pr
 
     sink_manager = new SinkMux[Command]( connection.transport_sink.map {x=>
       x.setCommandId(next_command_id)
-      debug("sending openwire command: %s", x)
+      trace("sending: %s", x)
       x
     })
     connection_session = new OverflowSink(sink_manager.open());
@@ -958,8 +958,13 @@ class OpenwireProtocolHandler extends Pr
               // session is full so use an overflow sink so to hold the message,
               // and then trigger closing the session once it empties out.
               val sink = new OverflowSink(downstream)
+              var disposed = false
               sink.refiller = ^{
-                dispose
+                // refiller could get triggered multiple times. only care about the first one.
+                if( !disposed ) {
+                  disposed = true
+                  dispose
+                }
               }
               sink.offer(delivery)
             } else {