QPID-176 Patch supplied by Rob Godfrey. Check for topic is now more sophisticated since simple null check is not adequate.

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

2 files changed, 244 insertions, 134 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/TopicPublisherAdapter.java b/java/client/src/main/java/org/apache/qpid/client/TopicPublisherAdapter.java
index 02da284b83..ea772d0fa1 100644
--- a/java/client/src/main/java/org/apache/qpid/client/TopicPublisherAdapter.java
+++ b/java/client/src/main/java/org/apache/qpid/client/TopicPublisherAdapter.java
@@ -1,138 +1,173 @@
 package org.apache.qpid.client;
 
-import javax.jms.Destination;
+import javax.jms.*;
 import javax.jms.IllegalStateException;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageProducer;
-import javax.jms.Topic;
-import javax.jms.TopicPublisher;
-
-public class TopicPublisherAdapter implements TopicPublisher {
-
-	private MessageProducer delegate;
-	private Topic topic;
-	private boolean closed = false;
-	
-	public TopicPublisherAdapter(MessageProducer msgProducer, Topic topic){
-		delegate = msgProducer;
-		this.topic = topic;
-	}
-	
-	public Topic getTopic() throws JMSException {
-		checkPreConditions();
-		return topic;
-	}
-
-	public void publish(Message msg) throws JMSException {
-		checkPreConditions();
-		delegate.send(msg);
-	}
-
-	public void publish(Topic topic, Message msg) throws JMSException {
-		checkPreConditions();
-		delegate.send(topic,msg);
-	}
-
-	public void publish(Message msg, int deliveryMode, int priority, long timeToLive)
-			throws JMSException {
-		checkPreConditions();
-		delegate.send(msg, deliveryMode,priority,timeToLive);
-	}
-
-	public void publish(Topic topic, Message msg, int deliveryMode, int priority, long timeToLive)
-			throws JMSException {
-		checkPreConditions();
-		delegate.send(topic,msg, deliveryMode,priority,timeToLive);
-	}
-
-	public void close() throws JMSException {
-		delegate.close();
-		closed = true;
-	}
-
-	public int getDeliveryMode() throws JMSException {
-		return delegate.getDeliveryMode();
-	}
-
-	public Destination getDestination() throws JMSException {
-		return delegate.getDestination();
-	}
-
-	public boolean getDisableMessageID() throws JMSException {
-		return delegate.getDisableMessageID();
-	}
-
-	public boolean getDisableMessageTimestamp() throws JMSException {
-		return delegate.getDisableMessageTimestamp();
-	}
-
-	public int getPriority() throws JMSException {
-		return delegate.getPriority();
-	}
-
-	public long getTimeToLive() throws JMSException {
-		return delegate.getTimeToLive();
-	}
-
-	public void send(Message msg) throws JMSException {
-		checkPreConditions();
-		delegate.send(msg);
-	}
-
-	public void send(Destination dest, Message msg) throws JMSException {
-		checkPreConditions();
-		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();
-		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 IllegalStateException, IllegalStateException {
-		if (closed){
-			throw new javax.jms.IllegalStateException("Publisher is closed");
-		}
-		
-		if(topic == null){
-			throw new UnsupportedOperationException("Topic is null");
-		}
-		
-		AMQSession session = ((BasicMessageProducer)delegate).getSession();
-		if(session == null || session.isClosed()){
-			throw new javax.jms.IllegalStateException("Invalid Session");
-		}
-	}
+
+public class TopicPublisherAdapter implements TopicPublisher
+{
+
+    private BasicMessageProducer _delegate;
+    private Topic _topic;
+
+    public TopicPublisherAdapter(BasicMessageProducer msgProducer, Topic topic)
+    {
+        _delegate = msgProducer;
+        _topic = topic;
+    }
+
+    public Topic getTopic() throws JMSException
+    {
+        checkPreConditions();
+        return _topic;
+    }
+
+    public void publish(Message msg) throws JMSException
+    {
+        checkPreConditions();
+        checkTopic(_topic);
+        _delegate.send(msg);
+    }
+
+    public void publish(Topic topic, Message msg) throws JMSException
+    {
+        checkPreConditions();
+        checkTopic(topic);
+        _delegate.send(topic, msg);
+    }
+
+    public void publish(Message msg, int deliveryMode, int priority, long timeToLive)
+            throws JMSException
+    {
+        checkPreConditions();
+        checkTopic(_topic);
+        _delegate.send(msg, deliveryMode, priority, timeToLive);
+    }
+
+    public void publish(Topic topic, Message msg, int deliveryMode, int priority, long timeToLive)
+            throws JMSException
+    {
+        checkPreConditions();
+        checkTopic(topic);
+        _delegate.send(topic, msg, deliveryMode, priority, timeToLive);
+    }
+
+    public void close() throws JMSException
+    {
+        _delegate.close();
+    }
+
+    public int getDeliveryMode() throws JMSException
+    {
+        return _delegate.getDeliveryMode();
+    }
+
+    public Destination getDestination() throws JMSException
+    {
+        return _delegate.getDestination();
+    }
+
+    public boolean getDisableMessageID() throws JMSException
+    {
+        return _delegate.getDisableMessageID();
+    }
+
+    public boolean getDisableMessageTimestamp() throws JMSException
+    {
+        return _delegate.getDisableMessageTimestamp();
+    }
+
+    public int getPriority() throws JMSException
+    {
+        return _delegate.getPriority();
+    }
+
+    public long getTimeToLive() throws JMSException
+    {
+        return _delegate.getTimeToLive();
+    }
+
+    public void send(Message msg) throws JMSException
+    {
+        checkPreConditions();
+        checkTopic(_topic);
+        _delegate.send(msg);
+    }
+
+    public void send(Destination dest, Message msg) throws JMSException
+    {
+        checkPreConditions();
+        checkTopic(_topic);
+        _delegate.send(dest, msg);
+    }
+
+    public void send(Message msg, int deliveryMode, int priority, long timeToLive)
+            throws JMSException
+    {
+        checkPreConditions();
+        checkTopic(_topic);
+        _delegate.send(msg, deliveryMode, priority, timeToLive);
+    }
+
+    public void send(Destination dest, Message msg, int deliveryMode, int priority, long timeToLive) throws JMSException
+    {
+        checkPreConditions();
+        checkTopic(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 IllegalStateException
+    {
+        if (_delegate.isClosed())
+        {
+            throw new javax.jms.IllegalStateException("Publisher is _closed");
+        }
+
+        AMQSession session = _delegate.getSession();
+        if (session == null || session.isClosed())
+        {
+            throw new javax.jms.IllegalStateException("Invalid Session");
+        }
+    }
+
+    private void checkTopic(Destination topic) throws InvalidDestinationException
+    {
+        if (topic == null)
+        {
+            throw new UnsupportedOperationException("Topic is null");
+        }
+        if (!(topic instanceof Topic))
+        {
+            throw new InvalidDestinationException("Destination " + topic + " is not a topic");
+        }
+    }
 }
diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicPublisherTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicPublisherTest.java
new file mode 100644
index 0000000000..4ffb3e8459
--- /dev/null
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicPublisherTest.java
@@ -0,0 +1,75 @@
+/*
+ *
+ * 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.test.unit.topic;
+
+import junit.framework.TestCase;
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.AMQTopic;
+import org.apache.qpid.client.transport.TransportConnection;
+
+import javax.jms.*;
+
+/**
+ * @author Apache Software Foundation
+ */
+public class TopicPublisherTest extends TestCase
+{
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        TransportConnection.createVMBroker(1);
+    }
+
+    protected void tearDown() throws Exception
+    {
+        super.tearDown();
+        TransportConnection.killAllVMBrokers();
+    }
+
+    public void testUnidentifiedProducer() throws Exception
+    {
+        AMQTopic topic = new AMQTopic("MyTopic");
+        AMQConnection con = new AMQConnection("vm://:1", "guest", "guest", "test", "/test");
+        TopicSession session1 = con.createTopicSession(false, AMQSession.NO_ACKNOWLEDGE);
+        TopicPublisher publisher = session1.createPublisher(null);
+        MessageConsumer consumer1 = session1.createConsumer(topic);
+        con.start();
+        publisher.publish(topic, session1.createTextMessage("Hello"));
+        TextMessage m = (TextMessage) consumer1.receive(2000);
+        assertNotNull(m);
+        try
+        {
+            publisher.publish(session1.createTextMessage("Goodbye"));
+            fail("Did not throw UnsupportedOperationException");
+        }
+        catch (UnsupportedOperationException e)
+        {
+            // PASS
+        }
+
+    }
+
+    public static junit.framework.Test suite()
+    {
+        return new junit.framework.TestSuite(TopicPublisherTest.class);
+    }
+}