thalesmg · May 23, 2023 20:09
diff --git a/avro_mqtt_sub.py b/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()
	# 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()