Сохраняйте весенний контекст до тех пор, пока не будут использованы сообщения JMS

У меня довольно стандартная настройка, связанная с JMS - Spring Boot и ActiveMQ. Он отлично работает, пока я не попытался выполнить простой интеграционный тест. После некоторого расследования я обнаружил, что и контекст Spring, и встроенный брокер закрыты после того, как первое сообщение JMS было израсходовано, независимо от того, что во время потребления происходит другое событие. Проблема брокера, которую я смог решить, добавив параметр useShutdownHook=false connection в тестовую настройку, то есть

spring.activemq.broker-url = vm://broker?async=false&broker.persistent=false&broker.useShutdownHook=false

То, что я ищу, в основном является способом заставить тест "остаться в живых" до тех пор, пока все сообщения JMS не будут потреблены (в этом случае их всего два). Я понимаю асинхронный характер всей установки, но все же во время тестов было бы полезно получить все результаты этих сообщений, которые производятся и потребляются.

Ниже моя настройка, хотя она довольно проста.

@EnableJms
public class ActiveMqConfig {

    @Bean
    public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
        JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
        jmsTemplate.setMessageConverter(messageConverter);
        return jmsTemplate;
    }

    @Bean
    public MessageConverter messageConverter() {
        MappingJackson2MessageConverter messageConverter = new MappingJackson2MessageConverter();
        messageConverter.setTargetType(MessageType.TEXT);
        messageConverter.setTypeIdPropertyName("_type");
        return messageConverter;
    }
}

Затем я получаю POJO с сообщением, который прослушивает данное событие:

@JmsListener(destination = "events")
public void applicationSubmitted(MyType event) {
    // do some work with the event here

    jmsTemplate.convertAndSend("commands", mymessage);
}

И еще один:

@JmsListener(destination = "commands")
public void onCommand(TextMessage textMessage) {

}

Одна вещь, которую я опробовал, и это сработало, - это добавить задержку, то есть sleep(200) после отправки сообщения. Тем не менее, это очень ненадежное, а также замедляет тесты, поскольку выполнение, возможно, занимает меньше 50 мс. Ниже самого теста. Если ожидание не выполнено, я никогда не добираюсь до второго слушателя событий, так как контекст приложения закрывается, тесты заканчиваются, и сообщение "забыто".

@SpringBootTest
class MyEventIntegrationTest extends Specification {

    @Autowired
    JmsTemplate jmsTemplate

    def "My event is successfully handled"() {

        given:
        def event = new MyEvent()

        when:
        jmsTemplate.convertAndSend("events", event)
        // sleep(200)

        then:
        1 == 1
    }
}

Ответы

Ответ 1

Ну, это стандартная проблема при тестировании систем на основе обмена асинхронными сообщениями. Обычно он решает в той части теста, которую вы пропустили - then часть.

Дело в том, что в ваших тестах вы обычно ожидаете, что система сделает что-то полезное, например, внесите изменения в БД, отправьте вызов отдыха в другую систему, отправьте сообщение в другую очередь и т.д. Мы могли бы подождать некоторое время, пока это не произойдет постоянно проверка результата - если результат достигается в течение установленного временного окна, то мы можем предположить, что тест прошел.

Псевдокод этого подхода выглядит следующим образом:

for (i to MAX_RETRIES; i++) {
   checkThatTheChangesInDBHasBeenMade();
   checkThatTheRestCallHasBeenMade();
   checkThatTheMessageIsPostedInAnotherQueue();

   Thread.sleep(50ms);
}

Таким образом, в лучшем случае ваш тест пройдет через 50 мс. В худшем случае это не удастся, и для выполнения теста потребуется MAX_RETRIES * 50 мс.

Кроме того, я должен упомянуть, что есть хороший инструмент под названием awaitility, который обеспечивает хороший API (кстати, он поддерживает массивную DSL), чтобы справиться с такими проблемами в мире асинхронизации:

await().atMost(5, SECONDS).until(customerStatusIsUpdated());

Ответ 2

Я думаю, что корень вашей проблемы - это асинхронная обработка событий. После того, как вы отправите мероприятие, ваш тест закончен. Это, конечно же, приведет к отключению контекста Spring и брокера. Слушатели JMS работают в другом потоке. Вы должны найти способ подождать их. В противном случае ваш поток (который является вашим тестовым примером) только что сделан.

Мы столкнулись с аналогичной проблемой в нашем последнем проекте и написали небольшую полезность, которая нам очень помогла. JMS предлагает возможность "просматривать" очередь и искать, если она пуста:

public final class JmsUtil {

    private static final int MAX_TRIES = 5000;
    private final JmsTemplate jmsTemplate;

    public JmsUtil(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    private int getMessageCount(String queueName) {
        return jmsTemplate.browseSelected(queueName, "true = true", (s, qb) -> Collections.list(qb.getEnumeration()).size());
    }

    public void waitForAll(String queueName) {
        int i = 0;
        while (i <= MAX_TRIES) {
            if (getMessageCount(queueName) == 0) {
                return;
            }
            i++;
        }
}

С помощью этой утилиты вы можете сделать что-то вроде этого:

def "My event is successfully handled"() {

        given:
        def event = new MyEvent()

        when:
        jmsTemplate.convertAndSend("events", event)
        jmsUtility.waitForAll("events"); // wait until the event has been consumed
        jmsUtility.waitForAll("commands"); // wait until the command has been consumed

        then:
        1 == 1
    }

Примечание. Эта утилита предполагает, что вы отправляете сообщения JMS в очередь. Просмотрев очередь, мы можем проверить, не пуст ли он. В случае какой-либо темы вам может понадобиться сделать еще одну проверку. Знайте об этом!