Docs

Cloud Native Message Engine

Best practices

Implement P2P command delivery through automatic subscription

MQTT devices typically specify only the access address of the MQTT server at the factory, but do not specify the MQTT Topic to which they subscribe. Cloud-based control services (usually Java, Go, or C++) aim to deliver commands based on a specified device ID. For example, a backend service implemented in Java wants to send a message to an MQTT device with clientid vin001, but the vin001 device has only established an MQTT connection and has not actively subscribed to any MQTT Topic. This scenario can utilize a combination of MQ and MQTT, together with MQTT automatic subscription, to achieve a communication method similar to "P2P" (point-to-point).

Architecture description

Preparations

Create a Message Queue for Kafka Edition instance and create a Topic named restart of Topic. For detailed instructions, see Create Instance and Create Topic.

Configure a Kafka user in Message Queue for Kafka Edition, granting read and write permissions to the restart Topic. For detailed instructions, see Create SASL user and Create ACL.

Test user: name root, password ******.

Complete user authentication and authorization configuration in MQTT Dashboard, referring to Quickly set up your first MQTT connection.

Test user: name user, password 123456.

Ensure that Message Queue for Kafka Edition and Cloud-native Message Engine (MQTT Edition) are in the same VPC.

Operation steps

Create a Kafka consumer connector and configure Kafka consumer subscription for EMQX.

For detailed instructions, see Create connector.

Select the connector type as Kafka consumer.

Enter the Kafka VPC endpoint, authentication, authentication method, username, and password.

Configure forwarding rules.

For detailed instructions, see Create rule.

On the right side of the interface, click Data input, then click Edit.

Configure the data input as Kafka consumer.

Select the input type Kafka consumer.

Configure the specified Kafka consumer connector.

Declare that the rule subscribes to the Kafka topic named restart.

Select the Key and Value encoding modes

In the example, they are Base64 and None, and can be modified as needed. If the encoding mode is changed, the processing functions for the corresponding fields in subsequent rule SQL statements must also be updated accordingly.

Click Action output, then click Add action to configure the action output.

Select the action type Message republish.

Declare that the topic name subscribed to by this rule is /restart/${key}.

After specifying the subscription to Kafka topic data, the MQTT topic forwarded by EMQX is a dynamic topic, which is derived from the Kafka message key. When the Java backend service sends a message to Kafka, the clientid is intentionally set as the message key, which is processed by the EMQX rule into /restart/key. The device automatically subscribes to its MQTT topic /restart/vin, where the message key and clientid are identical; for example, if both are vin001, the message is bound to the device subscription.

Set the MQTT payload to the {aim} field.

Edit the SQL rule.

The aim field in the value of the JSON-formatted message is used as the new aim field (according to the above rule, it is passed into the payload).

When configuring data input as a Kafka consumer, the SQL statement for the From field is automatically configured and does not require manual modification.

SELECT
value.aim as aim, -- payload is configured as {aim}
base64_decode(key) as key -- This field is not included in the payload, so it will not actually be transmitted in this example
FROM
"$bridges/kafka_consumer:kafka" -- This field is automatically updated by the system

Message validation.

Use the message sending function in the Kafka message queue console to send data to the Kafka restart topic.

If the data is correctly forwarded to the connected vin001 device, this indicates successful validation.

Last updated: 2026.01.26 16:22:55