Example of how to use Schema Registry + Rule Engine in EMQX 5.0 (avro, encoding)

avro_mqtt_sub.py

# Original source: https://github.com/terry-xiaoyu/schema-registry-examples/blob/master/avro/avro_mqtt.py
import paho.mqtt.client as mqtt
import io
import json
import avro.schema
import avro.datafile
import avro.io
import avro.ipc
import time

# Got this schema from https://avro.apache.org/docs/current/gettingstartedjava.html
SCHEMA = avro.schema.parse(json.dumps({
 "name"         : "User",
 "type"         : "record",
 "fields"       : [
     {"name": "name"            , "type": "string"},
     {"name": "favorite_number" , "type": ["int", "null"]},
     {"name": "favorite_color"  , "type": ["string", "null"]}
 ]
}))

# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, obj, flags, rc):
    print("Connected with result code "+str(rc))
    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    client.subscribe("avro_out")

# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
    datum_r = avro.io.DatumReader(SCHEMA)
    buf = io.BytesIO(msg.payload)
    decoder = avro.io.BinaryDecoder(buf)
    print("msg payload", msg.payload)
    decoded_payload = datum_r.read(decoder)
    print(msg.topic+" "+str(decoded_payload))

client = mqtt.Client(client_id = "111")
client.reconnect_delay_set(min_delay=120, max_delay=121)
client.on_connect = on_connect
client.on_message = on_message

client.connect("127.0.0.1", 1883, 300)

# Blocking call that processes network traffic, dispatches callbacks and
# handles reconnecting.
# Other loop*() functions are available that give a threaded interface and a
# manual interface.
client.loop_forever()