QPID-3386: move all server Subscription creation into the SubscriptionFactoryImpl, ensure all Subscription implementations share a common ID sequence generator

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

3 files changed, 88 insertions, 45 deletions

diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/SubscriptionFactoryImplTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/SubscriptionFactoryImplTest.java
new file mode 100644
index 0000000000..29f45bf7f4
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/SubscriptionFactoryImplTest.java
@@ -0,0 +1,84 @@
+/*
+ *
+ * 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.subscription;
+
+import org.apache.qpid.common.AMQPFilterTypes;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.flow.WindowCreditManager;
+import org.apache.qpid.server.protocol.ProtocolEngine_0_10;
+import org.apache.qpid.server.transport.ServerConnection;
+import org.apache.qpid.server.transport.ServerSession;
+import org.apache.qpid.server.transport.ServerSessionDelegate;
+import org.apache.qpid.server.util.InternalBrokerBaseCase;
+import org.apache.qpid.transport.Binary;
+import org.apache.qpid.transport.MessageAcceptMode;
+import org.apache.qpid.transport.MessageAcquireMode;
+import org.apache.qpid.transport.MessageFlowMode;
+import org.apache.qpid.transport.TestNetworkConnection;
+
+public class SubscriptionFactoryImplTest extends InternalBrokerBaseCase
+{
+    /**
+     * Tests that while creating Subscriptions of various types, the 
+     * ID numbers assigned are allocated from a common sequence
+     * (in increasing order).
+     */
+    public void testDifferingSubscriptionTypesShareCommonIdNumberingSequence() throws Exception
+    {
+        //create a No-Ack subscription, get the first Subscription ID
+        long previousId = 0;
+        Subscription noAckSub = SubscriptionFactoryImpl.INSTANCE.createSubscription(1, getSession(), new AMQShortString("1"), false, null, false, getChannel().getCreditManager());
+        previousId = noAckSub.getSubscriptionID();
+
+        //create an ack subscription, verify the next Subscription ID is used
+        Subscription ackSub = SubscriptionFactoryImpl.INSTANCE.createSubscription(1, getSession(), new AMQShortString("1"), true, null, false, getChannel().getCreditManager());
+        assertEquals("Unexpected Subscription ID allocated", previousId + 1, ackSub.getSubscriptionID());
+        previousId = ackSub.getSubscriptionID();
+
+        //create a browser subscription
+        FieldTable filters = new FieldTable();
+        filters.put(AMQPFilterTypes.NO_CONSUME.getValue(), true);
+        Subscription browerSub = SubscriptionFactoryImpl.INSTANCE.createSubscription(1, getSession(), new AMQShortString("1"), true, null, false, getChannel().getCreditManager());
+        assertEquals("Unexpected Subscription ID allocated", previousId + 1, browerSub.getSubscriptionID());
+        previousId = browerSub.getSubscriptionID();
+
+        //create an BasicGet NoAck subscription
+        Subscription getNoAckSub = SubscriptionFactoryImpl.INSTANCE.createBasicGetNoAckSubscription(getChannel(), getSession(), new AMQShortString("1"), null, false, 
+                getChannel().getCreditManager(),getChannel().getClientDeliveryMethod(), getChannel().getRecordDeliveryMethod());
+        assertEquals("Unexpected Subscription ID allocated", previousId + 1, getNoAckSub.getSubscriptionID());
+        previousId = getNoAckSub.getSubscriptionID();
+
+        //create a 0-10 subscription
+        ServerConnection conn = new ServerConnection(1);
+        ProtocolEngine_0_10 engine = new ProtocolEngine_0_10(conn, new TestNetworkConnection(), getRegistry());
+        conn.setVirtualHost(getVirtualHost());
+        conn.setConnectionConfig(engine);
+        ServerSessionDelegate sesDel = new ServerSessionDelegate();
+        Binary name = new Binary(new byte[]{new Byte("1")});
+        ServerSession session = new ServerSession(conn, sesDel, name, 0, engine);
+
+        Subscription sub_0_10 = SubscriptionFactoryImpl.INSTANCE.createSubscription(session, "1", MessageAcceptMode.EXPLICIT,
+                MessageAcquireMode.PRE_ACQUIRED, MessageFlowMode.WINDOW, new WindowCreditManager(), null, null);
+        assertEquals("Unexpected Subscription ID allocated", previousId + 1, sub_0_10.getSubscriptionID());
+    }
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
index ff94942457..a97134a58d 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
@@ -44,14 +44,13 @@ import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.TestableMemoryMessageStore;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.test.utils.QpidTestCase;
-import org.apache.qpid.util.MockChannel;
 
 
 public class InternalBrokerBaseCase extends QpidTestCase
 {
     private IApplicationRegistry _registry;
     private MessageStore _messageStore;
-    private MockChannel _channel;
+    private AMQChannel _channel;
     private InternalTestProtocolSession _session;
     private VirtualHost _virtualHost;
     private AMQQueue _queue;
@@ -111,7 +110,7 @@ public class InternalBrokerBaseCase extends QpidTestCase
         _session = new InternalTestProtocolSession(_virtualHost);
         CurrentActor.set(_session.getLogActor());
 
-        _channel = new MockChannel(_session, 1, _messageStore);
+        _channel = new AMQChannel(_session, 1, _messageStore);
 
         _session.addChannel(_channel);
     }
@@ -283,12 +282,12 @@ public class InternalBrokerBaseCase extends QpidTestCase
         _messageStore = messageStore;
     }
 
-    public MockChannel getChannel()
+    public AMQChannel getChannel()
     {
         return _channel;
     }
 
-    public void setChannel(MockChannel channel)
+    public void setChannel(AMQChannel channel)
     {
         _channel = channel;
     }
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/util/MockChannel.java b/qpid/java/broker/src/test/java/org/apache/qpid/util/MockChannel.java
deleted file mode 100644
index 9bd1e7c5e1..0000000000
--- a/qpid/java/broker/src/test/java/org/apache/qpid/util/MockChannel.java
+++ /dev/null
@@ -1,40 +0,0 @@
-/*
- *
- * 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.util;
-
-import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.subscription.Subscription;
-import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.protocol.AMQProtocolSession;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.AMQShortString;
-
-public class MockChannel extends AMQChannel
-{
-    public MockChannel(AMQProtocolSession session, int channelId, MessageStore messageStore)
-            throws AMQException
-    {
-        super(session, channelId, messageStore);
-    }
-
-
-
-}