You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ri...@apache.org on 2009/08/06 18:57:12 UTC
svn commit: r801719 - in /qpid/trunk/qpid/java:
broker/src/main/java/org/apache/qpid/server/subscription/
systests/src/main/java/org/apache/qpid/server/logging/
systests/src/main/java/org/apache/qpid/test/utils/
Author: ritchiem
Date: Thu Aug 6 16:57:12 2009
New Revision: 801719
URL: http://svn.apache.org/viewvc?rev=801719&view=rev
Log:
QPID-2002 : SubscriptionLogging Tests, update to SubscriptionImpl for new exclusive flag on setQueue
Added:
qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
Modified:
qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java
Modified: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java?rev=801719&r1=801718&r2=801719&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java Thu Aug 6 16:57:12 2009
@@ -33,6 +33,10 @@
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.logging.messages.SubscriptionMessages;
+import org.apache.qpid.server.logging.subjects.SubscriptionLogSubject;
+import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.subscription.Subscription;
@@ -63,14 +67,14 @@
private final AtomicReference<QueueEntry> _queueContext = new AtomicReference<QueueEntry>(null);
private final ClientDeliveryMethod _deliveryMethod;
private final RecordDeliveryMethod _recordMethod;
-
+
private QueueEntry.SubscriptionAcquiredState _owningState = new QueueEntry.SubscriptionAcquiredState(this);
private final Lock _stateChangeLock;
private static final AtomicLong idGenerator = new AtomicLong(0);
// Create a simple ID that increments for ever new Subscription
private final long _subscriptionID = idGenerator.getAndIncrement();
-
+ private LogSubject _logSubject;
static final class BrowserSubscription extends SubscriptionImpl
{
@@ -278,7 +282,7 @@
-
+
public SubscriptionImpl(AMQChannel channel , AMQProtocolSession protocolSession,
AMQShortString consumerTag, FieldTable arguments,
boolean noLocal, FlowCreditManager creditManager,
@@ -327,13 +331,51 @@
- public synchronized void setQueue(AMQQueue queue)
+ public synchronized void setQueue(AMQQueue queue, boolean exclusive)
{
if(getQueue() != null)
{
throw new IllegalStateException("Attempt to set queue for subscription " + this + " to " + queue + "when already set to " + getQueue());
}
_queue = queue;
+
+ _logSubject = new SubscriptionLogSubject(this);
+
+ if (CurrentActor.get().getRootMessageLogger().
+ isMessageEnabled(CurrentActor.get(), _logSubject))
+ {
+ // Get the string value of the filters
+ String filterLogString = null;
+ if (_filters != null && _filters.hasFilters())
+ {
+ filterLogString = _filters.toString();
+ }
+
+ if (isAutoClose())
+ {
+ if (filterLogString == null)
+ {
+ filterLogString = "";
+ }
+ else
+ {
+ filterLogString += ",";
+ }
+ filterLogString += "AutoClose";
+ }
+
+ if (isBrowser())
+ {
+ // We do not need to check for null here as all Browsers are AutoClose
+ filterLogString +=",Browser";
+ }
+
+ CurrentActor.get().
+ message(_logSubject,
+ SubscriptionMessages.SUB_1001(filterLogString,
+ queue.isDurable() && exclusive,
+ filterLogString != null));
+ }
}
public String toString()
@@ -480,20 +522,8 @@
}
- if (closed)
- {
- if (_logger.isDebugEnabled())
- {
- _logger.debug("Called close() on a closed subscription");
- }
-
- return;
- }
-
- if (_logger.isInfoEnabled())
- {
- _logger.info("Closing subscription (" + debugIdentity() + "):" + this);
- }
+ //Log Subscription closed
+ CurrentActor.get().message(_logSubject, SubscriptionMessages.SUB_1002());
}
public boolean isClosed()
@@ -544,7 +574,7 @@
public AMQQueue getQueue()
{
- return _queue;
+ return _queue;
}
public void restoreCredit(final QueueEntry queueEntry)
@@ -555,7 +585,7 @@
public void creditStateChanged(boolean hasCredit)
{
-
+
if(hasCredit)
{
if(_state.compareAndSet(State.SUSPENDED, State.ACTIVE))
Added: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java?rev=801719&view=auto
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java (added)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java Thu Aug 6 16:57:12 2009
@@ -0,0 +1,299 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.logging;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.Topic;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Subscription
+ *
+ * The Subscription test suite validates that the follow log messages as specified in the Functional Specification.
+ *
+ * This suite of tests validate that the Subscription messages occur correctly and according to the following format:
+ *
+ * SUB-1001 : Create : [Durable] [Arguments : <key=value>]
+ * SUB-1002 : Close
+ */
+public class SubscriptionLoggingTest extends AbstractTestLogging
+{
+ static final String SUB_PREFIX = "SUB-";
+
+ Connection _connection;
+ Session _session;
+ Queue _queue;
+ Topic _topic;
+
+ @Override
+ public void setUp() throws Exception
+ {
+ super.setUp();
+
+ _connection = getConnection();
+
+ _session = _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ _queue = (Queue) getInitialContext().lookup(QUEUE);
+ _topic = (Topic) getInitialContext().lookup(TOPIC);
+
+ _monitor.reset();
+ }
+
+ /**
+ * Description:
+ * When a Subscription is created it will be logged. This test validates that Subscribing to a transient queue is correctly logged.
+ * Input:
+ *
+ * 1. Running Broker
+ * 2. Create a new Subscription to a transient queue/topic.
+ * Output:
+ *
+ * <date> SUB-1001 : Create
+ *
+ * Validation Steps:
+ * 3. The SUB ID is correct
+ *
+ * @throws java.io.IOException - if there is a problem getting the matches
+ * @throws javax.jms.JMSException - if there is a problem creating the consumer
+ */
+ public void testSubscriptionCreate() throws JMSException, IOException
+ {
+ _session.createConsumer(_queue);
+
+ //Validate
+
+ List<String> results = _monitor.findMatches(SUB_PREFIX);
+
+ assertEquals("Result set larger than expected.", 1, results.size());
+
+ String log = getLog(results.get(0));
+
+ validateMessageID("SUB-1001", log);
+
+ assertEquals("Log Message not as expected", "Create", getMessageString(fromMessage(log)));
+ }
+
+ /**
+ * Description:
+ * The creation of a Durable Subscription, such as a JMS DurableTopicSubscriber will result in an extra Durable tag being included in the Create log message
+ * Input:
+ *
+ * 1. Running Broker
+ * 2. Creation of a JMS DurableTopicSubiber
+ * Output:
+ *
+ * <date> SUB-1001 : Create : Durable
+ *
+ * Validation Steps:
+ * 3. The SUB ID is correct
+ * 4. The Durable tag is present in the message
+ * NOTE: A Subscription is not Durable, the queue it consumes from is.
+ *
+ * @throws java.io.IOException - if there is a problem getting the matches
+ * @throws javax.jms.JMSException - if there is a problem creating the consumer
+ */
+ public void testSubscriptionCreateDurable() throws JMSException, IOException
+ {
+ _session.createDurableSubscriber(_topic, getName());
+
+ //Validate
+
+ List<String> results = _monitor.findMatches(SUB_PREFIX);
+
+ assertEquals("Result set larger than expected.", 1, results.size());
+
+ String log = getLog(results.get(0));
+
+ validateMessageID("SUB-1001", log);
+
+ String message = getMessageString(fromMessage(log));
+ assertTrue("Durable not on log message:" + message, message.contains("Durable"));
+ }
+
+ /**
+ * Description:
+ * The creation of a QueueBrowser will provides a number arguments and so should form part of the SUB-1001 Create message.
+ * Input:
+ *
+ * 1. Running Broker
+ * 2. Java Client creates a QueueBroweser
+ * Output:
+ *
+ * <date> SUB-1001 : Create : Arguments : <key=value>
+ *
+ * Validation Steps:
+ * 3. The SUB ID is correct
+ * 4. The Arguments are present in the message
+ * 5. Arguments keys include AutoClose and Browser.
+ *
+ * @throws java.io.IOException - if there is a problem getting the matches
+ * @throws javax.jms.JMSException - if there is a problem creating the consumer
+ */
+ public void testSubscriptionCreateQueueBrowser() throws JMSException, IOException
+ {
+ _session.createBrowser(_queue);
+
+ //Validate
+ List<String> results = _monitor.findMatches(SUB_PREFIX);
+
+ assertEquals("Result set larger than expected.", 2, results.size());
+
+ String log = getLog(results.get(0));
+
+ validateMessageID("SUB-1001", log);
+
+ String message = getMessageString(fromMessage(log));
+ assertTrue("Browser not on log message:" + message, message.contains("Browser"));
+ assertTrue("AutoClose not on log message:" + message, message.contains("AutoClose"));
+
+ // Beacause it is an auto close and we have no messages on the queue we
+ // will get a close message
+ log = getLog(results.get(1));
+ validateMessageID("SUB-1002", log);
+
+ }
+
+ /**
+ * Description:
+ * The creation of a Subscriber with a JMS Selector will result in the Argument field being populated. These argument key/value pairs are then shown in the log message.
+ * Input:
+ *
+ * 1. Running Broker
+ * 2. Subscriber created with a JMS Selector.
+ * Output:
+ *
+ * <date> SUB-1001 : Create : Arguments : <key=value>
+ *
+ * Validation Steps:
+ * 3. The SUB ID is correct
+ * 4. Argument tag is present in the message
+ *
+ * @throws java.io.IOException - if there is a problem getting the matches
+ * @throws javax.jms.JMSException - if there is a problem creating the consumer
+ */
+ public void testSubscriptionCreateWithArguments() throws JMSException, IOException
+ {
+ final String SELECTOR = "Selector='True'";
+ _session.createConsumer(_queue, SELECTOR);
+
+ //Validate
+
+ List<String> results = _monitor.findMatches(SUB_PREFIX);
+
+ assertEquals("Result set larger than expected.", 1, results.size());
+
+ String log = getLog(results.get(0));
+
+ validateMessageID("SUB-1001", log);
+
+ String message = getMessageString(fromMessage(log));
+ assertTrue("Selector not on log message:" + message, message.contains(SELECTOR));
+ }
+
+ /**
+ * Description:
+ * The final combination of SUB-1001 Create messages involves the creation of a Durable Subscription that also contains a set of Arguments, such as those provided via a JMS Selector.
+ * Input:
+ *
+ * 1. Running Broker
+ * 2. Java Client creates a Durable Subscription with Selector
+ * Output:
+ *
+ * <date> SUB-1001 : Create : Durable Arguments : <key=value>
+ *
+ * Validation Steps:
+ * 3. The SUB ID is correct
+ * 4. The tag Durable is present in the message
+ * 5. The Arguments are present in the message
+ *
+ * @throws java.io.IOException - if there is a problem getting the matches
+ * @throws javax.jms.JMSException - if there is a problem creating the consumer
+ */
+ public void testSubscriptionCreateDurableWithArguments() throws JMSException, IOException
+ {
+ final String SELECTOR = "Selector='True'";
+ _session.createDurableSubscriber(_topic, getName(), SELECTOR, false);
+
+ //Validate
+
+ List<String> results = _monitor.findMatches(SUB_PREFIX);
+
+ assertEquals("Result set larger than expected.", 1, results.size());
+
+ String log = getLog(results.get(0));
+
+ validateMessageID("SUB-1001", log);
+
+ String message = getMessageString(fromMessage(log));
+ assertTrue("Durable not on log message:" + message, message.contains("Durable"));
+ assertTrue("Selector not on log message:" + message, message.contains(SELECTOR));
+ }
+
+ /**
+ * Description:
+ * When a Subscription is closed it will log this so that it can be correlated with the Create.
+ * Input:
+ *
+ * 1. Running Broker
+ * 2. Client with a subscription.
+ * 3. The subscription is then closed.
+ * Output:
+ *
+ * <date> SUB-1002 : Close
+ *
+ * Validation Steps:
+ * 1. The SUB ID is correct
+ * 2. There must be a SUB-1001 Create message preceding this message
+ * 3. This must be the last message from the given Subscription
+ *
+ * @throws java.io.IOException - if there is a problem getting the matches
+ * @throws javax.jms.JMSException - if there is a problem creating the consumer
+ */
+ public void testSubscriptionClose() throws JMSException, IOException
+ {
+ _session.createConsumer(_queue).close();
+
+
+
+ //Validate
+ List<String> results = _monitor.findMatches(SUB_PREFIX);
+
+ //3
+ assertEquals("Result set larger than expected.", 2, results.size());
+
+ // 2
+ String log = getLog(results.get(0));
+ validateMessageID("SUB-1001", log);
+ // 1
+ log = getLog(results.get(1));
+ validateMessageID("SUB-1002", log);
+
+ String message = getMessageString(fromMessage(log));
+ assertEquals("Log message is not close", "Close", message);
+
+ }
+
+}
Modified: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java?rev=801719&r1=801718&r2=801719&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java (original)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java Thu Aug 6 16:57:12 2009
@@ -183,6 +183,7 @@
// the connections created for a given test
protected List<Connection> _connections = new ArrayList<Connection>();
public static final String QUEUE = "queue";
+ public static final String TOPIC = "topic";
public QpidTestCase(String name)
{
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org