Интеграция с MQTT

Описание

В данном примере демонстрируется взаимодействие с брокером сообщений ActiveMQ Classic по протоколу MQTT с использованием компонента camel-paho.

  1. Запустить ActiveMQ Classic в докере:

    docker pull apache/activemq-classic:latest
docker run -p 61617:61616 -p 8162:8161 -p 5673:5672 -p 1884:1883 apache/activemq-classic:latest

  2. Установить фичу Camel Paho:

    feature:install camel-stomp

  3. Убедиться, что установленные бандлы находятся в статусе Active:

    622 │ Active   │  50 │ 3.4.5              │ camel-paho
623 │ Active   │  50 │ 1.2.4              │ Paho MQTT Client

  4. Создать профиль системы с именем mqtt и добавить к нему:

    1. Входной кастомный коннектор (consumer). В точке кастомизации custom-route создаем маршрут, который подписывается на топик entaxy/queue и логирует каждое полученное сообщение:

      <?xml version="1.0" encoding="UTF-8"?>
<entaxy:object-input-route
    xmlns="http://camel.apache.org/schema/blueprint"
    xmlns:blueprint="http://www.osgi.org/xmlns/blueprint/v1.0.0"
    xmlns:entaxy="http://www.entaxy.ru/schemas/1.0"
    xmlns:m="http://www.entaxy.ru/schemas/entaxy-mediators/1.0">
    <from uri="paho:entaxy/queue?brokerUrl=tcp://localhost:1884"/>
    <m:log message="Received message: ${body} on topic ${header.CamelMqttTopic}"/>
    <m:respond now="true" continue="false"/>
</entaxy:object-input-route>

    2. Выходной кастомный коннектор (producer). В точке кастомизации custom-route создаем маршрут, который формирует текстовое сообщение и публикует его в топик entaxy/queue:

      <?xml version="1.0" encoding="UTF-8"?>
<entaxy:object-route
    xmlns="http://camel.apache.org/schema/blueprint"
    xmlns:blueprint="http://www.osgi.org/xmlns/blueprint/v1.0.0"
    xmlns:entaxy="http://www.entaxy.ru/schemas/1.0"
    xmlns:m="http://www.entaxy.ru/schemas/entaxy-mediators/1.0">
    <setBody>
        <simple>Hello MQTT Entaxy at ${date:now:HH:mm:ss}</simple>
    </setBody>
    <to uri="paho:entaxy/queue?brokerUrl=tcp://localhost:1884"/>
</entaxy:object-route>

    3. К выходному кастомному коннектору добавляем маршрут на базе компонента таймер, инициирующий вызов коннектора с заданным интервалом:

      <?xml version="1.0" encoding="UTF-8"?>
<entaxy:common-route
    xmlns="http://camel.apache.org/schema/blueprint"
    xmlns:blueprint="http://www.osgi.org/xmlns/blueprint/v1.0.0"
    xmlns:entaxy="http://www.entaxy.ru/schemas/1.0"
    xmlns:m="http://www.entaxy.ru/schemas/entaxy-mediators/1.0">
    <m:log message="in route ${routeId}" loggingLevel="INFO"/>
    <to uri="system:mqtt"/>
</entaxy:common-route>

  5. При запуске таймера в логах отображаются сообщения, полученные входным коннектором из топика entaxy/queue:

    17:08:44.443 INFO [MQTT Call: camel-paho25387628562400] ce019fd6-6afc-4ecd-8a13-bfdddc3f7a5c#{"service":"sys-26","sender":"mqtt"}# Received message: Hello MQTT Entaxy at 17:08:44 on topic entaxy/queue