Java 类javax.jms.IllegalStateException 实例源码

@Test
public void testRuntimeExceptionFromSendMessage() throws JMSException {
    JMSProducer producer = context.createProducer();

    MockJMSConnection connection = (MockJMSConnection) context.getConnection();
    connection.addConnectionListener(new MockJMSDefaultConnectionListener() {

        @Override
        public void onMessageSend(MockJMSSession session, Message message) throws JMSException {
            throw new IllegalStateException("Send Failed");
        }
    });

    try {
        producer.send(context.createTemporaryQueue(), context.createMessage());
        fail("Should have thrown an exception");
    } catch (IllegalStateRuntimeException isre) {}
}

@Test
public void testRuntimeExceptionFromSendByteBody() throws JMSException {
    JMSProducer producer = context.createProducer();

    MockJMSConnection connection = (MockJMSConnection) context.getConnection();
    connection.addConnectionListener(new MockJMSDefaultConnectionListener() {

        @Override
        public void onMessageSend(MockJMSSession session, Message message) throws JMSException {
            throw new IllegalStateException("Send Failed");
        }
    });

    try {
        producer.send(context.createTemporaryQueue(), new byte[0]);
        fail("Should have thrown an exception");
    } catch (IllegalStateRuntimeException isre) {}
}

@Test
public void testRuntimeExceptionFromSendMapBody() throws JMSException {
    JMSProducer producer = context.createProducer();

    MockJMSConnection connection = (MockJMSConnection) context.getConnection();
    connection.addConnectionListener(new MockJMSDefaultConnectionListener() {

        @Override
        public void onMessageSend(MockJMSSession session, Message message) throws JMSException {
            throw new IllegalStateException("Send Failed");
        }
    });

    try {
        producer.send(context.createTemporaryQueue(), Collections.<String, Object>emptyMap());
        fail("Should have thrown an exception");
    } catch (IllegalStateRuntimeException isre) {}
}

@Test
public void testRuntimeExceptionFromSendSerializableBody() throws JMSException {
    JMSProducer producer = context.createProducer();

    MockJMSConnection connection = (MockJMSConnection) context.getConnection();
    connection.addConnectionListener(new MockJMSDefaultConnectionListener() {

        @Override
        public void onMessageSend(MockJMSSession session, Message message) throws JMSException {
            throw new IllegalStateException("Send Failed");
        }
    });

    try {
        producer.send(context.createTemporaryQueue(), UUID.randomUUID());
        fail("Should have thrown an exception");
    } catch (IllegalStateRuntimeException isre) {}
}

@Test
public void testRuntimeExceptionFromSendStringBody() throws JMSException {
    JMSProducer producer = context.createProducer();

    MockJMSConnection connection = (MockJMSConnection) context.getConnection();
    connection.addConnectionListener(new MockJMSDefaultConnectionListener() {

        @Override
        public void onMessageSend(MockJMSSession session, Message message) throws JMSException {
            throw new IllegalStateException("Send Failed");
        }
    });

    try {
        producer.send(context.createTemporaryQueue(), "test");
        fail("Should have thrown an exception");
    } catch (IllegalStateRuntimeException isre) {}
}

@Override
public void setClientID(String clientID) throws JMSException {
    checkClosedOrFailed();

    if (explicitClientID) {
        throw new IllegalStateException("The clientID has already been set");
    }
    if (clientID == null || clientID.isEmpty()) {
        throw new InvalidClientIDException("Cannot have a null or empty clientID");
    }
    if (connected.get()) {
        throw new IllegalStateException("Cannot set the client id once connected.");
    }

    setClientID(clientID, true);

    // We weren't connected if we got this far, we should now connect to ensure the
    // configured clientID is valid.
    initialize();
}

void deleteTemporaryDestination(MockJMSTemporaryDestination destination) throws JMSException {
    checkClosedOrFailed();

    try {
        for (MockJMSSession session : sessions.values()) {
            if (session.isDestinationInUse(destination)) {
                throw new IllegalStateException("A consumer is consuming from the temporary destination");
            }
        }

        signalDeleteTemporaryDestination(destination);
        stats.temporaryDestinationDestroyed(destination);
        tempDestinations.remove(destination);
    } catch (Exception e) {
        throw JMSExceptionSupport.create(e);
    }
}

