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);
+ }
+ }
}
