message-driven-channel-adapter отбрасывает первое сообщение после запуска контекста приложения, если send не вызван с задержкой

Question

У меня есть интеграционный тест для моей конфигурации интеграции Spring, который потребляет сообщения из темы JMS с долгой подпиской. Для тестирования я использую ActiveMQ вместо Tibco EMS. Проблема заключается в том, что мне нужно отложить отправку первого сообщения в конечную точку, используя вызов сна в начале нашего тестового метода. В противном случае сообщение будет удалено. Если я удалю настройку для длительной подписки и селектора, то первое сообщение можно отправить сразу же без задержки. Я бы хотел избавиться от сна, что ненадежно. Есть ли способ проверить, полностью ли конечная точка установлена ​​до отправки сообщения? Ниже приведена конфигурация.

Спасибо за помощь!

<int-jms:message-driven-channel-adapter 
     id="myConsumer" connection-factory="myCachedConnectionFactory" 
     destination="myTopic" channel="myChannel" error-channel="errorChannel" 
     pub-sub-domain="true" subscription-durable="true" 
     durable-subscription-name="testDurable" 
     selector="..." 
     transaction-manager="emsTransactionManager" auto-startup="false"/> 

Answer 1

Если вы используете чистый встроенный ActiveMQ для испытания, долговечность подписки не имеет значения, пока подписка не будет установлена. Поэтому у вас нет выбора, кроме как ждать, пока это произойдет.

Вы можете избежать сна, отправив ряд сообщений о запуске и только начать реальный тест при получении последнего.

EDIT

Я забыл, что существует метод isRegisteredWithDestination() на DefaultMessageListenerContainer.

Javadocs ...

/** 
* Return whether at least one consumer has entered a fixed registration with the 
* target destination. This is particularly interesting for the pub-sub case where 
* it might be important to have an actual consumer registered that is guaranteed 
* not to miss any messages that are just about to be published. 
* <p>This method may be polled after a {@link #start()} call, until asynchronous 
* registration of consumers has happened which is when the method will start returning 
* {@code true} &ndash; provided that the listener container ever actually establishes 
* a fixed registration. It will then keep returning {@code true} until shutdown, 
* since the container will hold on to at least one consumer registration thereafter. 
* <p>Note that a listener container is not bound to having a fixed registration in 
* the first place. It may also keep recreating consumers for every invoker execution. 
* This particularly depends on the {@link #setCacheLevel cache level} setting: 
* only {@link #CACHE_CONSUMER} will lead to a fixed registration. 
*/ 

Мы используем его в some channel tests, где мы получаем контейнер с помощью отражения, а затем опрашивать метод пока не подписаны на эту тему.

/** 
* Blocks until the listener container has subscribed; if the container does not support 
* this test, or the caching mode is incompatible, true is returned. Otherwise blocks 
* until timeout milliseconds have passed, or the consumer has registered. 
* @see DefaultMessageListenerContainer#isRegisteredWithDestination() 
* @param timeout Timeout in milliseconds. 
* @return True if a subscriber has connected or the container/attributes does not support 
* the test. False if a valid container does not have a registered consumer within 
* timeout milliseconds. 
*/ 
private static boolean waitUntilRegisteredWithDestination(SubscribableJmsChannel channel, long timeout) { 
    AbstractMessageListenerContainer container = 
      (AbstractMessageListenerContainer) new DirectFieldAccessor(channel).getPropertyValue("container"); 
    if (container instanceof DefaultMessageListenerContainer) { 
     DefaultMessageListenerContainer listenerContainer = 
      (DefaultMessageListenerContainer) container; 
     if (listenerContainer.getCacheLevel() != DefaultMessageListenerContainer.CACHE_CONSUMER) { 
      return true; 
     } 
     while (timeout > 0) { 
      if (listenerContainer.isRegisteredWithDestination()) { 
       return true; 
      } 
      try { 
       Thread.sleep(100); 
      } catch (InterruptedException e) { } 
      timeout -= 100; 
     } 
     return false; 
    } 
    return true; 
}