You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2006/12/15 20:21:47 UTC
svn commit: r487625 - in
/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client:
AMQSession.java BasicMessageConsumer.java
Author: rgreig
Date: Fri Dec 15 11:21:46 2006
New Revision: 487625
URL: http://svn.apache.org/viewvc?view=rev&rev=487625
Log:
QPID-194 Patch supplied by Rob Godfrey.
Modified:
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java?view=diff&rev=487625&r1=487624&r2=487625
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java Fri Dec 15 11:21:46 2006
@@ -165,6 +165,8 @@
if (consumer == null)
{
_logger.warn("Received a message from queue " + message.deliverBody.consumerTag + " without a handler - ignoring...");
+ _logger.warn("Consumers that exist: " + _consumers);
+ _logger.warn("Session hashcode: " + System.identityHashCode(this));
}
else
{
@@ -958,26 +960,37 @@
* @param queueName
* @return the consumer tag generated by the broker
*/
- private String consumeFromQueue(String queueName, AMQProtocolHandler protocolHandler, int prefetchHigh, int prefetchLow,
- boolean noLocal, boolean exclusive, int acknowledgeMode, boolean nowait) throws AMQException
+ private void consumeFromQueue(BasicMessageConsumer consumer, String queueName, AMQProtocolHandler protocolHandler,
+ boolean nowait) throws AMQException
{
//fixme prefetch values are not used here. Do we need to have them as parametsrs?
//need to generate a consumer tag on the client so we can exploit the nowait flag
String tag = Integer.toString(_nextTag++);
- AMQFrame jmsConsume = BasicConsumeBody.createAMQFrame(_channelId, 0,
- queueName, tag, noLocal,
- acknowledgeMode == Session.NO_ACKNOWLEDGE,
- exclusive, nowait);
- if (nowait)
+ consumer.setConsumerTag(tag);
+ // we must register the consumer in the map before we actually start listening
+ _consumers.put(tag, consumer);
+ try
{
- protocolHandler.writeFrame(jmsConsume);
+ AMQFrame jmsConsume = BasicConsumeBody.createAMQFrame(_channelId, 0,
+ queueName, tag, consumer.isNoLocal(),
+ consumer.getAcknowledgeMode() == Session.NO_ACKNOWLEDGE,
+ consumer.isExclusive(), nowait);
+ if (nowait)
+ {
+ protocolHandler.writeFrame(jmsConsume);
+ }
+ else
+ {
+ protocolHandler.syncWrite(jmsConsume, BasicConsumeOkBody.class);
+ }
}
- else
+ catch (AMQException e)
{
- protocolHandler.syncWrite(jmsConsume, BasicConsumeOkBody.class);
+ // clean-up the map in the event of an error
+ _consumers.remove(tag);
+ throw e;
}
- return tag;
}
public Queue createQueue(String queueName) throws JMSException
@@ -1354,12 +1367,7 @@
bindQueue(amqd, queueName, protocolHandler, consumer.getRawSelectorFieldTable());
- String consumerTag = consumeFromQueue(queueName, protocolHandler, consumer.getPrefetchHigh(),
- consumer.getPrefetchLow(), consumer.isNoLocal(), consumer.isExclusive(),
- consumer.getAcknowledgeMode(), nowait);
-
- consumer.setConsumerTag(consumerTag);
- _consumers.put(consumerTag, consumer);
+ consumeFromQueue(consumer, queueName, protocolHandler, nowait);
}
/**
Modified: incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java?view=diff&rev=487625&r1=487624&r2=487625
==============================================================================
--- incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java (original)
+++ incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java Fri Dec 15 11:21:46 2006
@@ -280,7 +280,7 @@
public Message receive(long l) throws JMSException
{
checkPreConditions();
-
+
acquireReceiving();
try