Added amqp, fixed examples, lint, and tests #25
Conversation
| @@ -116,7 +286,43 @@ def publish(self, topic: str, payload: bytes, persist: bool) -> None: | |||
| if it should be removed immediately (False) | |||
| """ | |||
| # NOTE: RabbitMQ only works with QOS of 1 and 0, and seems to convert QOS2 to QOS1 | |||
| self._connection.publish(topic, payload, qos=2 if persist else 0) | |||
| if self._connection and self.is_connected(): | |||
There was a problem hiding this comment.
This is an improvement over the current code (since the current code can potentially try to call this function while the connection is down, leading to an exception being thrown), but I was thinking that it may make more sense to have an event flag in the MQTT client which can wait for the broker to reconnect and then send the message, instead of simply dropping the message if not connected at the instant this conditional is checked.
| # Resubscribe to all topics to ensure queues and bindings are re-established | ||
| for topic, _ in self._topics_to_handlers().items(): | ||
| logger.info(f'Resubscribing to topic: {topic}') | ||
| self.subscribe(topic, True) |
There was a problem hiding this comment.
Did the resubscription check in this callback fix anything? I would think you would only want to do this in _on_input_channel_open, because technically the output channel can start publishing events immediately once it verifies the exchange.
(when we start using the registry service approach, we can do exchange_declare(..., passive=True))
I've got some lint stuff to clean up still that is being a hassle. Other than that I think everything is ready. Took me a bit longer to fix all the tests, examples, and stuff than I expected.