@Test
public void testBridgeRegistration() throws Exception {
   BrokerService broker = new BrokerService();
   broker.setBrokerName(BROKER_NAME);
   broker.setUseJmx(true); // explicitly set this so no funny issues
   broker.start();
   broker.waitUntilStarted();

   // now create network connector over JMX
   ObjectName brokerObjectName = new ObjectName("org.apache.activemq:type=Broker,brokerName=" + BROKER_NAME);
   BrokerViewMBean proxy = (BrokerViewMBean) broker.getManagementContext().newProxyInstance(brokerObjectName, BrokerViewMBean.class, true);

   assertNotNull("We could not retrieve the broker from JMX", proxy);

   // let's add the NC
   String connectoName = proxy.addNetworkConnector("static:(tcp://localhost:61617)");
   assertEquals("NC", connectoName);

   // Make sure we can retrieve the NC through JMX
   ObjectName networkConnectorObjectName = new ObjectName("org.apache.activemq:type=Broker,brokerName=" + BROKER_NAME +
                                                             ",connector=networkConnectors,networkConnectorName=" + connectoName);
   NetworkConnectorViewMBean nc = (NetworkConnectorViewMBean) broker.getManagementContext().newProxyInstance(networkConnectorObjectName, NetworkConnectorViewMBean.class, true);

   assertNotNull(nc);
   assertEquals("NC", nc.getName());
}
 

@Test(timeout = 60000)
public void testDurableSubscriptionUnsubscribe() throws Exception {
    connection = createAmqpConnection();
    connection.setClientID("DURABLE-AMQP");
    connection.start();

    assertEquals(0, brokerService.getAdminView().getDurableTopicSubscribers().length);
    assertEquals(0, brokerService.getAdminView().getInactiveDurableTopicSubscribers().length);

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    assertNotNull(session);
    Topic topic = session.createTopic(name.getMethodName());
    session.createDurableSubscriber(topic, getSubscriptionName()).close();

    BrokerViewMBean broker = getProxyToBroker();
    assertEquals(1, broker.getInactiveDurableTopicSubscribers().length);

    session.unsubscribe(getSubscriptionName());

    assertEquals(0, broker.getInactiveDurableTopicSubscribers().length);
    assertEquals(0, broker.getDurableTopicSubscribers().length);
}
 

@Test(timeout = 60000)
public void testDurableSubscriptionUnsubscribeNoExistingSubThrowsJMSEx() throws Exception {
    connection = createAmqpConnection();
    connection.setClientID("DURABLE-AMQP");
    connection.start();

    assertEquals(0, brokerService.getAdminView().getDurableTopicSubscribers().length);
    assertEquals(0, brokerService.getAdminView().getInactiveDurableTopicSubscribers().length);

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    assertNotNull(session);

    BrokerViewMBean broker = getProxyToBroker();
    assertEquals(0, broker.getDurableTopicSubscribers().length);
    assertEquals(0, broker.getInactiveDurableTopicSubscribers().length);

    try {
        session.unsubscribe(getSubscriptionName());
        fail("Should have thrown an InvalidDestinationException");
    } catch (InvalidDestinationException ide) {
    }
}
 

@Test(timeout=30000)
public void testDeleteTemporaryQueue() throws Exception {
    connection = createAmqpConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    assertNotNull(queue);
    assertTrue(queue instanceof TemporaryQueue);

    final BrokerViewMBean broker = getProxyToBroker();
    assertEquals(1, broker.getTemporaryQueues().length);

    TemporaryQueue tempQueue = (TemporaryQueue) queue;
    tempQueue.delete();

    assertTrue("Temp Queue should be deleted.", Wait.waitFor(new Wait.Condition() {

        @Override
        public boolean isSatisfied() throws Exception {
            return broker.getTemporaryQueues().length == 0;
        }
    }, TimeUnit.SECONDS.toMillis(30), TimeUnit.MILLISECONDS.toMillis(50)));
}
 

