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

person.proto

// Generate `person_pb2.py` by running `protoc --python_out=. person.proto` before
// running `protobuf_mqtt.py`

syntax = "proto2";

package tutorial;

message Person {
  required string name = 1;
  required int32 id = 2;
  optional string email = 3;
}

protobuf_mqtt_sub.py

import paho.mqtt.client as mqtt
import io
import person_pb2
import time

# 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("protobuf_out")

# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
    print("msg payload", msg.payload)
    p = person_pb2.Person()
    p.ParseFromString(msg.payload)
    print(msg.topic+" "+str(p))

client = mqtt.Client(client_id = "protobuf")
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()