Cleanup fanout queues on ovs agent stop (part 2)

As a followup from the previous commit we here now also cleanup the
SubPort an Trunk fanout queues.

Closes-Bug: #1586731
Change-Id: I047603b647dec7787c2471d9edb70fa4ec599a2a

neutron/plugins/ml2/drivers/openvswitch/agent/ovs_capabilities.py

@@ -21,3 +21,9 @@ def register():
     """Register OVS capabilities."""
     # Add capabilities to be loaded during agent initialization
     capabilities.register(driver.init_handler, constants.AGENT_TYPE_OVS)
+
+
+def unregister():
+    """Unregister OVS capabilities."""
+    # To cleanup oslo.messaging queues
+    driver.unregister()

neutron/plugins/ml2/drivers/openvswitch/agent/ovs_neutron_agent.py

@@ -2957,3 +2957,4 @@ def main(bridge_classes):
             LOG.error("%s agent terminated!", e)
             sys.exit(1)
     agent.daemon_loop()
+    ovs_capabilities.unregister()

neutron/services/trunk/drivers/openvswitch/agent/driver.py

@@ -100,3 +100,9 @@ def init_handler(resource, event, trigger, payload=None):
     manager = trunk_manager.TrunkManager(trigger.int_br)
     handler = ovsdb_handler.OVSDBHandler(manager)
     TRUNK_SKELETON = OVSTrunkSkeleton(handler)
+
+
+def unregister():
+    """Cleanup fanout queues"""
+    global TRUNK_SKELETON
+    TRUNK_SKELETON.unregister()

neutron/services/trunk/rpc/agent.py

@@ -53,6 +53,9 @@ class TrunkSkeleton(object):
         self._connection.create_consumer(topic, endpoints, fanout=True)
         self._connection.consume_in_threads()
 
+    def unregister(self):
+        self._connection.close()
+
     @abc.abstractmethod
     def handle_trunks(self, context, resource_type, trunks, event_type):
         """Handle trunk events."""