QPID-2002 : SubscriptionLogging Tests, update to SubscriptionImpl for new exclusive flag on setQueue

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@801719 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 350 insertions, 20 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java b/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
index 51da884d1e..a959944434 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java
@@ -33,6 +33,10 @@ import org.apache.qpid.common.ClientProperties;
 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 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage
     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 abstract class SubscriptionImpl implements Subscription, FlowCreditManage
 
 
 
-    
+
     public SubscriptionImpl(AMQChannel channel , AMQProtocolSession protocolSession,
                             AMQShortString consumerTag, FieldTable arguments,
                             boolean noLocal, FlowCreditManager creditManager,
@@ -327,13 +331,51 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage
 
 
 
-    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 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage
         }
 
 
-        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 abstract class SubscriptionImpl implements Subscription, FlowCreditManage
 
     public AMQQueue getQueue()
     {
-        return _queue;        
+        return _queue;
     }
 
     public void restoreCredit(final QueueEntry queueEntry)
@@ -555,7 +585,7 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage
 
     public void creditStateChanged(boolean hasCredit)
     {
-        
+
         if(hasCredit)
         {
             if(_state.compareAndSet(State.SUSPENDED, State.ACTIVE))
diff --git a/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java b/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
new file mode 100644
index 0000000000..7283f7f0d0
--- /dev/null
+++ b/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
@@ -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);
+
+    }
+
+}
diff --git a/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java b/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java
index 4c73e70bc2..10e7fecc30 100644
--- a/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java
+++ b/java/systests/src/main/java/org/apache/qpid/test/utils/QpidTestCase.java
@@ -183,6 +183,7 @@ public class QpidTestCase extends TestCase
     // 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)
     {