@Test
public void testGetQueue() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueReceiver receiver = session.createReceiver(queue);

    assertNotNull(receiver.getQueue());
    assertSame(queue, receiver.getQueue());

    receiver.close();

    try {
        receiver.getQueue();
        fail("Cannot read topic on closed receiver");
    } catch (IllegalStateException ise) {}
}

@Test
public void testGetTopicSubscriber() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    JmsPoolQueueReceiver receiver = (JmsPoolQueueReceiver) session.createReceiver(queue);

    assertNotNull(receiver.getQueueReceiver());
    assertTrue(receiver.getQueueReceiver() instanceof MockJMSQueueReceiver);

    receiver.close();

    try {
        receiver.getQueueReceiver();
        fail("Cannot read state on closed receiver");
    } catch (IllegalStateException ise) {}
}

@Test
public void testReceive() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    Session session = connection.createSession();
    Queue queue = session.createTemporaryQueue();
    MessageConsumer consumer = session.createConsumer(queue, "Color = Red");

    assertNull(consumer.receive());

    consumer.close();

    try {
        consumer.receive();
        fail("Should not be able to interact with closed consumer");
    } catch (IllegalStateException ise) {}
}

@Test
public void testReceiveNoWait() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    Session session = connection.createSession();
    Queue queue = session.createTemporaryQueue();
    MessageConsumer consumer = session.createConsumer(queue, "Color = Red");

    assertNull(consumer.receiveNoWait());

    consumer.close();

    try {
        consumer.receiveNoWait();
        fail("Should not be able to interact with closed consumer");
    } catch (IllegalStateException ise) {}
}

@Test
public void testReceiveTimed() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    Session session = connection.createSession();
    Queue queue = session.createTemporaryQueue();
    MessageConsumer consumer = session.createConsumer(queue, "Color = Red");

    assertNull(consumer.receive(1));

    consumer.close();

    try {
        consumer.receive(1);
        fail("Should not be able to interact with closed consumer");
    } catch (IllegalStateException ise) {}
}

@Test
public void testGetMessageSelector() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    Session session = connection.createSession();
    Queue queue = session.createTemporaryQueue();
    MessageConsumer consumer = session.createConsumer(queue, "Color = Red");

    assertNotNull(consumer.getMessageSelector());
    assertEquals("Color = Red", consumer.getMessageSelector());

    consumer.close();

    try {
        consumer.getMessageSelector();
        fail("Should not be able to interact with closed consumer");
    } catch (IllegalStateException ise) {}
}

@Test
public void testGetTopic() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createTopicConnection();
    TopicSession session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTemporaryTopic();
    TopicSubscriber subscriber = session.createSubscriber(topic);

    assertNotNull(subscriber.getTopic());
    assertSame(topic, subscriber.getTopic());

    subscriber.close();

    try {
        subscriber.getTopic();
        fail("Cannot read topic on closed subscriber");
    } catch (IllegalStateException ise) {}
}

@Test
public void testGetNoLocal() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createTopicConnection();
    TopicSession session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTemporaryTopic();
    TopicSubscriber subscriber = session.createDurableSubscriber(topic, "name", "color = red", true);

    assertTrue(subscriber.getNoLocal());

    subscriber.close();

    try {
        subscriber.getNoLocal();
        fail("Cannot read state on closed subscriber");
    } catch (IllegalStateException ise) {}
}

@Test
public void testGetTopicSubscriber() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createTopicConnection();
    TopicSession session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTemporaryTopic();
    JmsPoolTopicSubscriber subscriber = (JmsPoolTopicSubscriber) session.createDurableSubscriber(topic, "name", "color = red", true);

    assertNotNull(subscriber.getTopicSubscriber());
    assertTrue(subscriber.getTopicSubscriber() instanceof MockJMSTopicSubscriber);

    subscriber.close();

    try {
        subscriber.getTopicSubscriber();
        fail("Cannot read state on closed subscriber");
    } catch (IllegalStateException ise) {}
}

