You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2011/10/06 18:29:00 UTC
svn commit: r1179694 - in
/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client: ./
message/
Author: robbie
Date: Thu Oct 6 16:28:59 2011
New Revision: 1179694
URL: http://svn.apache.org/viewvc?rev=1179694&view=rev
Log:
QPID-3526: move duplicated methods into the parent abstract message delegate, remove dead code from BassicMessageConsumer
Applied patch from Oleksandr Rudyy<or...@gmail.com> and myself.
Modified:
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java?rev=1179694&r1=1179693&r2=1179694&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java Thu Oct 6 16:28:59 2011
@@ -838,63 +838,6 @@ public abstract class BasicMessageConsum
return null;
}
- /**
- * Acknowledge up to last message delivered (if any). Used when commiting.
- */
- void acknowledgeDelivered()
- {
- synchronized(_commitLock)
- {
- ArrayList<Long> tagsToAck = new ArrayList<Long>();
-
- while (!_receivedDeliveryTags.isEmpty())
- {
- tagsToAck.add(_receivedDeliveryTags.poll());
- }
-
- Collections.sort(tagsToAck);
-
- long prevAcked = _lastAcked;
- long oldAckPoint = -1;
-
- while(oldAckPoint != prevAcked)
- {
- oldAckPoint = prevAcked;
-
- Iterator<Long> tagsToAckIterator = tagsToAck.iterator();
-
- while(tagsToAckIterator.hasNext() && tagsToAckIterator.next() == prevAcked+1)
- {
- tagsToAckIterator.remove();
- prevAcked++;
- }
-
- Iterator<Long> previousAckIterator = _previouslyAcked.iterator();
- while(previousAckIterator.hasNext() && previousAckIterator.next() == prevAcked+1)
- {
- previousAckIterator.remove();
- prevAcked++;
- }
-
- }
- if(prevAcked != _lastAcked)
- {
- _session.acknowledgeMessage(prevAcked, true);
- _lastAcked = prevAcked;
- }
-
- Iterator<Long> tagsToAckIterator = tagsToAck.iterator();
-
- while(tagsToAckIterator.hasNext())
- {
- Long tag = tagsToAckIterator.next();
- _session.acknowledgeMessage(tag, false);
- _previouslyAcked.add(tag);
- }
- }
- }
-
-
void notifyError(Throwable cause)
{
// synchronized (_closed)
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate.java?rev=1179694&r1=1179693&r2=1179694&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate.java Thu Oct 6 16:28:59 2011
@@ -26,9 +26,7 @@ import org.apache.qpid.client.AMQSession
import javax.jms.Destination;
import javax.jms.JMSException;
-import java.nio.ByteBuffer;
import java.util.Enumeration;
-import java.util.Map;
import java.util.UUID;
public interface AMQMessageDelegate
@@ -130,9 +128,9 @@ public interface AMQMessageDelegate
void removeProperty(final String propertyName) throws JMSException;
- void setAMQSession(final AMQSession s);
+ void setAMQSession(final AMQSession<?,?> s);
- AMQSession getAMQSession();
+ AMQSession<?,?> getAMQSession();
long getDeliveryTag();
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java?rev=1179694&r1=1179693&r2=1179694&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_10.java Thu Oct 6 16:28:59 2011
@@ -37,12 +37,10 @@ import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageFormatException;
import javax.jms.MessageNotWriteableException;
-import javax.jms.Session;
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQPInvalidClassException;
import org.apache.qpid.client.AMQDestination;
-import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.AMQSession_0_10;
import org.apache.qpid.client.CustomJMSXProperty;
import org.apache.qpid.framing.AMQShortString;
@@ -76,13 +74,8 @@ public class AMQMessageDelegate_0_10 ext
private Destination _destination;
-
private MessageProperties _messageProps;
private DeliveryProperties _deliveryProps;
- /** If the acknowledge mode is CLIENT_ACKNOWLEDGE the session is required */
- private AMQSession _session;
- private final long _deliveryTag;
-
protected AMQMessageDelegate_0_10()
{
@@ -92,9 +85,9 @@ public class AMQMessageDelegate_0_10 ext
protected AMQMessageDelegate_0_10(MessageProperties messageProps, DeliveryProperties deliveryProps, long deliveryTag)
{
+ super(deliveryTag);
_messageProps = messageProps;
_deliveryProps = deliveryProps;
- _deliveryTag = deliveryTag;
_readableProperties = (_messageProps != null);
AMQDestination dest;
@@ -325,14 +318,14 @@ public class AMQMessageDelegate_0_10 ext
{
try
{
- int type = ((AMQSession_0_10)_session).resolveAddressType(amqd);
+ int type = ((AMQSession_0_10)getAMQSession()).resolveAddressType(amqd);
if (type == AMQDestination.QUEUE_TYPE)
{
- ((AMQSession_0_10)_session).setLegacyFiledsForQueueType(amqd);
+ ((AMQSession_0_10)getAMQSession()).setLegacyFiledsForQueueType(amqd);
}
else
{
- ((AMQSession_0_10)_session).setLegacyFiledsForTopicType(amqd);
+ ((AMQSession_0_10)getAMQSession()).setLegacyFiledsForTopicType(amqd);
}
}
catch(AMQException ex)
@@ -905,64 +898,6 @@ public class AMQMessageDelegate_0_10 ext
_readableProperties = false;
}
-
- public void acknowledgeThis() throws JMSException
- {
- // the JMS 1.1 spec says in section 3.6 that calls to acknowledge are ignored when client acknowledge
- // is not specified. In our case, we only set the session field where client acknowledge mode is specified.
- if (_session != null && _session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
- {
- if (_session.getAMQConnection().isClosed())
- {
- throw new javax.jms.IllegalStateException("Connection is already closed");
- }
-
- // we set multiple to true here since acknowledgment implies acknowledge of all previous messages
- // received on the session
- _session.acknowledgeMessage(_deliveryTag, true);
- }
- }
-
- public void acknowledge() throws JMSException
- {
- if (_session != null && _session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
- {
- _session.acknowledge();
- }
- }
-
-
- /**
- * The session is set when CLIENT_ACKNOWLEDGE mode is used so that the CHANNEL ACK can be sent when the user calls
- * acknowledge()
- *
- * @param s the AMQ session that delivered this message
- */
- public void setAMQSession(AMQSession s)
- {
- _session = s;
- }
-
- public AMQSession getAMQSession()
- {
- return _session;
- }
-
- /**
- * Get the AMQ message number assigned to this message
- *
- * @return the message number
- */
- public long getDeliveryTag()
- {
- return _deliveryTag;
- }
-
-
-
-
-
-
protected void checkPropertyName(CharSequence propertyName)
{
if (propertyName == null)
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java?rev=1179694&r1=1179693&r2=1179694&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AMQMessageDelegate_0_8.java Thu Oct 6 16:28:59 2011
@@ -30,7 +30,6 @@ import java.util.UUID;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageNotWriteableException;
-import javax.jms.Session;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQQueue;
@@ -60,15 +59,12 @@ public class AMQMessageDelegate_0_8 exte
Boolean.parseBoolean(System.getProperties().getProperty(AMQSession.STRICT_AMQP, AMQSession.STRICT_AMQP_DEFAULT));
private ContentHeaderProperties _contentHeaderProperties;
- /** If the acknowledge mode is CLIENT_ACKNOWLEDGE the session is required */
- private AMQSession _session;
- private final long _deliveryTag;
// The base set of items that needs to be set.
private AMQMessageDelegate_0_8(BasicContentHeaderProperties properties, long deliveryTag)
{
+ super(deliveryTag);
_contentHeaderProperties = properties;
- _deliveryTag = deliveryTag;
_readableProperties = (_contentHeaderProperties != null);
_headerAdapter = new JMSHeaderAdapter(_readableProperties ? ((BasicContentHeaderProperties) _contentHeaderProperties).getHeaders()
: (new BasicContentHeaderProperties()).getHeaders() );
@@ -518,58 +514,4 @@ public class AMQMessageDelegate_0_8 exte
_readableProperties = false;
}
-
-
- public void acknowledgeThis() throws JMSException
- {
- // the JMS 1.1 spec says in section 3.6 that calls to acknowledge are ignored when client acknowledge
- // is not specified. In our case, we only set the session field where client acknowledge mode is specified.
- if (_session != null && _session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
- {
- if (_session.getAMQConnection().isClosed())
- {
- throw new javax.jms.IllegalStateException("Connection is already closed");
- }
-
- // we set multiple to true here since acknowledgement implies acknowledge of all previous messages
- // received on the session
- _session.acknowledgeMessage(_deliveryTag, true);
- }
- }
-
- public void acknowledge() throws JMSException
- {
- if (_session != null && _session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
- {
- _session.acknowledge();
- }
- }
-
-
- /**
- * The session is set when CLIENT_ACKNOWLEDGE mode is used so that the CHANNEL ACK can be sent when the user calls
- * acknowledge()
- *
- * @param s the AMQ session that delivered this message
- */
- public void setAMQSession(AMQSession s)
- {
- _session = s;
- }
-
- public AMQSession getAMQSession()
- {
- return _session;
- }
-
- /**
- * Get the AMQ message number assigned to this message
- *
- * @return the message number
- */
- public long getDeliveryTag()
- {
- return _deliveryTag;
- }
-
}
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java?rev=1179694&r1=1179693&r2=1179694&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/message/AbstractAMQMessageDelegate.java Thu Oct 6 16:28:59 2011
@@ -23,9 +23,13 @@ package org.apache.qpid.client.message;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import javax.jms.JMSException;
+import javax.jms.Session;
+
import org.apache.qpid.client.AMQAnyDestination;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.AMQTopic;
import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.framing.AMQShortString;
@@ -78,7 +82,25 @@ public abstract class AbstractAMQMessage
new ExchangeInfo(ExchangeDefaults.HEADERS_EXCHANGE_NAME.toString(),
ExchangeDefaults.HEADERS_EXCHANGE_CLASS.toString(),
AMQDestination.QUEUE_TYPE));
-
+ }
+
+ /** If the acknowledge mode is CLIENT_ACKNOWLEDGE the session is required */
+ private AMQSession<?,?> _session;
+ private final long _deliveryTag;
+
+ protected AbstractAMQMessageDelegate(long deliveryTag)
+ {
+ _deliveryTag = deliveryTag;
+ }
+
+ /**
+ * Get the AMQ message number assigned to this message
+ *
+ * @return the message number
+ */
+ public long getDeliveryTag()
+ {
+ return _deliveryTag;
}
/**
@@ -157,6 +179,47 @@ public abstract class AbstractAMQMessage
{
return _exchangeMap.containsKey(exchange);
}
+
+ public void acknowledgeThis() throws JMSException
+ {
+ // the JMS 1.1 spec says in section 3.6 that calls to acknowledge are ignored when client acknowledge
+ // is not specified. In our case, we only set the session field where client acknowledge mode is specified.
+ if (_session != null && _session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
+ {
+ if (_session.getAMQConnection().isClosed())
+ {
+ throw new javax.jms.IllegalStateException("Connection is already closed");
+ }
+
+ // we set multiple to true here since acknowledgement implies acknowledge of all previous messages
+ // received on the session
+ _session.acknowledgeMessage(getDeliveryTag(), true);
+ }
+ }
+
+ public void acknowledge() throws JMSException
+ {
+ if (_session != null && _session.getAcknowledgeMode() == Session.CLIENT_ACKNOWLEDGE)
+ {
+ _session.acknowledge();
+ }
+ }
+
+ /**
+ * The session is set when CLIENT_ACKNOWLEDGE mode is used so that the CHANNEL ACK can be sent when the user calls
+ * acknowledge()
+ *
+ * @param s the AMQ session that delivered this message
+ */
+ public void setAMQSession(AMQSession<?,?> s)
+ {
+ _session = s;
+ }
+
+ public AMQSession<?,?> getAMQSession()
+ {
+ return _session;
+ }
}
class ExchangeInfo
@@ -202,5 +265,5 @@ class ExchangeInfo
public void setDestType(int destType)
{
this.destType = destType;
- }
+ }
}
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org