svn commit: r718403 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/ActiveMQMessageConsumer.java main/java/org/apache/activemq/broker/TransportConnection.java test/java/org/apache/activemq/transport/failover/AMQ1925Test.java

[gtully]

Author: gtully
Date: Mon Nov 17 13:55:46 2008
New Revision: 718403

URL: http://svn.apache.org/viewvc?rev=718403&view=rev
Log:
resolve AMQ-2004

Modified:
    
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
    
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
    
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/AMQ1925Test.java

Modified: 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java?rev=718403&r1=718402&r2=718403&view=diff
==============================================================================
--- 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
 (original)
+++ 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
 Mon Nov 17 13:55:46 2008
@@ -617,7 +617,7 @@
                                ackCounter = 0;
                        }
                }
-            } else {
+            } else if (pendingAck != null && pendingAck.isStandardAck()) {
                 ack = pendingAck;
             }
             if (ack != null) {

Modified: 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java?rev=718403&r1=718402&r2=718403&view=diff
==============================================================================
--- 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
 (original)
+++ 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
 Mon Nov 17 13:55:46 2008
@@ -304,10 +304,12 @@
         try {
             response = command.visit(this);
         } catch (Throwable e) {
+            if (SERVICELOG.isDebugEnabled() && e.getClass() != 
BrokerStoppedException.class) {
+                SERVICELOG.debug("Error occured while processing "
+                        + (responseRequired ? "sync": "async")
+                        + " command: " + command + ", exception: " + e, e);
+            }
             if (responseRequired) {
-                if (SERVICELOG.isDebugEnabled() && e.getClass() != 
BrokerStoppedException.class) {
-                    SERVICELOG.debug("Error occured while processing sync 
command: " + e, e);
-                }
                 response = new ExceptionResponse(e);
             } else {
                 serviceException(e);

Modified: 
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/AMQ1925Test.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/AMQ1925Test.java?rev=718403&r1=718402&r2=718403&view=diff
==============================================================================
--- 
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/AMQ1925Test.java
 (original)
+++ 
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/AMQ1925Test.java
 Mon Nov 17 13:55:46 2008
@@ -24,6 +24,7 @@
 
 import javax.jms.Connection;
 import javax.jms.DeliveryMode;
+import javax.jms.ExceptionListener;
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
@@ -49,7 +50,7 @@
  * 
  * @version $Revision: 1.1 $
  */
-public class AMQ1925Test extends TestCase {
+public class AMQ1925Test extends TestCase implements ExceptionListener {
        private static final Logger log = Logger.getLogger(AMQ1925Test.class);
 
        private static final String QUEUE_NAME = "test.amq1925";
@@ -60,6 +61,8 @@
        private URI tcpUri;
        private ActiveMQConnectionFactory cf;
 
+    private JMSException exception;
+
        public void XtestAMQ1925_TXInProgress() throws Exception {
                Connection connection = cf.createConnection();
                connection.start();
@@ -255,6 +258,7 @@
        public void testAMQ1925_TXBegin() throws Exception {
                Connection connection = cf.createConnection();
                connection.start();
+               connection.setExceptionListener(this);
                Session session = connection.createSession(true,
                                Session.SESSION_TRANSACTED);
                MessageConsumer consumer = session.createConsumer(session
@@ -284,6 +288,7 @@
                connection.close();
 
                assertQueueEmpty();
+               assertNull("no exception on connection listener: " + exception, 
exception);
        }
 
        public void testAMQ1925_TXCommited() throws Exception {
@@ -371,6 +376,7 @@
        }
 
        protected void setUp() throws Exception {
+           exception = null;
                bs = new BrokerService();
                bs.setDeleteAllMessagesOnStartup(true);
                bs.setPersistent(true);
@@ -388,4 +394,8 @@
                new ServiceStopper().stop(bs);
        }
 
+    public void onException(JMSException exception) {
+        this.exception = exception;    
+    }
+
 }