    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 }