@Test(timeout = 60000)
public void testClose() throws Exception {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    assertEquals(0, connection.getNumtIdleSessions());
    session.close();
    assertEquals(1, connection.getNumtIdleSessions());

    try {
        session.close();
    } catch (JMSException ex) {
        fail("Shouldn't fail on second close call.");
    }

    try {
        session.createTemporaryQueue();
        fail("Session should be closed.");
    } catch (IllegalStateException ise) {}
}

@Test
public void testGetQueue() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueSender sender = session.createSender(queue);

    assertNotNull(sender.getQueue());
    assertSame(queue, sender.getQueue());

    sender.close();

    try {
        sender.getQueue();
        fail("Cannot read topic on closed sender");
    } catch (IllegalStateException ise) {}
}

@Test
public void testGetTopicSubscriber() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    JmsPoolQueueSender sender = (JmsPoolQueueSender) session.createSender(queue);

    assertNotNull(sender.getQueueSender());
    assertTrue(sender.getQueueSender() instanceof MockJMSQueueSender);

    sender.close();

    try {
        sender.getQueueSender();
        fail("Cannot read state on closed sender");
    } catch (IllegalStateException ise) {}
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithSameID() throws Exception {
    Connection connection = cf.createConnection();

    // test: call setClientID("newID") twice
    // this should be tolerated and not result in an exception
    connection.setClientID("newID");

    try {
        connection.setClientID("newID");
        connection.start();
        connection.close();
    } catch (IllegalStateException ise) {
        LOG.error("Repeated calls to newID2.setClientID(\"newID\") caused " + ise.getMessage());
        fail("Repeated calls to newID2.setClientID(\"newID\") caused " + ise.getMessage());
    } finally {
        cf.stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithDifferentID() throws Exception {
    Connection connection = cf.createConnection();

    // test: call setClientID() twice with different IDs
    // this should result in an IllegalStateException
    connection.setClientID("newID1");
    try {
        connection.setClientID("newID2");
        fail("calling Connection.setClientID() twice with different clientID must raise an IllegalStateException");
    } catch (IllegalStateException ise) {
        LOG.debug("Correctly received " + ise);
    } finally {
        connection.close();
        cf.stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDAfterConnectionStart() throws Exception {
    Connection connection = cf.createConnection();

    // test: try to call setClientID() after start()
    // should result in an exception
    try {
        connection.start();
        connection.setClientID("newID3");
        fail("Calling setClientID() after start() mut raise a JMSException.");
    } catch (IllegalStateException ise) {
        LOG.debug("Correctly received " + ise);
    } finally {
        connection.close();
        cf.stop();
    }

    LOG.debug("Test finished.");
}

@Test
public void testGetQueue() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueBrowser browser = session.createBrowser(queue);

    assertNotNull(browser.getQueue());

    browser.close();
    browser.close();

    try {
        browser.getQueue();
        fail("Should not be able to use a closed browser");
    } catch (IllegalStateException ise) {
    }
}

@Test
public void testGetQueueBrowser() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    JmsPoolQueueBrowser browser = (JmsPoolQueueBrowser) session.createBrowser(queue);

    assertNotNull(browser.getQueueBrowser());

    browser.close();

    try {
        browser.getQueueBrowser();
        fail("Should not be able to use a closed browser");
    } catch (IllegalStateException ise) {
    }
}

@Test
public void testGetMessageSelector() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueBrowser browser = session.createBrowser(queue, "color = red");

    assertNotNull(browser.getMessageSelector());
    assertEquals("color = red", browser.getMessageSelector());

    browser.close();

    try {
        browser.getMessageSelector();
        fail("Should not be able to use a closed browser");
    } catch (IllegalStateException ise) {
    }
}

@Test
public void testGetEnumeration() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueBrowser browser = session.createBrowser(queue);

    assertNotNull(browser.getEnumeration());

    browser.close();

    try {
        browser.getEnumeration();
        fail("Should not be able to use a closed browser");
    } catch (IllegalStateException ise) {
    }
}

@Test
public void testGetTopic() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createTopicConnection();
    TopicSession session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTemporaryTopic();
    TopicPublisher publisher = session.createPublisher(topic);

    assertNotNull(publisher.getTopic());
    assertSame(topic, publisher.getTopic());

    publisher.close();

    try {
        publisher.getTopic();
        fail("Cannot read topic on closed publisher");
    } catch (IllegalStateException ise) {}
}

