amqp unsub routine now works, make broker logic more robust

[Lance-Drane]

Will probably add this to a v0.6.3 release since the public API does not change at all, but note that internally AMQP functions very differently.

Possible changes needed for experiment controller

Whenever calling some of the ControlPlaneManager functions (add_subscription_channel and publish_message), you'll now need to add a persist boolean flag onto the calls - certain messages persist and certain messages don't.

AMQP CHANGES

• unsubscribe actually works now (added another AMQP E2E test to prove it)
• Now use topic exchanges instead of fanout exchanges. Instead of creating an exchange for each topic like we were doing (and queues for each), we use a single hardcoded exchange. Some queues are meant to be dedicated (for example: Services will handle Userspace messages in their own dedicated queue, and INTERSECT Core Services will always use a dedicated queue for all messages), while other queues are meant to be transient (for example: when listening to events, Clients will create a temporary transient queue).

What I'd like to do in the future would be to create the exchanges, the dedicated queues, and the queue/exchange bindings on RabbitMQ (the server) instead of in the SDK (both Services and Clients act as RabbitMQ clients). This will allow us to configure AUTH on the broker itself by topic, and not "rely" on something using the SDK inappropriately reading/writing messages from certain dedicated queues.

General Changes

Both the MQTT and the AMQP implementations are now more robust regarding reconnecting to the broker. Both will attempt to reconnect on failure, and both will automatically resubscribe to all topics they should be subscribed to once the reconnect is successful.

[marshallmcdonnell]

Overall, LGTM!

I only have general questions about certain things to help my understanding.

Besides that, the failing tests related to the channel stuff are the only blockers right now.

Is this a "false negative" test?
Something wrong with the lifetime of the rmq broker during the testing?

[Lance-Drane]

Besides that, the failing tests related to the channel stuff are the only blockers right now.

Is this a "false negative" test? Something wrong with the lifetime of the rmq broker during the testing?

I looked into it and what seemed to be happening is that a message on the AMQP channel would sometimes be published before the exchange was declared. This was specifically the "lifecycle startup" message being sent by the Service. This should be fixed with my most recent commit (which shows all tests passing), this was indeed an inconsistent error due to its asynchronous nature and the fact that three of the integration tests passed.

In theory this could happen with Clients as well, if they are publishing initial messages (if they are only waiting on events, this exception would never happen; only thing to be concerned about is whether or not the Client would get the event message). I had thought this was the original error before looking at the test exception stack trace (I missed it when developing because I wasn't arbitrarily removing exchanges at the time).

[marshallmcdonnell]

Awesome work! Thanks!