diff options
| -rw-r--r-- | java/client/src/main/java/org/apache/qpid/client/TopicPublisherAdapter.java | 303 | ||||
| -rw-r--r-- | java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicPublisherTest.java | 75 |
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); + } +} |