Added AMQQueueSessionAdaptor and AMQTopicSessionAdaptor and amended AMQConnection create methods to use these adaptor classes as per QPID-66.

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

3 files changed, 360 insertions, 2 deletions

diff --git a/client/src/org/apache/qpid/client/AMQConnection.java b/client/src/org/apache/qpid/client/AMQConnection.java
index 1f72484993..368d7d9b0c 100644
--- a/client/src/org/apache/qpid/client/AMQConnection.java
+++ b/client/src/org/apache/qpid/client/AMQConnection.java
@@ -494,14 +494,30 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         return _failoverPolicy;
     }
 
+    /**
+     * Returns an AMQQueueSessionAdaptor which wraps an AMQSession and throws IllegalStateExceptions
+     * where specified in the JMS spec
+     * @param transacted
+     * @param acknowledgeMode
+     * @return QueueSession
+     * @throws JMSException
+     */
     public QueueSession createQueueSession(boolean transacted, int acknowledgeMode) throws JMSException
     {
-        return (QueueSession) createSession(transacted, acknowledgeMode);
+        return new AMQQueueSessionAdaptor(createSession(transacted, acknowledgeMode));
     }
 
+    /**
+     * Returns an AMQTopicSessionAdapter which wraps an AMQSession and throws IllegalStateExceptions
+     * where specified in the JMS spec
+     * @param transacted
+     * @param acknowledgeMode
+     * @return TopicSession
+     * @throws JMSException
+     */
     public TopicSession createTopicSession(boolean transacted, int acknowledgeMode) throws JMSException
     {
-        return (TopicSession) createSession(transacted, acknowledgeMode);
+        return new AMQTopicSessionAdaptor(createSession(transacted, acknowledgeMode));
     }
 
     private boolean channelLimitReached()
diff --git a/client/src/org/apache/qpid/client/AMQQueueSessionAdaptor.java b/client/src/org/apache/qpid/client/AMQQueueSessionAdaptor.java
new file mode 100644
index 0000000000..759b4405eb
--- /dev/null
+++ b/client/src/org/apache/qpid/client/AMQQueueSessionAdaptor.java
@@ -0,0 +1,176 @@
+/*

+ *

+ * Copyright (c) 2006 The Apache Software Foundation

+ *

+ * Licensed 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.client;

+

+import javax.jms.*;

+import javax.jms.IllegalStateException;

+import java.io.Serializable;

+

+/**

+ * Need this adaptor class to conform to JMS spec and throw IllegalStateException

+ * from createDurableSubscriber, unsubscribe, createTopic & createTemporaryTopic

+ */

+public class AMQQueueSessionAdaptor implements QueueSession

