1 /***
2 *
3 * Copyright 2004 Protique Ltd
4 *
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 *
17 **/
18 package org.codehaus.activemq;
19
20 import org.apache.commons.logging.Log;
21 import org.apache.commons.logging.LogFactory;
22 import org.codehaus.activemq.message.ActiveMQDestination;
23
24 import javax.jms.Connection;
25 import javax.jms.Destination;
26 import javax.jms.JMSException;
27 import javax.jms.Message;
28 import javax.jms.MessageConsumer;
29 import javax.jms.MessageListener;
30 import javax.jms.MessageProducer;
31 import javax.jms.Session;
32 import javax.jms.TextMessage;
33 import javax.jms.Topic;
34 import java.util.List;
35 import java.util.Vector;
36
37 /***
38 * @version $Revision: 1.2 $
39 */
40 public class JmsCreateConsumerInOnMessageTest extends TestSupport implements MessageListener {
41 private final Log log = LogFactory.getLog(getClass());
42
43 private Connection connection;
44 private Session publisherSession;
45 private Session consumerSession;
46 private MessageConsumer consumer;
47 private MessageConsumer testConsumer;
48 private MessageProducer producer;
49 private Topic topic;
50 private Object lock = new Object();
51
52
53 /***
54 * @throws Exception
55 */
56 public void testCreateConsumer() throws Exception{
57 Message msg = super.createMessage();
58 producer.send(msg);
59 if (testConsumer == null){
60 synchronized(lock){
61 lock.wait(3000);
62 }
63 }
64 assertTrue(testConsumer != null);
65 }
66
67 /***
68 * Use the asynchronous subscription mechanism
69 * @param message
70 */
71 public void onMessage(Message message) {
72 try {
73 testConsumer = consumerSession.createConsumer(topic);
74 MessageProducer anotherProducer = consumerSession.createProducer(topic);
75 synchronized(lock){
76 lock.notify();
77 }
78 }catch (Exception ex){
79 ex.printStackTrace();
80 assertTrue(false);
81 }
82 }
83
84
85
86
87 protected void setUp() throws Exception {
88 super.setUp();
89 super.topic = true;
90 connection = createConnection();
91 connection.setClientID("connection:" + getSubject());
92 publisherSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
93 consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
94 topic = (Topic)super.createDestination("Test.Topic");
95 consumer = consumerSession.createConsumer(topic);
96 consumer.setMessageListener(this);
97 producer = publisherSession.createProducer(topic);
98 connection.start();
99 }
100
101 protected void tearDown() throws Exception {
102 super.tearDown();
103 connection.close();
104 }
105 }