Custom codec example - Avro

Rule requirements

The device publishes a binary message encoded using Avro, which needs to be matched by the rule engine and then republished to the topic associated with the “name” field. The format of the topic is “avro_user/${name}”.

For example, republish a message with the “name” field as “Shawn” to the topic “avro_user/Shawn”.

Create Schema

In the DashboardCustom codec example - Avro - 图1 (opens new window) interface of EMQX, create an Avro Schema using the following parameters:

  1. Name: avro_user

  2. Codec Type: avro

  3. Schema:

  1. {
  2. "type":"record",
  3. "fields":[
  4.     {"name":"name", "type":"string"},
  5.     {"name":"favorite_number", "type":["int", "null"]},
  6.     {"name":"favorite_color", "type":["string", "null"]}
  7. ]
  8. }

Creating rules

Use the Schema you have just created to write the rule SQL statement:

  1. SELECT
  2.   schema_decode('avro_user', payload) as avro_user, payload
  3. FROM
  4.   "t/#"
  5. WHERE
  6.   avro_user.name = 'Shawn'

The key point here is schema_decode('avro_user', payload):

  • The schema_decode function decodes the contents of the payload field according to the Schema “avro_user”;
  • as avro_user stores the decoded value in the variable “avro_user”;

Then add the action using the following parameters:

  • Action Type: Message republishing
  • Destination Topic: avro_user/${avro_user.name}
  • Message Content Template: ${avro_user}

This action sends the decoded “user” to the topic avro_user/${avro_user.name} in JSON format. ${avro_user.name} is a variable placeholder that will be replaced at runtime with the value of the “name” field in the message content.

Device side code

Once the rules have been created, it is time to simulate the data for testing.

The following code uses the Python language to fill a User message and encode it as binary data, then sends it to the “t/1” topic. See full codeCustom codec example - Avro - 图2 (opens new window) for details.

  1. def publish_msg(client):
  2.     datum_w = avro.io.DatumWriter(SCHEMA)
  3.     buf = io.BytesIO()
  4.     encoder = avro.io.BinaryEncoder(buf)
  5.     datum_w.write({"name": "Shawn", "favorite_number": 666, "favorite_color": "red"}, encoder)
  6.     message = buf.getvalue()
  7.     topic = "t/1"
  8.     print("publish to topic: t/1, payload:", message)
  9.     client.publish(topic, payload=message, qos=0, retain=False)

Checking rule execution results

  1. In the Dashboard’s WebsocketCustom codec example - Avro - 图3 (opens new window) tools, log in to an MQTT Client and subscribe to “avro_user/#”.

  2. Install the python dependency and execute the device-side code:

  1. $ pip3 install protobuf
  2. $ pip3 install paho-mqtt
  4. $ python3 avro_mqtt.py
  5. Connected with result code 0
  6. publish to topic: t/1, payload: b'\nShawn\x00\xb4\n\x00\x06red'
  1. Check that a message with the topic avro_user/Shawn is received on the Websocket side:
  1. {"favorite_color":"red","favorite_number":666,"name":"Shawn"}