在单元测试中重新启动嵌入式代理:VMTransportServer已绑定

Question

我正在尝试编写一个模拟"经纪人下行"阶段的测试.所以我想

1. 开始当地经纪人
2. 发送消息1
3. 停止经纪人
4. 发送消息2(当然不会到达)
5. 再次启动经纪人
6. 发送消息3

根据http://activemq.apache.org/how-do-i-restart-embedded-broker.html,建议初始化一个新的BrokerService以再次启动代理.所以代码看起来(几乎)像这样:

private BrokerService _broker;

private void startBroker() throws Exception {
    _broker = new BrokerService(); 
    _broker.addConnector("vm://localhost?broker.persistent=false");
    _broker.start();
    _broker.waitUntilStarted();
}

private void stopBroker() throws Exception {
    _broker.stop();
    _broker.waitUntilStopped();
}

@Test
public void publishMessagesWithServerBreakdownInBetween()
    throws Exception
{
    startBroker();
    ... send and receive message (works fine)
    stopBroker();
    ... send message (fails of course)  
    startBroker(); // this fails with java.io.IOException: VMTransportServer already bound at: vm://localhost?broker.persistent=false
    ... send and receive message
}

代码中的注释已经提到了这个问题:由于错误导致代理重启失败:java.io.IOException:VMTransportServer已经绑定在:vm:// localhost？broker.persistent = false

我在ActiveMQ论坛(http://activemq.2283324.n4.nabble.com/VMTransportServer-already-bound-td2364603.html)中发现了类似的问题,但在我的情况下,主机名不为空.

另一个想法是设置2个不同的经纪人名称,但这也没有帮助.

我究竟做错了什么？

Answer 1

您希望通过告诉 VM Transport 不要尝试为您创建代理来控制 VM Transport 的操作，因为您要将其添加到已创建的代理中。剩下的就很简单了：

public class AMQRestartTest {

    private BrokerService broker;
    private String connectorURI;
    private ActiveMQConnectionFactory factory;

    @Before
    public void startBroker() throws Exception {
        createBroker(true);
        factory = new ActiveMQConnectionFactory("failover://" + connectorURI);
    }

    private void createBroker(boolean deleteAllMessages) throws Exception {
        broker = new BrokerService();
        TransportConnector connector = broker.addConnector("vm://localhost?create=false");

        broker.setPersistent(false);
        broker.start();
        broker.waitUntilStarted();

        connectorURI = connector.getConnectUri().toString();
    }

    @Test(timeout = 60_000)
    public void test() throws Exception {
        Connection connection = factory.createConnection();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue("test");
        MessageConsumer consumer = session.createConsumer(queue);
        MessageProducer producer = session.createProducer(queue);

        connection.start();

        broker.stop();
        broker.waitUntilStopped();
        createBroker(false);

        producer.send(session.createTextMessage("help!"));

        Message received = consumer.receive();

        assertNotNull(received);
        assertTrue(received instanceof TextMessage);
    }
}