@Test
public void testGetTopicPublisher() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createTopicConnection();
    TopicSession session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTemporaryTopic();
    JmsPoolTopicPublisher publisher = (JmsPoolTopicPublisher) session.createPublisher(topic);

    assertNotNull(publisher.getTopicPublisher());
    assertTrue(publisher.getTopicPublisher() instanceof MockJMSTopicPublisher);

    publisher.close();

    try {
        publisher.getTopicPublisher();
        fail("Cannot read state on closed publisher");
    } catch (IllegalStateException ise) {}
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithSameID() throws Exception {
    LOG.debug("running testRepeatedSetClientIDCalls()");

    // test: call setClientID("newID") twice
    // this should be tolerated and not result in an exception
    ConnectionFactory cf = createPooledConnectionFactory();
    Connection conn = cf.createConnection();
    conn.setClientID("newID");

    try {
        conn.setClientID("newID");
        conn.start();
        conn.close();
    } catch (IllegalStateException ise) {
        LOG.error("Repeated calls to newID2.setClientID(\"newID\") caused " + ise.getMessage());
        fail("Repeated calls to newID2.setClientID(\"newID\") caused " + ise.getMessage());
    } finally {
        ((JmsPoolConnectionFactory) cf).stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithDifferentID() throws Exception {
    LOG.debug("running testRepeatedSetClientIDCalls()");

    ConnectionFactory cf = createPooledConnectionFactory();
    Connection conn = cf.createConnection();

    // test: call setClientID() twice with different IDs
    // this should result in an IllegalStateException
    conn.setClientID("newID1");
    try {
        conn.setClientID("newID2");
        fail("calling Connection.setClientID() twice with different clientID must raise an IllegalStateException");
    } catch (IllegalStateException ise) {
        LOG.debug("Correctly received " + ise);
    } finally {
        conn.close();
        ((JmsPoolConnectionFactory) cf).stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDAfterConnectionStart() throws Exception {
    LOG.debug("running testRepeatedSetClientIDCalls()");

    ConnectionFactory cf = createPooledConnectionFactory();
    Connection conn = cf.createConnection();

    // test: try to call setClientID() after start()
    // should result in an exception
    try {
        conn.start();
        conn.setClientID("newID3");
        fail("Calling setClientID() after start() mut raise a JMSException.");
    } catch (IllegalStateException ise) {
        LOG.debug("Correctly received " + ise);
    } finally {
        conn.close();
        ((JmsPoolConnectionFactory) cf).stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithSameID() throws Exception {
    // test: call setClientID("newID") twice
    // this should be tolerated and not result in an exception
    Connection conn = cf.createConnection();
    conn.setClientID("newID");

    try {
        conn.setClientID("newID");
        conn.start();
        conn.close();
    } catch (IllegalStateException ise) {
        LOG.error("Repeated calls to newID2.setClientID(\"newID\") caused " + ise.getMessage());
        fail("Repeated calls to newID2.setClientID(\"newID\") caused " + ise.getMessage());
    } finally {
        cf.stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithDifferentID() throws Exception {
    Connection conn = cf.createConnection();

    // test: call setClientID() twice with different IDs
    // this should result in an IllegalStateException
    conn.setClientID("newID1");
    try {
        conn.setClientID("newID2");
        fail("calling Connection.setClientID() twice with different clientID must raise an IllegalStateException");
    } catch (IllegalStateException ise) {
        LOG.debug("Correctly received " + ise);
    } finally {
        conn.close();
        cf.stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDAfterConnectionStart() throws Exception {
    Connection conn = cf.createConnection();

    // test: try to call setClientID() after start()
    // should result in an exception
    try {
        conn.start();
        conn.setClientID("newID3");
        fail("Calling setClientID() after start() mut raise a JMSException.");
    } catch (IllegalStateException ise) {
        LOG.debug("Correctly received " + ise);
    } finally {
        conn.close();
        cf.stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithSameID() throws Exception {
    LOG.debug("running testRepeatedSetClientIDCalls()");

    // test: call setClientID("newID") twice
    // this should be tolerated and not result in an exception
    JmsPoolConnectionFactory cf = createPooledConnectionFactory();
    Connection conn = cf.createConnection();
    conn.setClientID("newID");

    try {
        conn.setClientID("newID");
        conn.start();
        conn.close();
    } catch (IllegalStateException ise) {
        LOG.error("Repeated calls to newID2.setClientID(\"newID\") caused " + ise.getMessage());
        fail("Repeated calls to newID2.setClientID(\"newID\") caused " + ise.getMessage());
    } finally {
        cf.stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDTwiceWithDifferentID() throws Exception {
    LOG.debug("running testRepeatedSetClientIDCalls()");

    JmsPoolConnectionFactory cf = createPooledConnectionFactory();
    Connection conn = cf.createConnection();

    // test: call setClientID() twice with different IDs
    // this should result in an IllegalStateException
    conn.setClientID("newID1");
    try {
        conn.setClientID("newID2");
        fail("calling Connection.setClientID() twice with different clientID must raise an IllegalStateException");
    } catch (IllegalStateException ise) {
        LOG.debug("Correctly received " + ise);
    } finally {
        conn.close();
        cf.stop();
    }

    LOG.debug("Test finished.");
}

@Test(timeout = 60000)
public void testSetClientIDAfterConnectionStart() throws Exception {
    LOG.debug("running testRepeatedSetClientIDCalls()");

    JmsPoolConnectionFactory cf = createPooledConnectionFactory();
    Connection conn = cf.createConnection();

    // test: try to call setClientID() after start()
    // should result in an exception
    try {
        conn.start();
        conn.setClientID("newID3");
        fail("Calling setClientID() after start() mut raise a JMSException.");
    } catch (IllegalStateException ise) {
        LOG.debug("Correctly received " + ise);
    } finally {
        conn.close();
        cf.stop();
    }

    LOG.debug("Test finished.");
}

@Test
public void testFailsWhenOverriddenGetListenerMethodNameReturnsNull() throws Exception {
    final TextMessage message = mock(TextMessage.class);
    given(message.getText()).willReturn(TEXT);

    final MessageListenerAdapter adapter = new MessageListenerAdapter() {
        @Override
        protected void handleListenerException(Throwable ex) {
            assertTrue(ex instanceof javax.jms.IllegalStateException);
        }
        @Override
        protected String getListenerMethodName(Message originalMessage, Object extractedMessage) {
            return null;
        }
    };
    adapter.setDefaultListenerMethod(null);
    adapter.onMessage(message);
}

@Override
public TemporaryTopic createTemporaryTopic() throws JMSException {
   // As per spec. section 4.11
   if (sessionType == ActiveMQSession.TYPE_QUEUE_SESSION) {
      throw new IllegalStateException("Cannot create a temporary topic on a QueueSession");
   }

   try {
      ActiveMQTemporaryTopic topic = ActiveMQDestination.createTemporaryTopic(this);

      SimpleString simpleAddress = topic.getSimpleAddress();

      // We create a dummy subscription on the topic, that never receives messages - this is so we can perform JMS
      // checks when routing messages to a topic that
      // does not exist - otherwise we would not be able to distinguish from a non existent topic and one with no
      // subscriptions - core has no notion of a topic

      session.createTemporaryQueue(simpleAddress, simpleAddress, ActiveMQSession.REJECTING_FILTER);

      connection.addTemporaryQueue(simpleAddress);

      return topic;
   } catch (ActiveMQException e) {
      throw JMSExceptionHelper.convertFromActiveMQException(e);
   }
}

@Override
public MessageConsumer createSharedDurableConsumer(Topic topic,
                                                   String name,
                                                   String messageSelector) throws JMSException {
   if (sessionType == ActiveMQSession.TYPE_QUEUE_SESSION) {
      throw new IllegalStateException("Cannot create a shared durable consumer on a QueueSession");
   }

   checkTopic(topic);

   ActiveMQTopic localTopic;

   if (topic instanceof ActiveMQTopic) {
      localTopic = (ActiveMQTopic) topic;
   } else {
      localTopic = new ActiveMQTopic(topic.getTopicName());
   }
   return internalCreateSharedConsumer(localTopic, name, messageSelector, ConsumerDurability.DURABLE);
}