+{

+    //holds a session for delegation

+    protected final AMQSession _session;

+

+    /**

+     * Construct an adaptor with a session to wrap

+     * @param session

+     */

+    public AMQQueueSessionAdaptor(Session session)

+    {

+        _session = (AMQSession) session;

+    }

+

+    public TemporaryQueue createTemporaryQueue() throws JMSException {

+        return _session.createTemporaryQueue();

+    }

+

+    public Queue createQueue(String string) throws JMSException {

+        return _session.createQueue(string);

+    }

+

+    public QueueReceiver createReceiver(Queue queue) throws JMSException {

+        return _session.createReceiver(queue);

+    }

+

+    public QueueReceiver createReceiver(Queue queue, String string) throws JMSException {

+        return _session.createReceiver(queue, string);

+    }

+

+    public QueueSender createSender(Queue queue) throws JMSException {

+        return _session.createSender(queue);

+    }

+

+    public QueueBrowser createBrowser(Queue queue) throws JMSException {

+        return _session.createBrowser(queue);

+    }

+

+    public QueueBrowser createBrowser(Queue queue, String string) throws JMSException {

+        return _session.createBrowser(queue, string);

+    }

+

+    public BytesMessage createBytesMessage() throws JMSException {

+        return _session.createBytesMessage();

+    }

+

+    public MapMessage createMapMessage() throws JMSException {

+        return _session.createMapMessage();

+    }

+

+    public Message createMessage() throws JMSException {

+        return _session.createMessage();

+    }

+

+    public ObjectMessage createObjectMessage() throws JMSException {

+        return _session.createObjectMessage();

+    }

+

+    public ObjectMessage createObjectMessage(Serializable serializable) throws JMSException {

+        return _session.createObjectMessage(serializable);

+    }

+

+    public StreamMessage createStreamMessage() throws JMSException {

+        return _session.createStreamMessage();

+    }

+

+    public TextMessage createTextMessage() throws JMSException {

+        return _session.createTextMessage();

+    }

+

+    public TextMessage createTextMessage(String string) throws JMSException {

+        return _session.createTextMessage(string);

+    }

+

+    public boolean getTransacted() throws JMSException {

+        return _session.getTransacted();

+    }

+

+    public int getAcknowledgeMode() throws JMSException {

+       return _session.getAcknowledgeMode();

+    }

+

+    public void commit() throws JMSException {

+        _session.commit();

+    }

+

+    public void rollback() throws JMSException {

+        _session.rollback();

+    }

+

+    public void close() throws JMSException {

+        _session.close();

+    }

+

+    public void recover() throws JMSException {

+        _session.recover();

+    }

+

+    public MessageListener getMessageListener() throws JMSException {

+       return _session.getMessageListener();

+    }

+

+    public void setMessageListener(MessageListener messageListener) throws JMSException {

+        _session.setMessageListener(messageListener);

+    }

+

+    public void run() {

+        _session.run();

+    }

+

+    public MessageProducer createProducer(Destination destination) throws JMSException {

+        return _session.createProducer(destination);

+    }

+

+    public MessageConsumer createConsumer(Destination destination) throws JMSException {

+        return _session.createConsumer(destination);

+    }

+

+    public MessageConsumer createConsumer(Destination destination, String string) throws JMSException {

+        return _session.createConsumer(destination,string);

+    }

+

+    public MessageConsumer createConsumer(Destination destination, String string, boolean b) throws JMSException {

+        return _session.createConsumer(destination,string,b);

+    }

+

+    //The following methods cannot be called from a QueueSession as per JMS spec

+

+    public Topic createTopic(String string) throws JMSException {

+        throw new IllegalStateException("Cannot call createTopic from QueueSession");

+    }

+

+    public TopicSubscriber createDurableSubscriber(Topic topic, String string) throws JMSException {

+        throw new IllegalStateException("Cannot call createDurableSubscriber from QueueSession");

+    }

+

+    public TopicSubscriber createDurableSubscriber(Topic topic, String string, String string1, boolean b) throws JMSException {

+         throw new IllegalStateException("Cannot call createDurableSubscriber from QueueSession");

+    }

+

+    public TemporaryTopic createTemporaryTopic() throws JMSException {

+        throw new IllegalStateException("Cannot call createTemporaryTopic from QueueSession");

+    }

+

+    public void unsubscribe(String string) throws JMSException {

+        throw new IllegalStateException("Cannot call unsubscribe from QueueSession");

+    }

+

+}

