Spring webflux stream не может получать сообщения, используемые адаптером, управляемым сообщениями JMS - PullRequest
Новая
       21

Spring webflux stream не может получать сообщения, используемые адаптером, управляемым сообщениями JMS

1 голос
/ 06 февраля 2020

Я экспериментирую с Spring Webflux и Spring Integration для создания реактивного потока (Flux) из очереди JMS.

Я пытаюсь создать реактивный поток (Spring Webflux) из очереди JMS (IBM MQ использует Spring Integration) для клиентов, чтобы получать сообщения JMS асинхронно. Я полагаю, что у меня все правильно подключено, поскольку сообщения воспринимаются реактивным слушателем. Однако мой поток реактивного потока не может воспроизвести эти сообщения. Буду признателен за любую помощь.

Вот код, который я использую, чтобы мой JMS-слушатель реагировал:

UM Gateway 

@Named
@Slf4j
public class UmGateway {

  @Autowired
  private ConnectionFactory connectionFactory;


  @Autowired
  private JmsTemplate jmsTemplate;

  @Value("${um.mq.queueName}")
  private String queueName;

  @Bean
  public Publisher<Message<MilestoneNotification>> jmsReactiveSource() {
    return IntegrationFlows
        .from(Jms.messageDrivenChannelAdapter(this.connectionFactory)
            .destination(queueName))
        .channel(MessageChannels.queue())
        .log(Level.DEBUG)
        .log()
        .toReactivePublisher();
  }

  public Flux<MilestoneNotification> findAll() {
      return Flux.from(jmsReactiveSource())
          .map(Message::getPayload);
  }


  /**
   * Method which sends Milestone Notifications to the UM Queue.
   */
  public void send(final MilestoneNotification message) {
      jmsTemplate.convertAndSend(queueName, message);
  }
}

Контроллер 

@RestController
@Slf4j
@RequiredArgsConstructor
@RequestMapping(ApiConstants.MILESTONE_UM)
public class MilestoneUmController {

  @Autowired
  private UmGateway umGateway;

  @RequestMapping(value = "/message", method = RequestMethod.POST)
  public ResponseEntity<Boolean> sendMessage(
      final @RequestBody MilestoneNotification notification) {
    umGateway.send(notification);
    return new ResponseEntity<>(HttpStatus.OK);
  }

  @GetMapping(path = "/milestone-notification/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
  public Flux<MilestoneNotification> feed() {
    return this.umGateway.findAll();
  }

}

Вот журналы: 

 - 2020.02.06 13:53:04.900 [jmsReactiveSource.org.springframework.jms.listener.DefaultMessageListenerContainer#0-1] INFO  o.s.i.h.LoggingHandler - GenericMessage [payload={"messageId":"MAHESH_007","NotificationTag":"MAHESH_007","messageTimeStamp":"2020-01-21T10:56:33Z","processMilestoneId":"MAHESH_007","processMilestoneName":"MAHESH_007","accountNumber":"12345","previousStatus":"In Progress","currentStatus":"complete","isNew":true}, headers={JMS_IBM_Character_Set=UTF-8, JMS_IBM_MsgType=8, jms_destination=queue:///NOTIFICATIONQUEUE, _type=com.jpmc.wss.portal.domain.um.MilestoneNotification, JMSXUserID=cibcfdid    , JMS_IBM_Encoding=273, priority=4, jms_timestamp=1580997184878, JMSXAppID=jpmc.wss.portal.Application , JMS_IBM_PutApplType=28, JMS_IBM_Format=MQSTR   , jms_redelivered=false, JMS_IBM_PutDate=20200206, JMSXDeliveryCount=1, JMS_IBM_PutTime=13530511, id=5d277be2-49f5-3e5d-8916-5793db3b76e7, jms_messageId=ID:414d51204e41544d31383820202020201d9f3b5e03738521, timestamp=1580997184900}]
 - 2020.02.06 13:53:04.968 [qtp2132762784-23] DEBUG c.j.w.p.u.RequestLoggingInterceptor - Returning status code 200 for POST request to /common/dataservice/di/milestone/um/message with query=[null] and http-user=[null]
 - 2020.02.06 13:53:53.521 [qtp2132762784-18] INFO  c.j.w.p.u.RequestLoggingInterceptor - Received GET request to /common/dataservice/di/milestone/um/milestone-notification/stream with query=[null] and http-user=[null]
 - 2020.02.06 13:54:09.070 [qtp2132762784-16] INFO  c.j.w.p.u.RequestLoggingInterceptor - Received POST request to /common/dataservice/di/milestone/um/message with query=[null] and http-user=[null]
 - 2020.02.06 13:54:09.541 [jmsReactiveSource.org.springframework.jms.listener.DefaultMessageListenerContainer#0-1] INFO  o.s.i.h.LoggingHandler - GenericMessage [payload={"messageId":"MAHESH_007","diNotificationTag":"MAHESH_007","messageTimeStamp":"2020-01-21T10:56:33Z","processMilestoneId":"MAHESH_007","processMilestoneName":"MAHESH_007","accountNumber":"12345","previousStatus":"In Progress","currentStatus":"complete","isNew":true}, headers={JMS_IBM_Character_Set=UTF-8, JMS_IBM_MsgType=8, jms_destination=queue:///NOTIFICATIONQUEUE, _type=com.jpmc.wss.portal.domain.um.MilestoneNotification, JMSXUserID=cibcfdid    , JMS_IBM_Encoding=273, priority=4, jms_timestamp=1580997249519, JMSXAppID=jpmc.wss.portal.Application , JMS_IBM_PutApplType=28, JMS_IBM_Format=MQSTR   , jms_redelivered=false, JMS_IBM_PutDate=20200206, JMSXDeliveryCount=1, JMS_IBM_PutTime=13540975, id=5421898e-5ef6-1f9b-aaa6-81ebc7668f50, jms_messageId=ID:414d51204e41544d31383820202020201d9f3b5e08738521, timestamp=1580997249541}]
 - 2020.02.06 13:54:09.593 [qtp2132762784-16] DEBUG c.j.w.p.u.RequestLoggingInterceptor - Returning status code 200 for POST request to /common/dataservice/di/milestone/um/message with query=[null] and http-user=[null]

Поток потока в браузере

1 Ответ

1 голос
/ 07 февраля 2020

Таким образом, вы не можете просто сделать URL-запрос из браузера, когда вам нужны события на стороне сервера.

Вам нужен какой-то определенный JavaScript API для использования с вашей веб-страницы: https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events

Я могу сделать это для тестирования с помощью инструмента curl -N: cURL - Структурирование запроса для проверки событий, отправленных сервером

Или с использованием WebTestClient Форма Spring WebFlux: 

Flux<String> stream =
            this.webTestClient.get().uri("/stream")
                    .exchange()
                    .returnResult(String.class)
                    .getResponseBody();

    StepVerifier
            .create(stream)
            .expectNext("m1", "m2", "m3")
            .thenCancel()
            .verify();
...