title: How to create a Spring Reactor Flux from a ActiveMQ queue?
body: <p>I am experimenting with the Spring Reactor 3 components and Spring Integration to create a reactive stream (Flux) from a JMS queue. </p>
  
  <p>I am attempting to create a reactive stream (Spring Reactor 3 Flux) from a JMS queue (ActiveMQ using Spring Integration) for clients to get the JMS messages asynchronously. I believe that I have everything hooked up correctly but the client does not receive any of the JMS messages until the server is stopped. Then all of the messages get "pushed" to the client a once.</p>
  
  <p>Any help would be appreciated.</p>
  
  <p>Here is the configuration file that I am using to configure the JMS, Integration components and the reactive publisher:</p>
  
  <pre><code>@Configuration
  @EnableJms
  @EnableIntegration
  public class JmsConfiguration {
  
      @Value("${spring.activemq.broker-url:tcp://localhost:61616}")
      private String defaultBrokerUrl;
  
      @Value("${queues.patient:patient}")
      private String patientQueue;
  
      @Autowired
      MessageListenerAdapter messageListenerAdapter;
  
      @Bean
      public DefaultJmsListenerContainerFactory myFactory(
              DefaultJmsListenerContainerFactoryConfigurer configurer) {
          DefaultJmsListenerContainerFactory factory =
                  new DefaultJmsListenerContainerFactory();
          configurer.configure(factory, jmsConnectionFactory());
          return factory;
      }
  
      @Bean
      public Queue patientQueue() {
          return new ActiveMQQueue(patientQueue);
  
      }
  
      @Bean
      public ActiveMQConnectionFactory jmsConnectionFactory() {
          ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
          connectionFactory.setBrokerURL(defaultBrokerUrl);
          connectionFactory.setTrustedPackages(Arrays.asList("com.sapinero"));
          return connectionFactory;
      }
  
      // Set the jackson message converter
      @Bean
      public JmsTemplate jmsTemplate() {
          JmsTemplate template = new JmsTemplate();
          template.setConnectionFactory(jmsConnectionFactory());
          template.setDefaultDestinationName(patientQueue);
          template.setMessageConverter(jacksonJmsMessageConverter());
          return template;
      }
  
      @Bean
      public MessageListenerAdapter messageListenerAdapter() {
          MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter();
          messageListenerAdapter.setMessageConverter(jacksonJmsMessageConverter());
          return messageListenerAdapter;
      }
  
      @Bean
      public AbstractMessageListenerContainer messageListenerContainer() {
          DefaultMessageListenerContainer defaultMessageListenerContainer = new DefaultMessageListenerContainer();
          defaultMessageListenerContainer.setMessageConverter(jacksonJmsMessageConverter());
          defaultMessageListenerContainer.setConnectionFactory(jmsConnectionFactory());
          defaultMessageListenerContainer.setDestinationName(patientQueue);
          defaultMessageListenerContainer.setMessageListener(messageListenerAdapter());
          defaultMessageListenerContainer.setCacheLevel(100);
          defaultMessageListenerContainer.setErrorHandler(new ErrorHandler() {
              @Override
              public void handleError(Throwable t) {
                  t.printStackTrace();
              }
          });
  
          return defaultMessageListenerContainer;
      }
  
      @Bean // Serialize message content to json using TextMessage
      public MessageConverter jacksonJmsMessageConverter() {
          MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
          converter.setTargetType(MessageType.TEXT);
          converter.setTypeIdPropertyName("_type");
          return converter;
      }
  
  
      @Bean
      public MessageChannel jmsOutboundInboundReplyChannel() {
          return MessageChannels.queue().get();
      }
  
      @Bean
      public Publisher&lt;Message&lt;String&gt;&gt; pollableReactiveFlow() {
          return IntegrationFlows
                  .from(Jms.messageDrivenChannelAdapter(messageListenerContainer()).get())
                  .channel(MessageChannels.queue())
                  .log(LoggingHandler.Level.DEBUG)
                  .log()
                  .toReactivePublisher();
      }
  
      @Bean
      public MessageChannel jmsChannel() {
          return new DirectChannel();
      }
  </code></pre>
  
  <p>The controller that creates the Flux is:</p>
  
  <pre><code>@RestController
  @RequestMapping("patients")
  public class PatientChangePushController {
      private LocalDateTime lastTimePatientDataRetrieved = LocalDateTime.now();
      private int durationInSeconds = 30;
      private Patient patient;
      AtomicReference&lt;SignalType&gt; checkFinally = new AtomicReference&lt;&gt;();
  
      @Autowired
      PatientService patientService;
  
      @Autowired
      @Qualifier("pollableReactiveFlow")
      private
      Publisher&lt;Message&lt;String&gt;&gt; pollableReactiveFlow;
  
      @Autowired
      private JmsTemplate jmsTemplate;
  
      @Autowired
      private Queue patientQueue;
  
      /**
       * Subscribe to a Flux of a patient that has been updated.
       *
       * @param id
       * @return
       */
      @GetMapping(value = "/{id}/alerts", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
      public Flux&lt;Message&lt;String&gt;&gt; getPatientAlerts(@PathVariable Long id) {
  
          Flux&lt;Message&lt;String&gt;&gt; messageFlux = Flux.from(pollableReactiveFlow);
          return messageFlux;
      }
  
      @GetMapping(value = "/generate")
      public void generateJmsMessage() {
          for (long i = 0L; i &lt; 100; i++) {
              Patient patient = new Patient();
              patient.setId(i);
              send(patient);
              System.out.println("Message was sent to the Queue");
          }
  
      }
  
      void send(Patient patient) {
          this.jmsTemplate.convertAndSend(this.patientQueue, patient);
      }
  
  }
  </code></pre>
  
  <p>If anyone can tell me why the messages do not get sent to the client until after the server is killed, I would appreciate it.</p>