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
## ${topic}: the pulsar topics to which the messages will be published.
## ${filter}: the mqtt topic (may contain wildcard) on which the action will be performed .
## Client Connected Record Hook
bridge.pulsar.hook.client.connected.1 = {"topic": "client_connected"}
## Client Disconnected Record Hook
bridge.pulsar.hook.client.disconnected.1 = {"topic": "client_disconnected"}
## Session Subscribed Record Hook
bridge.pulsar.hook.session.subscribed.1 = {"filter": "#", "topic": "session_subscribed"}
## Session Unsubscribed Record Hook
bridge.pulsar.hook.session.unsubscribed.1 = {"filter": "#", "topic": "session_unsubscribed"}
## Message Publish Record Hook
bridge.pulsar.hook.message.publish.1 = {"filter": "#", "topic": "message_publish"}
## Message Delivered Record Hook
bridge.pulsar.hook.message.delivered.1 = {"filter": "#", "topic": "message_delivered"}
bridge.pulsar.hook.message.acked.1 = {"filter": "#", "topic": "message_acked"}
## More Configures
## partitioner strategy:
## Option: random | roundrobin | first_key_dispatch
## Example: bridge.pulsar.hook.message.publish.1 = {"filter":"#", "topic":"message_publish", "strategy":"random"}
## Option: ${clientid} | ${username}
## Example: bridge.pulsar.hook.message.publish.1 = {"filter":"#", "topic":"message_publish", "key":"${clientid}"}
## format:
## Option: json | json
## 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:
topic = "client_connected",
value = {
"client_id": ${clientid},
"username": ${username},
"node": ${node},
"ts": ${ts}
}
Client goes offline, EMQX forwards ‘client_disconnected’ event message to Pulsar:
topic = "client_disconnected",
value = {
"client_id": ${clientid},
"username": ${username},
"reason": ${reason},
"node": ${node},
"ts": ${ts}
}
Forward Subscription Event to Pulsar
Forward Unsubscription Event to Pulsar
topic = session_unsubscribed
value = {
"client_id": ${clientid},
"qos": ${qos},
"node": ${node},
"ts": ${timestamp}
topic = message_publish
value = {
"client_id": ${clientid},
"username": ${username},
"topic": ${topic},
"payload": ${payload},
"qos": ${qos},
"node": ${node},
"ts": ${timestamp}
}
Forwarding MQTT Message Deliver Event to Pulsar
topic = message_delivered
value = {"client_id": ${clientid},
"username": ${username},
"from": ${fromClientId},
"topic": ${topic},
"payload": ${payload},
"qos": ${qos},
"node": ${node},
"ts": ${timestamp}
}
Forwarding MQTT Message Ack Event to Pulsar
Pulsar consumes MQTT clients connected / disconnected event messages:
sh pulsar-client consume client_connected -s "client_connected" -n 1000
sh pulsar-client consume client_disconnected -s "client_disconnected" -n 1000
sh pulsar-client consume session_subscribed -s "session_subscribed" -n 1000
sh pulsar-client consume session_unsubscribed -s "session_unsubscribed" -n 1000
Pulsar consumes MQTT published messages:
sh pulsar-client consume message_publish -s "message_publish" -n 1000
Pulsar consumes MQTT message Deliver and Ack event messages:
TIP