java - 在单元测试 : VMTransportServer already bound 中重新启动嵌入式代理

Question

我正在尝试编写一个模拟“中断”阶段的测试。
因此我想

启动本地代理

发送消息1

停止经纪人

发送 message2(当然不会到达)

再次启动代理

发送消息3

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

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 已绑定(bind) 在:vm://localhost?broker.persistent=false

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

另一个想法是设置 2 个不同的经纪人名称，但这也无济于事。

我究竟做错了什么？

Answer 1

您希望通过告诉它不要尝试为您创建代理来控制 VM 传输的功能，因为您正在将其添加到已创建的代理中。剩下的就很简单了:

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);
    }
}