@Test(timeout=30000)
public void testDeleteTemporaryTopic() throws Exception {
    connection = createAmqpConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTemporaryTopic();
    assertNotNull(topic);
    assertTrue(topic instanceof TemporaryTopic);

    final BrokerViewMBean broker = getProxyToBroker();
    assertEquals(1, broker.getTemporaryTopics().length);

    TemporaryTopic tempTopic = (TemporaryTopic) topic;
    tempTopic.delete();

    assertTrue("Temp Topic should be deleted.", Wait.waitFor(new Wait.Condition() {

        @Override
        public boolean isSatisfied() throws Exception {
            return broker.getTemporaryTopics().length == 0;
        }
    }, TimeUnit.SECONDS.toMillis(30), TimeUnit.MILLISECONDS.toMillis(50)));
}
 

protected BrokerViewMBean getProxyToBroker() throws MalformedObjectNameException, JMSException {
    ObjectName brokerViewMBean = new ObjectName(
        "org.apache.activemq:type=Broker,brokerName=" + brokerService.getBrokerName());
    BrokerViewMBean proxy = (BrokerViewMBean) brokerService.getManagementContext()
            .newProxyInstance(brokerViewMBean, BrokerViewMBean.class, true);
    return proxy;
}
 

protected BrokerViewMBean getProxyToBroker() throws MalformedObjectNameException, JMSException {
    ObjectName brokerViewMBean = new ObjectName(
        "org.apache.activemq:type=Broker,brokerName=" + brokerService.getBrokerName());
    BrokerViewMBean proxy = (BrokerViewMBean) brokerService.getManagementContext()
            .newProxyInstance(brokerViewMBean, BrokerViewMBean.class, true);
    return proxy;
}
 

protected BrokerViewMBean getProxyToBroker() throws MalformedObjectNameException, JMSException {
    ObjectName brokerViewMBean = new ObjectName(
        "org.apache.activemq:type=Broker,brokerName=localhost");
    BrokerViewMBean proxy = (BrokerViewMBean) brokerService.getManagementContext()
            .newProxyInstance(brokerViewMBean, BrokerViewMBean.class, true);
    return proxy;
}
 

protected BrokerViewMBean getProxyToBroker() throws MalformedObjectNameException, JMSException {
    ObjectName brokerViewMBean = new ObjectName(
        "org.apache.activemq:type=Broker,brokerName=localhost");
    BrokerViewMBean proxy = (BrokerViewMBean) brokerService.getManagementContext()
            .newProxyInstance(brokerViewMBean, BrokerViewMBean.class, true);
    return proxy;
}
 

protected BrokerViewMBean getProxyToBroker() throws MalformedObjectNameException, JMSException {
    ObjectName brokerViewMBean = new ObjectName(
        "org.apache.activemq:type=Broker,brokerName=localhost");
    BrokerViewMBean proxy = (BrokerViewMBean) brokerService.getManagementContext()
            .newProxyInstance(brokerViewMBean, BrokerViewMBean.class, true);
    return proxy;
}
 

@Test
public void testMessageIsSent() throws Exception {
    producer.sendMessage("Hello: " + name.getMethodName());

    // Should have our send plus the one's sent by the run of MessageProducer by Spring
    QueueViewMBean queueView = getProxyToQueue("example");
    assertEquals(20, queueView.getEnqueueCount());

    BrokerViewMBean brokerView = getProxyToBroker();
    assertEquals(1, brokerView.getCurrentConnectionsCount());
}
 

protected BrokerViewMBean getProxyToBroker() throws MalformedObjectNameException, JMSException {
    ObjectName brokerViewMBean = new ObjectName(
        "org.apache.activemq:type=Broker,brokerName=localhost");
    BrokerViewMBean proxy = (BrokerViewMBean) brokerService.getManagementContext()
            .newProxyInstance(brokerViewMBean, BrokerViewMBean.class, true);
    return proxy;
}
 

private void assertQueueExistsOn(BrokerService broker, String queueName) throws Exception {
   BrokerViewMBean brokerView = broker.getAdminView();
   ObjectName[] queueNames = brokerView.getQueues();
   assertEquals(1, queueNames.length);

   assertTrue(queueNames[0].toString().contains(queueName));
}
 

