Pulsar Bridge

    After EMQX version 3.1, a powerful rule engine is introduced to replace plug-ins. It is recommended that you use it. See Bridge data to Pulsar to setup pulasr bridges in rule engine.

    EMQX bridges and forwards MQTT messages to Pulsar cluster:

    Configure Pulsar Bridge Hooks

    1. ## ${topic}: the pulsar topics to which the messages will be published.
    2. ## ${filter}: the mqtt topic (may contain wildcard) on which the action will be performed .
    3. ## Client Connected Record Hook
    4. bridge.pulsar.hook.client.connected.1 = {"topic": "client_connected"}
    5. ## Client Disconnected Record Hook
    6. bridge.pulsar.hook.client.disconnected.1 = {"topic": "client_disconnected"}
    7. ## Session Subscribed Record Hook
    8. bridge.pulsar.hook.session.subscribed.1 = {"filter": "#", "topic": "session_subscribed"}
    9. ## Session Unsubscribed Record Hook
    10. bridge.pulsar.hook.session.unsubscribed.1 = {"filter": "#", "topic": "session_unsubscribed"}
    11. ## Message Publish Record Hook
    12. bridge.pulsar.hook.message.publish.1 = {"filter": "#", "topic": "message_publish"}
    13. ## Message Delivered Record Hook
    14. bridge.pulsar.hook.message.delivered.1 = {"filter": "#", "topic": "message_delivered"}
    15. bridge.pulsar.hook.message.acked.1 = {"filter": "#", "topic": "message_acked"}
    16. ## More Configures
    17. ## partitioner strategy:
    18. ## Option: random | roundrobin | first_key_dispatch
    19. ## Example: bridge.pulsar.hook.message.publish.1 = {"filter":"#", "topic":"message_publish", "strategy":"random"}
    20. ## Option: ${clientid} | ${username}
    21. ## Example: bridge.pulsar.hook.message.publish.1 = {"filter":"#", "topic":"message_publish", "key":"${clientid}"}
    22. ## format:
    23. ## Option: json | json
    24. ## Example: bridge.pulsar.hook.message.publish.1 = {"filter":"#", "topic":"message_publish", "format":"json"}

    Description of Pulsar Bridge Hooks

    Client goes online, EMQX forwards ‘client_connected’ event message to Pulsar:

    1. topic = "client_connected",
    2. value = {
    3. "client_id": ${clientid},
    4. "username": ${username},
    5. "node": ${node},
    6. "ts": ${ts}
    7. }

    Client goes offline, EMQX forwards ‘client_disconnected’ event message to Pulsar:

    1. topic = "client_disconnected",
    2. value = {
    3. "client_id": ${clientid},
    4. "username": ${username},
    5. "reason": ${reason},
    6. "node": ${node},
    7. "ts": ${ts}
    8. }

    Forward Subscription Event to Pulsar

    Forward Unsubscription Event to Pulsar

    1. topic = session_unsubscribed
    2. value = {
    3. "client_id": ${clientid},
    4. "qos": ${qos},
    5. "node": ${node},
    6. "ts": ${timestamp}
    1. topic = message_publish
    2. value = {
    3. "client_id": ${clientid},
    4. "username": ${username},
    5. "topic": ${topic},
    6. "payload": ${payload},
    7. "qos": ${qos},
    8. "node": ${node},
    9. "ts": ${timestamp}
    10. }

    Forwarding MQTT Message Deliver Event to Pulsar

    1. topic = message_delivered
    2. value = {"client_id": ${clientid},
    3. "username": ${username},
    4. "from": ${fromClientId},
    5. "topic": ${topic},
    6. "payload": ${payload},
    7. "qos": ${qos},
    8. "node": ${node},
    9. "ts": ${timestamp}
    10. }

    Forwarding MQTT Message Ack Event to Pulsar

    Pulsar consumes MQTT clients connected / disconnected event messages:

    1. sh pulsar-client consume client_connected -s "client_connected" -n 1000
    2. sh pulsar-client consume client_disconnected -s "client_disconnected" -n 1000
    1. sh pulsar-client consume session_subscribed -s "session_subscribed" -n 1000
    2. sh pulsar-client consume session_unsubscribed -s "session_unsubscribed" -n 1000

    Pulsar consumes MQTT published messages:

    1. sh pulsar-client consume message_publish -s "message_publish" -n 1000

    Pulsar consumes MQTT message Deliver and Ack event messages:

    TIP

    Enable Pulsar Bridge