def register(callback, agent_type):
    """Subscribe callback to init event for the specified agent.

    :param agent_type: an agent type as defined in neutron_lib.constants.
    :param callback: a callback that can process the agent init event.
    """
    registry.subscribe(callback, agent_type, events.AFTER_INIT)

def register():
    """Register the driver."""
    global DRIVER
    DRIVER = OVSDriver.create()
    # To set the bridge_name in a parent port's vif_details.
    registry.subscribe(vif_details_bridge_name_handler, agent_consts.OVS_BRIDGE_NAME, events.BEFORE_READ)
    LOG.debug("Open vSwitch trunk driver registered")

 def test_security_group_precommit_create_event_fail(self):
     registry.subscribe(fake_callback, resources.SECURITY_GROUP, events.PRECOMMIT_CREATE)
     with mock.patch.object(sqlalchemy.orm.session.SessionTransaction, "rollback") as mock_rollback:
         self.assertRaises(
             securitygroup.SecurityGroupConflict, self.mixin.create_security_group, self.ctx, FAKE_SECGROUP
         )
         self.assertTrue(mock_rollback.called)

def subscribe():
    registry.subscribe(_update_segment_host_mapping_for_agent,
                       resources.AGENT,
                       events.AFTER_CREATE)
    registry.subscribe(_update_segment_host_mapping_for_agent,
                       resources.AGENT,
                       events.AFTER_UPDATE)

    def __init__(self):
        # FIXME(jamielennox): A notifier is being created for each Controller
        # and each Notifier is handling it's own auth. That means that we are
        # authenticating the exact same thing len(controllers) times. This
        # should be an easy thing to optimize.
        # FIXME(kevinbenton): remove this comment and the one above once the
        # switch to pecan is complete since only one notifier is constructed
        # in the pecan notification hook.
        auth = ks_loading.load_auth_from_conf_options(cfg.CONF, 'nova')

        session = ks_loading.load_session_from_conf_options(
            cfg.CONF,
            'nova',
            auth=auth)

        extensions = [
            ext for ext in nova_client.discover_extensions(NOVA_API_VERSION)
            if ext.name == "server_external_events"]
        self.nclient = nova_client.Client(
            NOVA_API_VERSION,
            session=session,
            region_name=cfg.CONF.nova.region_name,
            endpoint_type=cfg.CONF.nova.endpoint_type,
            extensions=extensions)
        self.batch_notifier = batch_notifier.BatchNotifier(
            cfg.CONF.send_events_interval, self.send_events)

        # register callbacks for events pertaining resources affecting Nova
        callback_resources = (
            resources.FLOATING_IP,
            resources.PORT,
        )
        for resource in callback_resources:
            registry.subscribe(self._send_nova_notification,
                               resource, events.BEFORE_RESPONSE)

 def setUp(self):
     super(TestStatusBarriers, self).setUp()
     self.ctx = n_ctx.get_admin_context()
     self.provisioned = mock.Mock()
     self.port = self._make_port()
     registry.subscribe(self.provisioned, resources.PORT,
                        pb.PROVISIONING_COMPLETE)

 def __init__(self, l3_agent):
     self.metadata_port = l3_agent.conf.metadata_port
     self.metadata_access_mark = l3_agent.conf.metadata_access_mark
     registry.subscribe(
         after_router_added, resources.ROUTER, events.AFTER_CREATE)
     registry.subscribe(
         before_router_removed, resources.ROUTER, events.BEFORE_DELETE)

    def setup_sg_rpc_callbacks(self):
        # following way to register call back functions start in kilo
        self._create_sg_f = self.bsn_create_sg_callback
        self._delete_sg_f = self.bsn_delete_sg_callback
        self._update_sg_f = self.bsn_update_sg_callback
        self._create_sg_rule_f = self.bsn_create_sg_rule_callback
        self._delete_sg_rule_f = self.bsn_delete_sg_rule_callback
        registry.subscribe(self._create_sg_f,
                           resources.SECURITY_GROUP, events.AFTER_CREATE)
        registry.subscribe(self._delete_sg_f,
                           resources.SECURITY_GROUP, events.AFTER_DELETE)
        registry.subscribe(self._update_sg_f,
                           resources.SECURITY_GROUP, events.AFTER_UPDATE)
        registry.subscribe(self._create_sg_rule_f,
                           resources.SECURITY_GROUP_RULE, events.AFTER_CREATE)
        registry.subscribe(self._delete_sg_rule_f,
                           resources.SECURITY_GROUP_RULE, events.AFTER_DELETE)

        # the above does not cover the cases where security groups are
        # initially created or when they are deleted since those actions
        # aren't needed by the L2 agent. In order to receive those, we
        # subscribe to the notifications topic that receives all of the
        # API create/update/delete events.
        # Notifications are published at the 'info' level so they will result
        # in a call to the 'info' function below. From there we can check
        # the event type and determine what to do from there.
        target = oslo_messaging.Target(topic='#',
                                       server=cfg.CONF.host)
        keystone_target = oslo_messaging.Target(
             topic='#', exchange='keystone', server=cfg.CONF.host)
        self.listener = oslo_messaging.get_notification_listener(
            n_rpc.TRANSPORT, [target, keystone_target], [self],
            executor='eventlet', allow_requeue=False)
        self.listener.start()

    def subscribe(self):
        # REVISIT(Ryu): Keep an extra set of strong references to the callback
        # methods so that they are not prematurely garbage collected.
        # This hack is required in the Liberty branch (not stable/kilo) because
        # currently the Liberty mechanism driver is expected to work with Kilo
        # ML2 plugin for MidoNet, and in Kilo ML2, there is a bug in which
        # these methods were referenced as weakrefs where a garbage collection
        # could remove these callbacks if they were part of an object.

        self._create_sg_f = self.create_security_group
        self._update_sg_f = self.update_security_group
        self._delete_sg_f = self.delete_security_group
        self._create_sgr_f = self.create_security_group_rule
        self._delete_sgr_f = self.delete_security_group_rule

        registry.subscribe(
            self._create_sg_f, resources.SECURITY_GROUP, events.AFTER_CREATE)
        registry.subscribe(
            self._update_sg_f, resources.SECURITY_GROUP, events.AFTER_UPDATE)
        registry.subscribe(
            self._delete_sg_f, resources.SECURITY_GROUP, events.AFTER_DELETE)
        registry.subscribe(
            self._create_sgr_f, resources.SECURITY_GROUP_RULE,
            events.AFTER_CREATE)
        registry.subscribe(
            self._delete_sgr_f, resources.SECURITY_GROUP_RULE,
            events.AFTER_DELETE)

 def __init__(self):
     db_base_plugin_v2.NeutronDbPluginV2.register_dict_extend_funcs(
         attributes.PORTS, [_extend_port_trunk_details])
     self._segmentation_types = {}
     registry.subscribe(rules.enforce_port_deletion_rules,
                        resources.PORT, events.BEFORE_DELETE)
     registry.notify(constants.TRUNK_PLUGIN, events.AFTER_INIT, self)
     LOG.debug('Trunk plugin loaded')

 def test_treat_devices_removed_notify(self):
     handler = mock.Mock()
     registry.subscribe(handler, resources.PORT_DEVICE, events.AFTER_DELETE)
     devices = [DEVICE_1]
     self.agent.treat_devices_removed(devices)
     handler.assert_called_once_with(mock.ANY, mock.ANY, self.agent,
                                     context=mock.ANY, device=DEVICE_1,
                                     port_id=mock.ANY)

    def __init__(self, trunk_manager):
        self._context = n_context.get_admin_context_without_session()
        self.trunk_manager = trunk_manager
        self.trunk_rpc = agent.TrunkStub()

        registry.subscribe(self.process_trunk_port_events,
                           ovs_agent_constants.OVSDB_RESOURCE,
                           events.AFTER_READ)

 def __init__(self, topic=topics.DHCP_AGENT, plugin=None):
     self._plugin = plugin
     target = oslo_messaging.Target(topic=topic, version='1.0')
     self.client = n_rpc.get_client(target)
     # register callbacks for router interface changes
     registry.subscribe(self._after_router_interface_created,
                        resources.ROUTER_INTERFACE, events.AFTER_CREATE)
     registry.subscribe(self._after_router_interface_deleted,
                        resources.ROUTER_INTERFACE, events.AFTER_DELETE)

    def __init__(self, service_plugin):
        super(BaGPipeBGPVPNDriver, self).__init__(service_plugin)

        self.agent_rpc = rpc_client.BGPVPNAgentNotifyApi()

        registry.subscribe(self.registry_port_updated, resources.PORT,
                           events.AFTER_UPDATE)

        registry.subscribe(self.registry_port_deleted, resources.PORT,
                           events.AFTER_DELETE)

 def test_security_group_precommit_delete_event_fail(self):
     registry.subscribe(fake_callback, resources.SECURITY_GROUP,
                        events.PRECOMMIT_DELETE)
     sg_dict = self.mixin.create_security_group(self.ctx, FAKE_SECGROUP)
     with mock.patch.object(sqlalchemy.orm.session.SessionTransaction,
                           'rollback') as mock_rollback:
         self.assertRaises(securitygroup.SecurityGroupInUse,
                           self.mixin.delete_security_group,
                           self.ctx, sg_dict['id'])
         self.assertTrue(mock_rollback.called)

 def __init__(self):
     super(OVNPlugin, self).__init__()
     LOG.info(_("Starting OVNPlugin"))
     self.vif_type = portbindings.VIF_TYPE_OVS
     # When set to True, Nova plugs the VIF directly into the ovs bridge
     # instead of using the hybrid mode.
     self.vif_details = {portbindings.CAP_PORT_FILTER: True}
     registry.subscribe(self.post_fork_initialize, resources.PROCESS, events.AFTER_CREATE)
     self._setup_dhcp()
     self._start_rpc_notifiers()

    def register_legacy_notification_callbacks(self, legacy_interface):
        """Emulates the server-side notifications from ml2 AgentNotifierApi.

        legacy_interface is an object with 'delete'/'update' methods for
        core resources.
        """
        self._legacy_interface = legacy_interface
        for e in (callback_events.AFTER_UPDATE, callback_events.AFTER_DELETE):
            for r in (resources.PORT, resources.NETWORK):
                registry.subscribe(self._legacy_notifier, r, e)

 def test_record_resource_delete_ignores_dups(self):
     received_kw = []
     receiver = lambda *a, **k: received_kw.append(k)
     registry.subscribe(receiver, 'goose', events.AFTER_DELETE)
     self.rcache.record_resource_delete(self.ctx, 'goose', 3)
     self.assertEqual(1, len(received_kw))
     self.rcache.record_resource_delete(self.ctx, 'goose', 4)
     self.assertEqual(2, len(received_kw))
     self.rcache.record_resource_delete(self.ctx, 'goose', 3)
     self.assertEqual(2, len(received_kw))

 def test_adding_component_for_new_resource_type(self):
     provisioned = mock.Mock()
     registry.subscribe(provisioned, 'NETWORK', pb.PROVISIONING_COMPLETE)
     net = self._make_net()
     # expect failed because the model was not registered for the type
     with testtools.ExpectedException(RuntimeError):
         pb.add_provisioning_component(self.ctx, net.id, 'NETWORK', 'ent')
     pb.add_model_for_resource('NETWORK', models_v2.Network)
     pb.add_provisioning_component(self.ctx, net.id, 'NETWORK', 'ent')
     pb.provisioning_complete(self.ctx, net.id, 'NETWORK', 'ent')
     self.assertTrue(provisioned.called)

 def register(self, resource, event, trigger, **kwargs):
     super(OVNTrunkDriver, self).register(
         resource, event, trigger, **kwargs)
     self._handler = OVNTrunkHandler(self.plugin_driver)
     for event in (events.AFTER_CREATE, events.AFTER_DELETE):
         registry.subscribe(self._handler.trunk_event,
                            trunk_consts.TRUNK,
                            event)
         registry.subscribe(self._handler.subport_event,
                            trunk_consts.SUBPORTS,
                            event)