Fix for https://issues.apache.org/activemq/browse/AMQ-1735

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@658154 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java

@@ -143,6 +143,7 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
     private boolean watchTopicAdvisories = true;
     private long warnAboutUnstartedConnectionTimeout = 500L;
     private int sendTimeout =0;
+    private boolean sendAcksAsync=true;
 
     private final Transport transport;
     private final IdGenerator clientIdGenerator;
@@ -1534,6 +1535,20 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
         this.sendTimeout = sendTimeout;
     }
     
+    /**
+     * @return the sendAcksAsync
+     */
+    public boolean isSendAcksAsync() {
+        return sendAcksAsync;
+    }
+
+    /**
+     * @param sendAcksAsync the sendAcksAsync to set
+     */
+    public void setSendAcksAsync(boolean sendAcksAsync) {
+        this.sendAcksAsync = sendAcksAsync;
+    }
+
 
     /**
      * Returns the time this connection was created

activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java

@@ -107,6 +107,7 @@ public class ActiveMQConnectionFactory extends JNDIBaseStorable implements Conne
     private int producerWindowSize = DEFAULT_PRODUCER_WINDOW_SIZE;
     private long warnAboutUnstartedConnectionTimeout = 500L;
     private int sendTimeout =0;
+    private boolean sendAcksAsync=true;
     private TransportListener transportListener;
 
     // /////////////////////////////////////////////
@@ -304,6 +305,7 @@ public class ActiveMQConnectionFactory extends JNDIBaseStorable implements Conne
         connection.setProducerWindowSize(getProducerWindowSize());
         connection.setWarnAboutUnstartedConnectionTimeout(getWarnAboutUnstartedConnectionTimeout());
         connection.setSendTimeout(getSendTimeout());
+        connection.setSendAcksAsync(isSendAcksAsync());
         if (transportListener != null) {
             connection.addTransportListener(transportListener);
         }
@@ -550,6 +552,20 @@ public class ActiveMQConnectionFactory extends JNDIBaseStorable implements Conne
         this.sendTimeout = sendTimeout;
     }
     
+    /**
+     * @return the sendAcksAsync
+     */
+    public boolean isSendAcksAsync() {
+        return sendAcksAsync;
+    }
+
+    /**
+     * @param sendAcksAsync the sendAcksAsync to set
+     */
+    public void setSendAcksAsync(boolean sendAcksAsync) {
+        this.sendAcksAsync = sendAcksAsync;
+    }
+
 
     /**
      * Sets the transformer used to transform messages before they are sent on
@@ -645,6 +661,7 @@ public class ActiveMQConnectionFactory extends JNDIBaseStorable implements Conne
         props.setProperty("alwaysSyncSend", Boolean.toString(isAlwaysSyncSend()));
         props.setProperty("producerWindowSize", Integer.toString(getProducerWindowSize()));
         props.setProperty("sendTimeout", Integer.toString(getSendTimeout()));
+        props.setProperty("sendAcksAsync",Boolean.toString(isSendAcksAsync()));
     }
 
     public boolean isUseCompression() {

activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java

@@ -624,7 +624,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
                 executorService.submit(new Runnable() {
                     public void run() {
                         try {
-                            session.asyncSendPacket(ackToSend);
+                            session.sendAck(ackToSend,true);
                         } catch (JMSException e) {
                             LOG.error(getConsumerId() + " failed to delivered acknowledgements", e);
                         } finally {
@@ -757,7 +757,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
                                 if (ackCounter >= (info
                                         .getCurrentPrefetchSize() * .65)) {
                                     MessageAck ack = new MessageAck(md,MessageAck.STANDARD_ACK_TYPE,deliveredMessages.size());
-                                    session.asyncSendPacket(ack);
+                                    session.sendAck(ack);
                                     ackCounter = 0;
                                     deliveredMessages.clear();
                                 }
@@ -765,7 +765,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
                             }
                         } else {
                             MessageAck ack = new MessageAck(md,MessageAck.STANDARD_ACK_TYPE,deliveredMessages.size());
-                            session.asyncSendPacket(ack);
+                            session.sendAck(ack);
                             deliveredMessages.clear();
                         }
                     }
@@ -815,7 +815,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
         if ((0.5 * info.getPrefetchSize()) <= (deliveredCounter - additionalWindowSize)) {
             MessageAck ack = new MessageAck(md, ackType, deliveredCounter);
             ack.setTransactionId(session.getTransactionContext().getTransactionId());
-            session.asyncSendPacket(ack);
+            session.sendAck(ack);
             additionalWindowSize = deliveredCounter;
 
             // When using DUPS ok, we do a real ack.
@@ -845,7 +845,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
                 session.doStartTransaction();
                 ack.setTransactionId(session.getTransactionContext().getTransactionId());
             }
-            session.asyncSendPacket(ack);
+            session.sendAck(ack);
     
             // Adjust the counters
             deliveredCounter -= deliveredMessages.size();
@@ -859,7 +859,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
     
     void acknowledge(MessageDispatch md) throws JMSException {
         MessageAck ack = new MessageAck(md,MessageAck.INDIVIDUAL_ACK_TYPE,1);
-        session.asyncSendPacket(ack);
+        session.sendAck(ack);
         synchronized(deliveredMessages){
             deliveredMessages.remove(md);
         }
@@ -910,7 +910,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
                     // Acknowledge the last message.
                     
                     MessageAck ack = new MessageAck(lastMd, MessageAck.POSION_ACK_TYPE, deliveredMessages.size());
-                    session.asyncSendPacket(ack);
+                    session.sendAck(ack,true);
                     // ensure we don't filter this as a duplicate
                     session.connection.rollbackDuplicate(this, lastMd.getMessage());
                     // Adjust the window size.
@@ -919,7 +919,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
                 } else {
                     
                     MessageAck ack = new MessageAck(lastMd, MessageAck.REDELIVERED_ACK_TYPE, deliveredMessages.size());
-                    session.asyncSendPacket(ack);
+                    session.sendAck(ack,true);
     
                     // stop the delivery of messages.
                     unconsumedMessages.stop();

activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java

@@ -1904,4 +1904,16 @@ public class ActiveMQSession implements Session, QueueSession, TopicSession, Sta
         return false;
     }
     
+    protected void sendAck(MessageAck ack) throws JMSException {
+        sendAck(ack,false);
+    }
+    
+    protected void sendAck(MessageAck ack, boolean lazy) throws JMSException {
+        if (lazy || connection.isSendAcksAsync() || isTransacted()) {
+            asyncSendPacket(ack);
+        } else {
+            syncSendPacket(ack);
+        }
+    }
+
 }