activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rajdav...@apache.org
Subject svn commit: r658154 - in /activemq/trunk/activemq-core/src/main/java/org/apache/activemq: ActiveMQConnection.java ActiveMQConnectionFactory.java ActiveMQMessageConsumer.java ActiveMQSession.java
Date Tue, 20 May 2008 08:14:48 GMT
Author: rajdavies
Date: Tue May 20 01:14:47 2008
New Revision: 658154

URL: http://svn.apache.org/viewvc?rev=658154&view=rev
Log:
Fix for https://issues.apache.org/activemq/browse/AMQ-1735

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java?rev=658154&r1=658153&r2=658154&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
Tue May 20 01:14:47 2008
@@ -143,6 +143,7 @@
     private boolean watchTopicAdvisories = true;
     private long warnAboutUnstartedConnectionTimeout = 500L;
     private int sendTimeout =0;
+    private boolean sendAcksAsync=true;
 
     private final Transport transport;
     private final IdGenerator clientIdGenerator;
@@ -1533,6 +1534,20 @@
     public void setSendTimeout(int sendTimeout) {
         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;
+    }
 
 
     /**

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java?rev=658154&r1=658153&r2=658154&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
Tue May 20 01:14:47 2008
@@ -107,6 +107,7 @@
     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 @@
         connection.setProducerWindowSize(getProducerWindowSize());
         connection.setWarnAboutUnstartedConnectionTimeout(getWarnAboutUnstartedConnectionTimeout());
         connection.setSendTimeout(getSendTimeout());
+        connection.setSendAcksAsync(isSendAcksAsync());
         if (transportListener != null) {
             connection.addTransportListener(transportListener);
         }
@@ -549,6 +551,20 @@
     public void setSendTimeout(int sendTimeout) {
         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;
+    }
 
 
     /**
@@ -645,6 +661,7 @@
         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() {

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=658154&r1=658153&r2=658154&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
Tue May 20 01:14:47 2008
@@ -624,7 +624,7 @@
                 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 @@
                                 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 @@
                             }
                         } else {
                             MessageAck ack = new MessageAck(md,MessageAck.STANDARD_ACK_TYPE,deliveredMessages.size());
-                            session.asyncSendPacket(ack);
+                            session.sendAck(ack);
                             deliveredMessages.clear();
                         }
                     }
@@ -815,7 +815,7 @@
         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 @@
                 session.doStartTransaction();
                 ack.setTransactionId(session.getTransactionContext().getTransactionId());
             }
-            session.asyncSendPacket(ack);
+            session.sendAck(ack);
     
             // Adjust the counters
             deliveredCounter -= deliveredMessages.size();
@@ -859,7 +859,7 @@
     
     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 @@
                     // 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 @@
                 } 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();

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java?rev=658154&r1=658153&r2=658154&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java Tue
May 20 01:14:47 2008
@@ -1903,5 +1903,17 @@
         }
         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);
+        }
+    }
 
 }



Mime
View raw message