diff --git a/client/src/org/apache/qpid/client/AMQTopicSessionAdaptor.java b/client/src/org/apache/qpid/client/AMQTopicSessionAdaptor.java
new file mode 100644
index 0000000000..9cbf7924b3
--- /dev/null
+++ b/client/src/org/apache/qpid/client/AMQTopicSessionAdaptor.java
@@ -0,0 +1,166 @@
+/*

+ *

+ * Copyright (c) 2006 The Apache Software Foundation

+ *

+ * Licensed 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.client;

+

+import javax.jms.*;

+import javax.jms.IllegalStateException;

+import java.io.Serializable;

+

+public class AMQTopicSessionAdaptor implements TopicSession

+{

+    protected final AMQSession _session;

+

+    public AMQTopicSessionAdaptor(Session session)

+    {

+        _session = (AMQSession) session;

+    }

+

+    public Topic createTopic(String string) throws JMSException {

+        return _session.createTopic(string);

+    }

+

+    public TopicSubscriber createSubscriber(Topic topic) throws JMSException {

+        return _session.createSubscriber(topic);

+    }

+

+    public TopicSubscriber createSubscriber(Topic topic, String string, boolean b) throws JMSException {

+        return _session.createSubscriber(topic, string, b);

+    }

+

+    public TopicSubscriber createDurableSubscriber(Topic topic, String string) throws JMSException {

+        return _session.createDurableSubscriber(topic, string);

+    }

+

+    public TopicSubscriber createDurableSubscriber(Topic topic, String string, String string1, boolean b) throws JMSException {

+        return _session.createDurableSubscriber(topic, string, string1, b);

+    }

+

+    public TopicPublisher createPublisher(Topic topic) throws JMSException {

+        return _session.createPublisher(topic);

+    }

+

+    public TemporaryTopic createTemporaryTopic() throws JMSException {

+        return _session.createTemporaryTopic();

+    }

+

+    public void unsubscribe(String string) throws JMSException {

+        _session.unsubscribe(string);

+    }

+

+    public BytesMessage createBytesMessage() throws JMSException {

+        return _session.createBytesMessage();

+    }

+

+    public MapMessage createMapMessage() throws JMSException {

+        return _session.createMapMessage();

+    }

+

+    public Message createMessage() throws JMSException {

+        return _session.createMessage();

+    }

+

+    public ObjectMessage createObjectMessage() throws JMSException {

+        return _session.createObjectMessage();

+    }

+

+    public ObjectMessage createObjectMessage(Serializable serializable) throws JMSException {

+        return _session.createObjectMessage();

+    }

+

+    public StreamMessage createStreamMessage() throws JMSException {

+        return _session.createStreamMessage();

+    }

+

+    public TextMessage createTextMessage() throws JMSException {

+        return _session.createTextMessage();

+    }

+

+    public TextMessage createTextMessage(String string) throws JMSException {

+        return _session.createTextMessage();

+    }

+

+    public boolean getTransacted() throws JMSException {

+        return _session.getTransacted();

+    }

+

+    public int getAcknowledgeMode() throws JMSException {

+        return _session.getAcknowledgeMode();

+    }

+

+    public void commit() throws JMSException {

+        _session.commit();

+    }

+

+    public void rollback() throws JMSException {

+        _session.rollback();

+    }

+

+    public void close() throws JMSException {

+        _session.close();

+    }

+

+    public void recover() throws JMSException {

+        _session.recover();

+    }

+

+    public MessageListener getMessageListener() throws JMSException {

+        return _session.getMessageListener();

+    }

+

+    public void setMessageListener(MessageListener messageListener) throws JMSException {

+        _session.setMessageListener(messageListener);

+    }

+

+    public void run() {

+        _session.run();

+    }

+

+    public MessageProducer createProducer(Destination destination) throws JMSException {

+        return _session.createProducer(destination);

+    }

+

+    public MessageConsumer createConsumer(Destination destination) throws JMSException {

+        return _session.createConsumer(destination);

+    }

+

+    public MessageConsumer createConsumer(Destination destination, String string) throws JMSException {

+        return _session.createConsumer(destination, string);

+    }

+

+    public MessageConsumer createConsumer(Destination destination, String string, boolean b) throws JMSException {

+        return _session.createConsumer(destination, string, b);

+    }

+

+    //The following methods cannot be called from a TopicSession as per JMS spec

+    public Queue createQueue(String string) throws JMSException {

+        throw new IllegalStateException("Cannot call createQueue from TopicSession");

+    }

+

+    public QueueBrowser createBrowser(Queue queue) throws JMSException {

+        throw new IllegalStateException("Cannot call createBrowser from TopicSession");

+    }

+

+    public QueueBrowser createBrowser(Queue queue, String string) throws JMSException {

+        throw new IllegalStateException("Cannot call createBrowser from TopicSession");

+    }

+

+    public TemporaryQueue createTemporaryQueue() throws JMSException {

+        throw new IllegalStateException("Cannot call createTemporaryQueue from TopicSession");

+    }

+

+}