Как отправить запланированное сообщение в служебную шину Azure с помощью JMS

Служебная шина Azure может отправлять сообщения по расписанию. Отправка запланированных сообщений с использованием протокола AMQP, описанного здесь: https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-amqp-request-response#message-operations

Планирует сообщения. Запрос

Сообщение запроса должно включать следующие свойства приложения:

| Ключ | Значение | Тип | Обязательно | Значение Содержание

| операция | строка | Да | com.microsoft:schedule-message

| com.microsoft:server-timeout | uint | Нет | Тайм-аут сервера операций в миллисекундах.

Я работаю со служебной шиной Azure с java JmsTemplate из Spring Framework. Как отображать заголовки сообщений для отправки запланированного сообщения?

    @Test
public void sendMessageWithHeaders() {


    jmsTemplate.send("test-topic-2", new MessageCreator() {
        @Override
        public Message createMessage(Session session) throws JMSException {
            TextMessage textMessage = session.createTextMessage("test-123");
            ((JmsTextMessage) textMessage).setValidatePropertyNames(false);
            textMessage.setStringProperty("operation", "com.microsoft:schedule-message");

            textMessage.setIntProperty("com.microsoft:server-timeout", 100000);
            return textMessage;
        }
    });
}

-вывести порядковое сообщение


azure azureservicebus amqp jmstemplate
person qwazer    schedule 11.08.2017    source источник

Ответы (2)


arrow_upward
3
arrow_downward

Этот код работает:

Azure SB использует недокументированный заголовок аннотации сообщения x-opt-scheduled-enqueue-time

static final long ONE_MINUTE_IN_MILLIS=60000;//millisecs

@Test
public void sendMessageWithHeaders() {


    jmsTemplate.send(queueName, new MessageCreator() {
        @Override
        public Message createMessage(Session session) throws JMSException {
            TextMessage textMessage = session.createTextMessage("test-123");
            ((JmsTextMessage) textMessage).setValidatePropertyNames(false);

            org.apache.qpid.proton.message.Message amqpMessage = ((AmqpJmsTextMessageFacade)((JmsTextMessage)textMessage).getFacade()).getAmqpMessage();
            HashMap applicationPropertiesMap = new HashMap();
            applicationPropertiesMap.put("operation", "com.microsoft:schedule-message");
            applicationPropertiesMap.put("com.microsoft:server-timeout", 100000000);
            amqpMessage.setApplicationProperties(new ApplicationProperties(applicationPropertiesMap));

            Calendar date = Calendar.getInstance();
            long t= date.getTimeInMillis();
            Date afterAddingTenMins=new Date(t + (10 * ONE_MINUTE_IN_MILLIS));

            amqpMessage.getMessageAnnotations().getValue().put(Symbol.valueOf("x-opt-scheduled-enqueue-time"), afterAddingTenMins);

            return textMessage;
        }
    });
}
person qwazer    schedule 11.08.2017
comment
.getAmqpMessage () недоступен для класса AmqpJmsTextMessageFacade для меня. может быть они его удалили. - person Saurabh Gangwar; 17.09.2020

arrow_upward
1
arrow_downward

Обновление 2021 года: как вы можете видеть в моем ответе на этот вопрос, вам нужен немного другой подход, поскольку .getAmqpMessage() недоступен больше:

public void sendDelayedMessage() {
    final var now = ZonedDateTime.now();
    jmsTemplate.send("test-queue", session -> {
        final var tenMinutesFromNow = now.plusMinutes(10);
        final var textMessage = session.createTextMessage("Hello Service Bus!");
        ((JmsTextMessage) textMessage).getFacade().setTracingAnnotation("x-opt-scheduled-enqueue-time", Date.from(tenMinutesFromNow.toInstant()));
        return textMessage;
    });
    log.info("Sent at: " + now);
}
person Raphael Winkler    schedule 07.04.2021