@SuppressWarnings("unused")
private void assertOneDurableSubOn(BrokerService broker, String subName) throws Exception {
   BrokerViewMBean brokerView = broker.getAdminView();
   ObjectName[] activeDurableSubs = brokerView.getDurableTopicSubscribers();
   ObjectName[] inactiveDurableSubs = brokerView.getInactiveDurableTopicSubscribers();
   ObjectName[] allDurables = (ObjectName[]) ArrayUtils.addAll(activeDurableSubs, inactiveDurableSubs);
   assertEquals(1, allDurables.length);

   // at this point our assertions should prove that we have only on durable sub
   DurableSubscriptionViewMBean durableSubView = (DurableSubscriptionViewMBean) broker.getManagementContext().newProxyInstance(allDurables[0], DurableSubscriptionViewMBean.class, true);

   assertEquals(subName, durableSubView.getClientId());
}
 

public void assertAuthentication(JMXConnector connector) throws Exception {
   connector.connect();
   MBeanServerConnection connection = connector.getMBeanServerConnection();
   ObjectName name = new ObjectName("test.domain:type=Broker,brokerName=localhost");
   BrokerViewMBean mbean = MBeanServerInvocationHandler.newProxyInstance(connection, name, BrokerViewMBean.class, true);
   LOG.info("Broker " + mbean.getBrokerId() + " - " + mbean.getBrokerName());
}
 

@Test(timeout = 60000)
public void testDurableSubscriptionUnsubscribeInUseThrowsAndRecovers() throws Exception {
    connection = createAmqpConnection();
    connection.setClientID("DURABLE-AMQP");
    connection.start();

    assertEquals(0, brokerService.getAdminView().getDurableTopicSubscribers().length);
    assertEquals(0, brokerService.getAdminView().getInactiveDurableTopicSubscribers().length);

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    assertNotNull(session);
    Topic topic = session.createTopic(name.getMethodName());
    MessageConsumer consumer = session.createDurableSubscriber(topic, getSubscriptionName());
    assertNotNull(consumer);

    BrokerViewMBean broker = getProxyToBroker();
    assertEquals(1, broker.getDurableTopicSubscribers().length);
    assertEquals(0, broker.getInactiveDurableTopicSubscribers().length);

    try {
        session.unsubscribe(getSubscriptionName());
        fail("Should have thrown a JMSException");
    } catch (JMSException ex) {
    }

    assertEquals(1, broker.getDurableTopicSubscribers().length);
    assertEquals(0, broker.getInactiveDurableTopicSubscribers().length);

    consumer.close();

    assertEquals(0, broker.getDurableTopicSubscribers().length);
    assertEquals(1, broker.getInactiveDurableTopicSubscribers().length);

    session.unsubscribe(getSubscriptionName());

    assertEquals(0, broker.getDurableTopicSubscribers().length);
    assertEquals(0, broker.getInactiveDurableTopicSubscribers().length);
}
 

@Test(timeout=30000)
public void testCreateTemporaryQueue() throws Exception {
    connection = createAmqpConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    assertNotNull(queue);
    assertTrue(queue instanceof TemporaryQueue);

    final BrokerViewMBean broker = getProxyToBroker();
    assertEquals(1, broker.getTemporaryQueues().length);
}
 

@Test(timeout=30000)
public void testCreateTemporaryTopic() throws Exception {
    connection = createAmqpConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTemporaryTopic();
    assertNotNull(topic);
    assertTrue(topic instanceof TemporaryTopic);

    final BrokerViewMBean broker = getProxyToBroker();
    assertEquals(1, broker.getTemporaryTopics().length);
}
 

protected BrokerViewMBean getProxyToBroker() throws MalformedObjectNameException, JMSException {
    return getProxyToBroker(brokerService);
}
 

protected BrokerViewMBean getProxyToBroker(BrokerService broker) throws MalformedObjectNameException, JMSException {
    ObjectName brokerViewMBean = broker.getBrokerObjectName();
    BrokerViewMBean proxy = (BrokerViewMBean) brokerService.getManagementContext()
            .newProxyInstance(brokerViewMBean, BrokerViewMBean.class, true);
    return proxy;
}