diff options
Diffstat (limited to 'java/client')
7 files changed, 274 insertions, 251 deletions
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java index 0e3d99eeba..347f5728e2 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java @@ -96,7 +96,8 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect private AMQProtocolHandler _protocolHandler; /** Maps from session id (Integer) to AMQSession instance */ - private final Map _sessions = new LinkedHashMap(); // fixme this is map is replicated in amqprotocolsession as _channelId2SessionMap + private final Map<Integer,AMQSession> _sessions = new LinkedHashMap<Integer,AMQSession>(); + private String _clientName; @@ -508,7 +509,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect AMQSession session = new AMQSession(AMQConnection.this, channelId, transacted, acknowledgeMode, prefetchHigh, prefetchLow); - _protocolHandler.addSessionByChannel(channelId, session); + //_protocolHandler.addSessionByChannel(channelId, session); registerSession(channelId, session); boolean success = false; @@ -527,7 +528,6 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect { if (!success) { - _protocolHandler.removeSessionByChannel(channelId); deregisterSession(channelId); } } @@ -589,7 +589,6 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect } catch (AMQException e) { - _protocolHandler.removeSessionByChannel(channelId); deregisterSession(channelId); throw new AMQException("Error reopening channel " + channelId + " after failover: " + e, e); } @@ -1136,7 +1135,7 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect for (Iterator it = sessions.iterator(); it.hasNext();) { AMQSession s = (AMQSession) it.next(); - _protocolHandler.addSessionByChannel(s.getChannelId(), s); + //_protocolHandler.addSessionByChannel(s.getChannelId(), s); reopenChannel(s.getChannelId(), s.getDefaultPrefetchHigh(), s.getDefaultPrefetchLow(), s.getTransacted()); s.resubscribe(); } @@ -1223,4 +1222,10 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect { _taskPool.execute(task); } + + + public AMQSession getSession(int channelId) + { + return _sessions.get(channelId); + } } diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java index 700222dabb..10101976eb 100644 --- a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java +++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java @@ -209,6 +209,12 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi private final boolean _strictAMQP; + /** System property to enable strickt AMQP compliance */ + public static final String STRICT_AMQP_FATAL = "STRICT_AMQP_FATAL"; + /** Strickt AMQP default */ + public static final String STRICT_AMQP_FATAL_DEFAULT = "true"; + + private final boolean _strictAMQPFATAL; /** System property to enable immediate message prefetching */ public static final String IMMEDIATE_PREFETCH = "IMMEDIATE_PREFETCH"; @@ -429,23 +435,14 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi } } - AMQSession(AMQConnection con, int channelId, boolean transacted, int acknowledgeMode, - MessageFactoryRegistry messageFactoryRegistry) - { - this(con, channelId, transacted, acknowledgeMode, messageFactoryRegistry, DEFAULT_PREFETCH_HIGH_MARK, DEFAULT_PREFETCH_LOW_MARK); - } - AMQSession(AMQConnection con, int channelId, boolean transacted, int acknowledgeMode, - MessageFactoryRegistry messageFactoryRegistry, int defaultPrefetch) - { - this(con, channelId, transacted, acknowledgeMode, messageFactoryRegistry, defaultPrefetch, defaultPrefetch); - } AMQSession(AMQConnection con, int channelId, boolean transacted, int acknowledgeMode, MessageFactoryRegistry messageFactoryRegistry, int defaultPrefetchHighMark, int defaultPrefetchLowMark) { _strictAMQP = Boolean.parseBoolean(System.getProperties().getProperty(STRICT_AMQP, STRICT_AMQP_DEFAULT)); + _strictAMQPFATAL = Boolean.parseBoolean(System.getProperties().getProperty(STRICT_AMQP_FATAL, STRICT_AMQP_FATAL_DEFAULT)); _immediatePrefetch = Boolean.parseBoolean(System.getProperties().getProperty(IMMEDIATE_PREFETCH, IMMEDIATE_PREFETCH_DEFAULT)); _connection = con; @@ -493,15 +490,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi } } - AMQSession(AMQConnection con, int channelId, boolean transacted, int acknowledgeMode) - { - this(con, channelId, transacted, acknowledgeMode, MessageFactoryRegistry.newDefaultRegistry()); - } - AMQSession(AMQConnection con, int channelId, boolean transacted, int acknowledgeMode, int defaultPrefetch) - { - this(con, channelId, transacted, acknowledgeMode, MessageFactoryRegistry.newDefaultRegistry(), defaultPrefetch); - } AMQSession(AMQConnection con, int channelId, boolean transacted, int acknowledgeMode, int defaultPrefetchHigh, int defaultPrefetchLow) { @@ -796,7 +785,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi amqe = new AMQException("Closing session forcibly", e); } _connection.deregisterSession(_channelId); - closeProducersAndConsumers(amqe); + closeProducersAndConsumers(amqe); } } @@ -809,6 +798,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi _closed.set(true); _connection.deregisterSession(_channelId); markClosedProducersAndConsumers(); + } private void markClosedProducersAndConsumers() @@ -941,7 +931,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi getProtocolMajorVersion(), getProtocolMinorVersion(), false)); // requeue - _logger.warn("Session Recover cannot be guaranteed with STRICT_AMQP. Messages may arrive out of order."); + _logger.warn("Session Recover cannot be guaranteed with STRICT_AMQP. Messages may arrive out of order."); } else { @@ -1229,13 +1219,30 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi final int prefetchLow, final boolean noLocal, final boolean exclusive, - final String selector, + String selector, final FieldTable rawSelector, final boolean noConsume, final boolean autoClose) throws JMSException { checkTemporaryDestination(destination); + final String messageSelector; + + if (_strictAMQP && !(selector == null || selector.equals(""))) + { + if (_strictAMQPFATAL) + { + throw new UnsupportedOperationException("Selectors not currently supported by AMQP."); + } + else + { + messageSelector = null; + } + } + else + { + messageSelector = selector; + } return (org.apache.qpid.jms.MessageConsumer) new FailoverSupport() { @@ -1246,6 +1253,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi AMQDestination amqd = (AMQDestination) destination; final AMQProtocolHandler protocolHandler = getProtocolHandler(); + // TODO: Define selectors in AMQP // TODO: construct the rawSelector from the selector string if rawSelector == null final FieldTable ft = FieldTableFactory.newFieldTable(); //if (rawSelector != null) @@ -1254,7 +1262,8 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi { ft.addAll(rawSelector); } - BasicMessageConsumer consumer = new BasicMessageConsumer(_channelId, _connection, amqd, selector, noLocal, + + BasicMessageConsumer consumer = new BasicMessageConsumer(_channelId, _connection, amqd, messageSelector, noLocal, _messageFactoryRegistry, AMQSession.this, protocolHandler, ft, prefetchHigh, prefetchLow, exclusive, _acknowledgeMode, noConsume, autoClose); @@ -1647,6 +1656,8 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException { + + checkNotClosed(); AMQTopic origTopic = checkValidTopic(topic); AMQTopic dest = AMQTopic.createDurableTopic(origTopic, name, _connection); @@ -1674,13 +1685,31 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi { topicName = new AMQShortString(topic.getTopicName()); } - // if the queue is bound to the exchange but NOT for this topic, then the JMS spec - // says we must trash the subscription. - if (isQueueBound(dest.getExchangeName(), dest.getAMQQueueName()) && - !isQueueBound(dest.getExchangeName(), dest.getAMQQueueName(), topicName)) + + if (_strictAMQP) { + if (_strictAMQPFATAL) + { + throw new UnsupportedOperationException("JMS Durable not currently supported by AMQP."); + } + else + { + _logger.warn("Unable to determine if subscription already exists for '" + topicName + "' " + + "for creation durableSubscriber. Requesting queue deletion regardless."); + } + deleteQueue(dest.getAMQQueueName()); } + else + { + // if the queue is bound to the exchange but NOT for this topic, then the JMS spec + // says we must trash the subscription. + if (isQueueBound(dest.getExchangeName(), dest.getAMQQueueName()) && + !isQueueBound(dest.getExchangeName(), dest.getAMQQueueName(), topicName)) + { + deleteQueue(dest.getAMQQueueName()); + } + } } subscriber = new TopicSubscriberAdaptor(dest, (BasicMessageConsumer) createConsumer(dest)); @@ -1778,13 +1807,31 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi } else { - if (isQueueBound(getDefaultTopicExchangeName(), AMQTopic.getDurableTopicQueueName(name, _connection))) + if (_strictAMQP) { + if (_strictAMQPFATAL) + { + throw new UnsupportedOperationException("JMS Durable not currently supported by AMQP."); + } + else + { + _logger.warn("Unable to determine if subscription already exists for '" + name + "' for unsubscribe." + + " Requesting queue deletion regardless."); + } + deleteQueue(AMQTopic.getDurableTopicQueueName(name, _connection)); } else { - throw new InvalidDestinationException("Unknown subscription exchange:" + name); + + if (isQueueBound(getDefaultTopicExchangeName(), AMQTopic.getDurableTopicQueueName(name, _connection))) + { + deleteQueue(AMQTopic.getDurableTopicQueueName(name, _connection)); + } + else + { + throw new InvalidDestinationException("Unknown subscription exchange:" + name); + } } } } @@ -1796,10 +1843,6 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi boolean isQueueBound(AMQShortString exchangeName, AMQShortString queueName, AMQShortString routingKey) throws JMSException { - if (isStrictAMQP()) - { - throw new UnsupportedOperationException(); - } // TODO: Be aware of possible changes to parameter order as versions change. AMQFrame boundFrame = ExchangeBoundBody.createAMQFrame(_channelId, diff --git a/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java b/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java index e0c4b61333..a219f7d655 100644 --- a/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java +++ b/java/client/src/main/java/org/apache/qpid/client/QueueSenderAdapter.java @@ -9,119 +9,141 @@ import javax.jms.Queue; import javax.jms.QueueSender; import javax.jms.InvalidDestinationException; -public class QueueSenderAdapter implements QueueSender { - - private BasicMessageProducer _delegate; - private Queue _queue; - private boolean closed = false; - - public QueueSenderAdapter(BasicMessageProducer msgProducer, Queue queue){ - _delegate = msgProducer; - _queue = queue; - } - - public Queue getQueue() throws JMSException { - checkPreConditions(); - return _queue; - } - - public void send(Message msg) throws JMSException { - checkPreConditions(); - _delegate.send(msg); - } - - public void send(Queue queue, Message msg) throws JMSException { - checkPreConditions(queue); - _delegate.send(queue, msg); - } - - public void publish(Message msg, int deliveryMode, int priority, long timeToLive) - throws JMSException { - checkPreConditions(); - _delegate.send(msg, deliveryMode,priority,timeToLive); - } - - public void send(Queue queue,Message msg, int deliveryMode, int priority, long timeToLive) - throws JMSException { - checkPreConditions(queue); - _delegate.send(queue,msg, deliveryMode,priority,timeToLive); - } - - public void close() throws JMSException { - _delegate.close(); - closed = true; - } - - public int getDeliveryMode() throws JMSException { - checkPreConditions(); - return _delegate.getDeliveryMode(); - } - - public Destination getDestination() throws JMSException { - checkPreConditions(); - return _delegate.getDestination(); - } - - public boolean getDisableMessageID() throws JMSException { - checkPreConditions(); - return _delegate.getDisableMessageID(); - } - - public boolean getDisableMessageTimestamp() throws JMSException { - checkPreConditions(); - return _delegate.getDisableMessageTimestamp(); - } - - public int getPriority() throws JMSException { - checkPreConditions(); - return _delegate.getPriority(); - } - - public long getTimeToLive() throws JMSException { - checkPreConditions(); - return _delegate.getTimeToLive(); - } - - public void send(Destination dest, Message msg) throws JMSException { - checkPreConditions((Queue)dest); - _delegate.send(dest,msg); - } - - public void send(Message msg, int deliveryMode, int priority, long timeToLive) - throws JMSException { - checkPreConditions(); - _delegate.send(msg, deliveryMode,priority,timeToLive); - } - - public void send(Destination dest, Message msg, int deliveryMode, int priority, long timeToLive) throws JMSException { - checkPreConditions((Queue)dest); - _delegate.send(dest,msg, deliveryMode,priority,timeToLive); - } - - public void setDeliveryMode(int deliveryMode) throws JMSException { - checkPreConditions(); - _delegate.setDeliveryMode(deliveryMode); - } - - public void setDisableMessageID(boolean disableMessageID) throws JMSException { - checkPreConditions(); - _delegate.setDisableMessageID(disableMessageID); - } - - public void setDisableMessageTimestamp(boolean disableMessageTimestamp) throws JMSException { - checkPreConditions(); - _delegate.setDisableMessageTimestamp(disableMessageTimestamp); - } - - public void setPriority(int priority) throws JMSException { - checkPreConditions(); - _delegate.setPriority(priority); - } - - public void setTimeToLive(long timeToLive) throws JMSException { - checkPreConditions(); - _delegate.setTimeToLive(timeToLive); - } +public class QueueSenderAdapter implements QueueSender +{ + + private BasicMessageProducer _delegate; + private Queue _queue; + private boolean closed = false; + + public QueueSenderAdapter(BasicMessageProducer msgProducer, Queue queue) + { + _delegate = msgProducer; + _queue = queue; + } + + public Queue getQueue() throws JMSException + { + checkPreConditions(); + return _queue; + } + + public void send(Message msg) throws JMSException + { + checkPreConditions(); + _delegate.send(msg); + } + + public void send(Queue queue, Message msg) throws JMSException + { + checkPreConditions(queue); + _delegate.send(queue, msg); + } + + public void publish(Message msg, int deliveryMode, int priority, long timeToLive) + throws JMSException + { + checkPreConditions(); + _delegate.send(msg, deliveryMode, priority, timeToLive); + } + + public void send(Queue queue, Message msg, int deliveryMode, int priority, long timeToLive) + throws JMSException + { + checkPreConditions(queue); + _delegate.send(queue, msg, deliveryMode, priority, timeToLive); + } + + public void close() throws JMSException + { + _delegate.close(); + closed = true; + } + + public int getDeliveryMode() throws JMSException + { + checkPreConditions(); + return _delegate.getDeliveryMode(); + } + + public Destination getDestination() throws JMSException + { + checkPreConditions(); + return _delegate.getDestination(); + } + + public boolean getDisableMessageID() throws JMSException + { + checkPreConditions(); + return _delegate.getDisableMessageID(); + } + + public boolean getDisableMessageTimestamp() throws JMSException + { + checkPreConditions(); + return _delegate.getDisableMessageTimestamp(); + } + + public int getPriority() throws JMSException + { + checkPreConditions(); + return _delegate.getPriority(); + } + + public long getTimeToLive() throws JMSException + { + checkPreConditions(); + return _delegate.getTimeToLive(); + } + + public void send(Destination dest, Message msg) throws JMSException + { + checkPreConditions((Queue) dest); + _delegate.send(dest, msg); + } + + public void send(Message msg, int deliveryMode, int priority, long timeToLive) + throws JMSException + { + checkPreConditions(); + _delegate.send(msg, deliveryMode, priority, timeToLive); + } + + public void send(Destination dest, Message msg, int deliveryMode, int priority, long timeToLive) throws JMSException + { + checkPreConditions((Queue) dest); + _delegate.send(dest, msg, deliveryMode, priority, timeToLive); + } + + public void setDeliveryMode(int deliveryMode) throws JMSException + { + checkPreConditions(); + _delegate.setDeliveryMode(deliveryMode); + } + + public void setDisableMessageID(boolean disableMessageID) throws JMSException + { + checkPreConditions(); + _delegate.setDisableMessageID(disableMessageID); + } + + public void setDisableMessageTimestamp(boolean disableMessageTimestamp) throws JMSException + { + checkPreConditions(); + _delegate.setDisableMessageTimestamp(disableMessageTimestamp); + } + + public void setPriority(int priority) throws JMSException + { + checkPreConditions(); + _delegate.setPriority(priority); + } + + public void setTimeToLive(long timeToLive) throws JMSException + { + checkPreConditions(); + _delegate.setTimeToLive(timeToLive); + } private void checkPreConditions() throws JMSException { @@ -130,31 +152,41 @@ public class QueueSenderAdapter implements QueueSender { private void checkPreConditions(Queue queue) throws JMSException { - if (closed){ - throw new javax.jms.IllegalStateException("Publisher is closed"); - } - - AMQSession session = ((BasicMessageProducer) _delegate).getSession(); - - if(session == null || session.isClosed()){ - throw new javax.jms.IllegalStateException("Invalid Session"); - } - - if(!(queue instanceof AMQDestination)) + if (closed) + { + throw new javax.jms.IllegalStateException("Publisher is closed"); + } + + AMQSession session = ((BasicMessageProducer) _delegate).getSession(); + + if (session == null || session.isClosed()) + { + throw new javax.jms.IllegalStateException("Invalid Session"); + } + + if (!(queue instanceof AMQDestination)) { throw new InvalidDestinationException("Queue: " + queue + " is not a valid Qpid queue"); } AMQDestination destination = (AMQDestination) queue; - if(!destination.isValidated() && checkQueueBeforePublish()) + if (!destination.isValidated() && checkQueueBeforePublish()) { - if (_delegate.isBound(destination)) + if (_delegate.getSession().isStrictAMQP()) { + _delegate._logger.warn("AMQP does not support destination validation before publish, "); destination.setValidated(true); } else { - throw new InvalidDestinationException("Queue: " + queue + " is not a valid destination (no bindings on server"); + if (_delegate.isBound(destination)) + { + destination.setValidated(true); + } + else + { + throw new InvalidDestinationException("Queue: " + queue + " is not a valid destination (no bindings on server"); + } } } } diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java index 5bc1555df7..addef94215 100644 --- a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java @@ -490,27 +490,7 @@ public class AMQProtocolHandler extends IoHandlerAdapter new SpecificMethodFrameListener(frame.getChannel(), responseClass), timeout); } - /** - * Convenience method to register an AMQSession with the protocol handler. Registering a session with the protocol - * handler will ensure that messages are delivered to the consumer(s) on that session. - * - * @param channelId the channel id of the session - * @param session the session instance. - */ - public void addSessionByChannel(int channelId, AMQSession session) - { - _protocolSession.addSessionByChannel(channelId, session); - } - /** - * Convenience method to deregister an AMQSession with the protocol handler. - * - * @param channelId then channel id of the session - */ - public void removeSessionByChannel(int channelId) - { - _protocolSession.removeSessionByChannel(channelId); - } public void closeSession(AMQSession session) throws AMQException { diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java index 35aa69bd82..a557fc8027 100644 --- a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java +++ b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java @@ -85,7 +85,7 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession protected final AMQProtocolHandler _protocolHandler; /** Maps from the channel id to the AMQSession that it represents. */ - protected ConcurrentMap _channelId2SessionMap = new ConcurrentHashMap(); + protected ConcurrentMap<Integer, AMQSession> _channelId2SessionMap = new ConcurrentHashMap<Integer, AMQSession>(); protected ConcurrentMap _closingChannels = new ConcurrentHashMap(); @@ -104,26 +104,13 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession private VersionSpecificRegistry _registry = MainRegistry.getVersionSpecificRegistry(ProtocolVersion.getLatestSupportedVersion()); - /** - * No-arg constructor for use by test subclass - has to initialise final vars NOT intended for use other then for - * test - */ - public AMQProtocolSession() - { - _protocolHandler = null; - _minaProtocolSession = null; - _stateManager = new AMQStateManager(this); - } + private final AMQConnection _connection; + public AMQProtocolSession(AMQProtocolHandler protocolHandler, IoSession protocolSession, AMQConnection connection) { - _protocolHandler = protocolHandler; - _minaProtocolSession = protocolSession; - // properties of the connection are made available to the event handlers - _minaProtocolSession.setAttribute(AMQ_CONNECTION, connection); - // fixme - real value needed - _minaProtocolSession.setWriteTimeout(LAST_WRITE_FUTURE_JOIN_TIMEOUT); - _stateManager = new AMQStateManager(this); + this(protocolHandler,protocolSession,connection, new AMQStateManager()); + } public AMQProtocolSession(AMQProtocolHandler protocolHandler, IoSession protocolSession, AMQConnection connection, @@ -138,6 +125,7 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession _minaProtocolSession.setWriteTimeout(LAST_WRITE_FUTURE_JOIN_TIMEOUT); _stateManager = stateManager; _stateManager.setProtocolSession(this); + _connection = connection; } @@ -305,11 +293,16 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession */ private void deliverMessageToAMQSession(int channelId, UnprocessedMessage msg) { - AMQSession session = (AMQSession) _channelId2SessionMap.get(channelId); + AMQSession session = getSession(channelId); session.messageReceived(msg); _channelId2UnprocessedMsgMap.remove(channelId); } + protected AMQSession getSession(int channelId) + { + return _connection.getSession(channelId); + } + /** * Convenience method that writes a frame to the protocol session. Equivalent to calling * getProtocolSession().write(). @@ -335,32 +328,6 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession } } - public void addSessionByChannel(int channelId, AMQSession session) - { - if (channelId <= 0) - { - throw new IllegalArgumentException("Attempt to register a session with a channel id <= zero"); - } - - if (session == null) - { - throw new IllegalArgumentException("Attempt to register a null session"); - } - - _logger.debug("Add session with channel id " + channelId); - _channelId2SessionMap.put(channelId, session); - } - - public void removeSessionByChannel(int channelId) - { - if (channelId <= 0) - { - throw new IllegalArgumentException("Attempt to deregister a session with a channel id <= zero"); - } - - _logger.debug("Removing session with channelId " + channelId); - _channelId2SessionMap.remove(channelId); - } /** * Starts the process of closing a session @@ -393,11 +360,11 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession */ public boolean channelClosed(int channelId, AMQConstant code, String text) throws AMQException { - final Integer chId = channelId; + // if this is not a response to an earlier request to close the channel - if (_closingChannels.remove(chId) == null) + if (_closingChannels.remove(channelId) == null) { - final AMQSession session = (AMQSession) _channelId2SessionMap.get(chId); + final AMQSession session = getSession(channelId); try { session.closed(new AMQException(code, text)); @@ -469,8 +436,7 @@ public class AMQProtocolSession implements AMQVersionAwareProtocolSession public void confirmConsumerCancelled(int channelId, AMQShortString consumerTag) { - final Integer chId = channelId; - final AMQSession session = (AMQSession) _channelId2SessionMap.get(chId); + final AMQSession session = getSession(channelId); session.confirmConsumerCancelled(consumerTag); } diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java index 4374329fb0..3776ff767f 100644 --- a/java/client/src/test/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java +++ b/java/client/src/test/java/org/apache/qpid/test/unit/client/protocol/AMQProtocolSessionTest.java @@ -32,9 +32,6 @@ public class AMQProtocolSessionTest extends TestCase { private static class AMQProtSession extends AMQProtocolSession { - public AMQProtSession() - { - } public AMQProtSession(AMQProtocolHandler protocolHandler, IoSession protocolSession, AMQConnection connection) { diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java index 685fe20048..8aaa760537 100644 --- a/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java +++ b/java/client/src/test/java/org/apache/qpid/test/unit/transacted/CommitRollbackTest.java @@ -390,7 +390,6 @@ public class CommitRollbackTest extends TestCase assertEquals("1", ((TextMessage) result).getText()); assertTrue("Messasge is marked as redelivered" + result, !result.getJMSRedelivered()); - _logger.info("Closing Consumer"); _consumer.close(); @@ -398,31 +397,32 @@ public class CommitRollbackTest extends TestCase _consumer = _session.createConsumer(_jmsQueue); _logger.info("receiving result"); + + // NOTE: Both msg 1 & 2 will be marked as redelivered as they have both will have been rejected. + // Only the occasion where it is not rejected will it mean it hasn't arrived at the client yet. result = _consumer.receive(1000); assertNotNull("test message was consumed and rolled back, but is gone", result); + + // The first message back will be either 1 or 2 being redelivered if (result.getJMSRedelivered()) { - assertEquals("1", ((TextMessage) result).getText()); - - result = _consumer.receive(1000); - assertNotNull("test message was consumed and rolled back, but is gone", result); - assertEquals("2", ((TextMessage) result).getText()); assertTrue("Messasge is not marked as redelivered" + result, result.getJMSRedelivered()); } - else + else // or it will be msg 2 arriving the first time due to latency. { - assertEquals("2", ((TextMessage) result).getText()); - assertTrue("Messasge is marked as redelivered" + result, !result.getJMSRedelivered()); + _logger.info("Message 2 wasn't prefetched so wasn't rejected"); + assertEquals("2", ((TextMessage) result).getText()); + } - result = _consumer.receive(1000); - assertNotNull("test message was consumed and rolled back, but is gone", result); - assertEquals("1", ((TextMessage) result).getText()); - assertTrue("Messasge is not marked as redelivered" + result, result.getJMSRedelivered()); + result = _consumer.receive(1000); + assertNotNull("test message was consumed and rolled back, but is gone", result); + assertTrue("Messasge is not marked as redelivered" + result, result.getJMSRedelivered()); - } result = _consumer.receive(1000); assertNull("test message should be null:" + result, result); + _session.commit(); + } |