Add MQTT Sink #659
Add MQTT Sink #659
Conversation
There was a problem hiding this comment.
@SteveRosam I also need this sink and willing to contribute to your PR if you don't mind
quixstreams/sinks/community/mqtt.py
Outdated
| if isinstance(data, bytes): | ||
| data = data.decode("utf-8") # Decode bytes to string using utf-8 | ||
|
|
||
| json_data = json.dumps(data) |
There was a problem hiding this comment.
Some sinks has concept of
value_serializer: Callable[[Any], str] = json.dumps,
key_serializer: Callable[[Any], str] = bytes.decode,in its constructors. That is very handy, specially when you are dealing with non json data. I don't see a use case for key_serializer, but value_serializer is definitely something nice to have.
There was a problem hiding this comment.
@devova I have added support for this, please let me know what you think!
quixstreams/sinks/community/mqtt.py
Outdated
| self.mqtt_topic_root + "/" + message_key_string, | ||
| payload=json_data, | ||
| qos=self.qos, |
There was a problem hiding this comment.
According to docs, it would be great to have a control over retain and properties arguments.
Ideally the constructor should accept either constant or callbacks
retain: bool | Callable[[Any], bool] = False,
properties: paho.mqtt.properties.Properties | Callable[[Any], paho.mqtt.properties.Properties] | None = None,There was a problem hiding this comment.
@devova I have added support for these, please let me know what you think!
quixstreams/sinks/community/mqtt.py
Outdated
| "utf-8" | ||
| ) # Convert to string using utf-8 encoding | ||
| # publish to MQTT | ||
| self.mqtt_client.publish( |
There was a problem hiding this comment.
It would be good to add sync primitive here. Client.publish returns MQTTMessageInfo which has
wait_for_publish()
| self.mqtt_client.publish( | |
| self.message_infos.add(self.mqtt_client.publish(...)) | |
| def flush(self, topic: str, partition: str): | |
| for message_info in self.message_infos: | |
| message_info.wait_for_publish() |
see quix-streams sink flush docs
There was a problem hiding this comment.
Heya @devova , sorry that it's been a while, but I have overhauled this MQTT sink and will be attempting to get this merged next week. =)
It now better handles the flushing/tracking of messages when using QoS = 1. Feel free to check it out and let me know what you think.
tim-quix
force-pushed
the
feature/mqtt-sink
branch
from
02ead3d to
32c5f63
Compare
Added MQTT connector
Added tests
Updated
pyproject.tomlUpdated
tests/requirement.txtResolves #658