def record_resource_update(self, context, rtype, resource):
        """Takes in an OVO and generates an event on relevant changes.

        A change is deemed to be relevant if it is not stale and if any
        fields changed beyond the revision number and update time.

        Both creates and updates are handled in this function.
        """
        if self._is_stale(rtype, resource):
            LOG.debug("Ignoring stale update for %s: %s", rtype, resource)
            return
        existing = self._type_cache(rtype).get(resource.id)
        self._type_cache(rtype)[resource.id] = resource
        changed_fields = self._get_changed_fields(existing, resource)
        if not changed_fields:
            LOG.debug("Received resource %s update without any changes: %s",
                      rtype, resource.id)
            return
        if existing:
            LOG.debug("Resource %s %s updated (revision_number %s->%s). "
                      "Old fields: %s New fields: %s",
                      rtype, existing.id, existing.revision_number,
                      resource.revision_number,
                      {f: existing.get(f) for f in changed_fields},
                      {f: resource.get(f) for f in changed_fields})
        else:
            LOG.debug("Received new resource %s: %s", rtype, resource)
        # local notification for agent internals to subscribe to
        registry.notify(rtype, events.AFTER_UPDATE, self,
                        context=context, changed_fields=changed_fields,
                        existing=existing, updated=resource,
                        resource_id=resource.id)

    def _update(self, request, id, body, **kwargs):
        body = Controller.prepare_request_body(request.context,
                                               body, False,
                                               self._resource, self._attr_info,
                                               allow_bulk=self._allow_bulk)
        action = self._plugin_handlers[self.UPDATE]
        # Load object to check authz
        # but pass only attributes in the original body and required
        # by the policy engine to the policy 'brain'
        field_list = [name for (name, value) in self._attr_info.items()
                      if (value.get('required_by_policy') or
                          value.get('primary_key') or
                          'default' not in value)]
        # Ensure policy engine is initialized
        policy.init()
        parent_id = kwargs.get(self._parent_id_name)
        orig_obj = self._item(request, id, field_list=field_list,
                              parent_id=parent_id)
        orig_object_copy = copy.copy(orig_obj)
        orig_obj.update(body[self._resource])
        # Make a list of attributes to be updated to inform the policy engine
        # which attributes are set explicitly so that it can distinguish them
        # from the ones that are set to their default values.
        orig_obj[n_const.ATTRIBUTES_TO_UPDATE] = body[self._resource].keys()
        try:
            policy.enforce(request.context,
                           action,
                           orig_obj,
                           pluralized=self._collection)
        except oslo_policy.PolicyNotAuthorized:
            # To avoid giving away information, pretend that it
            # doesn't exist if policy does not authorize SHOW
            with excutils.save_and_reraise_exception() as ctxt:
                if not policy.check(request.context,
                                    self._plugin_handlers[self.SHOW],
                                    orig_obj,
                                    pluralized=self._collection):
                    ctxt.reraise = False
            msg = _('The resource could not be found.')
            raise webob.exc.HTTPNotFound(msg)

        obj_updater = getattr(self._plugin, action)
        kwargs = {self._resource: body}
        if parent_id:
            kwargs[self._parent_id_name] = parent_id
        obj = obj_updater(request.context, id, **kwargs)
        # Usually an update operation does not alter resource usage, but as
        # there might be side effects it might be worth checking for changes
        # in resource usage here as well (e.g: a tenant port is created when a
        # router interface is added)
        resource_registry.set_resources_dirty(request.context)

        result = {self._resource: self._view(request.context, obj)}
        notifier_method = self._resource + '.update.end'
        self._notifier.info(request.context, notifier_method, result)
        registry.notify(self._resource, events.BEFORE_RESPONSE, self,
                        context=request.context, data=result,
                        method_name=notifier_method, action=action,
                        original=orig_object_copy)
        return result

def _update_segment_host_mapping_for_agent(resource, event, trigger,
                                           payload=None):
    plugin = payload.metadata.get('plugin')
    agent = payload.desired_state
    host = payload.metadata.get('host')
    context = payload.context

    check_segment_for_agent = getattr(plugin, 'check_segment_for_agent', None)
    if (not check_user_configured_segment_plugin() or
            not check_segment_for_agent):
        return
    phys_nets = _get_phys_nets(agent)
    if not phys_nets:
        return
    start_flag = agent.get('start_flag', None)
    if host in reported_hosts and not start_flag:
        return
    reported_hosts.add(host)
    segments = get_segments_with_phys_nets(context, phys_nets)
    current_segment_ids = {
        segment['id'] for segment in segments
        if check_segment_for_agent(segment, agent)}
    update_segment_host_mapping(context, host, current_segment_ids)
    registry.notify(resources.SEGMENT_HOST_MAPPING, events.AFTER_CREATE,
                    plugin, context=context, host=host,
                    current_segment_ids=current_segment_ids)

 def _process_added_router(self, router):
     self._router_added(router['id'], router)
     ri = self.router_info[router['id']]
     ri.router = router
     ri.process()
     registry.notify(resources.ROUTER, events.AFTER_CREATE, self, router=ri)
     self.l3_ext_manager.add_router(self.context, router)

 def test_subscribe_address_scope_of_subnetpool(self, gp):
     l3_db.L3RpcNotifierMixin()
     registry.notify(resources.SUBNETPOOL_ADDRESS_SCOPE,
                     events.AFTER_UPDATE, mock.ANY,
                     context=mock.MagicMock(),
                     subnetpool_id='fake_id')
     self.assertTrue(gp.return_value.notify_routers_updated.called)

    def _setup_test_create_floatingip(
        self, fip, floatingip_db, router_db):
        port = {
            'id': '1234',
            portbindings.HOST_ID: 'myhost',
            'network_id': 'external_net'
        }

        with mock.patch.object(self.mixin, 'get_router') as grtr,\
                mock.patch.object(self.mixin,
                                  '_get_dvr_service_port_hostid') as vmp,\
                mock.patch.object(
                    self.mixin,
                    '_get_dvr_migrating_service_port_hostid'
                                 ) as mvmp,\
                mock.patch.object(
                    self.mixin,
                    'create_fip_agent_gw_port_if_not_exists') as c_fip,\
                mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin,
                                  '_update_fip_assoc'):
            grtr.return_value = router_db
            vmp.return_value = 'my-host'
            mvmp.return_value = 'my-future-host'
            registry.notify(resources.FLOATING_IP, events.AFTER_UPDATE, self,
                            context=mock.Mock(), router_id=router_db['id'],
                            fixed_port_id=port['id'], floating_ip_id=fip['id'],
                            floating_network_id=fip['floating_network_id'],
                            fixed_ip_address='1.2.3.4')
            return c_fip

    def _validate_router_migration(self, context, router_db, router_res):
        """Allow transition only when admin_state_up=False"""
        original_distributed_state = router_db.extra_attributes.distributed
        requested_distributed_state = router_res.get('distributed', None)

        distributed_changed = (
            requested_distributed_state is not None and
            requested_distributed_state != original_distributed_state)
        if not distributed_changed:
            return False
        if router_db.admin_state_up:
            msg = _("Cannot change the 'distributed' attribute of active "
                    "routers. Please set router admin_state_up to False "
                    "prior to upgrade")
            raise n_exc.BadRequest(resource='router', msg=msg)

        # Notify advanced services of the imminent state transition
        # for the router.
        try:
            kwargs = {'context': context, 'router': router_db}
            registry.notify(
                resources.ROUTER, events.BEFORE_UPDATE, self, **kwargs)
        except exceptions.CallbackFailure as e:
            # NOTE(armax): preserve old check's behavior
            if len(e.errors) == 1:
                raise e.errors[0].error
            raise l3.RouterInUse(router_id=router_db['id'], reason=e)
        return True

    def _sync_entry(self, context, entry):
        _log_entry(LOG_PROCESSING, entry)
        method, urlpath, to_send = self._json_data(entry)

        # TODO(mkolesni): This logic is weirdly written, need to refactor it.
        try:
            self.client.sendjson(method, urlpath, to_send)
            registry.notify(entry.object_type, odl_const.BEFORE_COMPLETE,
                            self, context=context, operation=entry.operation,
                            row=entry)
            entry_complete(context, entry)
            self._retry_reset()
            _log_entry(LOG_COMPLETED, entry)
        except exceptions.ConnectionError:
            # Don't raise the retry count, just log an error & break
            entry_reset(context, entry)
            LOG.error("Cannot connect to the OpenDaylight Controller,"
                      " will not process additional entries")
            self._retry_sleep()
            return True
        except Exception:
            _log_entry(LOG_ERROR_PROCESSING, entry,
                       log_level=logging.ERROR, exc_info=True)
            entry_update_state_by_retry_count(
                context, entry, self._max_retry_count)

        return False

    def delete_segment(self, context, uuid, for_net_delete=False):
        """Delete an existing segment."""
        segment_dict = self.get_segment(context, uuid)
        # Do some preliminary operations before deleting the segment
        registry.publish(resources.SEGMENT, events.BEFORE_DELETE,
                         self.delete_segment,
                         payload=events.DBEventPayload(
                             context, metadata={
                                 'for_net_delete': for_net_delete},
                             states=(segment_dict,),
                             resource_id=uuid))

        # Delete segment in DB
        with db_api.CONTEXT_WRITER.using(context):
            if not network.NetworkSegment.delete_objects(context, id=uuid):
                raise exceptions.SegmentNotFound(segment_id=uuid)
            # Do some preliminary operations before deleting segment in db
            registry.notify(resources.SEGMENT, events.PRECOMMIT_DELETE,
                            self.delete_segment, context=context,
                            segment=segment_dict)

        registry.publish(resources.SEGMENT, events.AFTER_DELETE,
                         self.delete_segment,
                         payload=events.DBEventPayload(
                             context, states=(segment_dict,),
                             resource_id=uuid))

    def update_security_group(self, context, id, security_group):
        s = security_group['security_group']

        kwargs = {
            'context': context,
            'security_group_id': id,
            'security_group': s,
        }
        self._registry_notify(resources.SECURITY_GROUP, events.BEFORE_UPDATE,
                              exc_cls=ext_sg.SecurityGroupConflict, **kwargs)

        with db_api.CONTEXT_WRITER.using(context):
            sg = self._get_security_group(context, id)
            if sg.name == 'default' and 'name' in s:
                raise ext_sg.SecurityGroupCannotUpdateDefault()
            sg_dict = self._make_security_group_dict(sg)
            kwargs['original_security_group'] = sg_dict
            sg.update_fields(s)
            sg.update()
            sg_dict = self._make_security_group_dict(sg)
            kwargs['security_group'] = sg_dict
            self._registry_notify(
                    resources.SECURITY_GROUP,
                    events.PRECOMMIT_UPDATE,
                    exc_cls=ext_sg.SecurityGroupConflict,
                    payload=events.DBEventPayload(
                        context, request_body=s,
                        states=(kwargs['original_security_group'],),
                        resource_id=id, desired_state=sg_dict))
        registry.notify(resources.SECURITY_GROUP, events.AFTER_UPDATE, self,
                        **kwargs)
        return sg_dict

    def update_security_group(self, context, id, security_group):
        s = security_group['security_group']

        kwargs = {
            'context': context,
            'security_group_id': id,
            'security_group': s,
        }
        self._registry_notify(resources.SECURITY_GROUP, events.BEFORE_UPDATE,
                              exc_cls=ext_sg.SecurityGroupConflict, **kwargs)

        with db_api.context_manager.writer.using(context):
            sg = self._get_security_group(context, id)
            if sg.name == 'default' and 'name' in s:
                raise ext_sg.SecurityGroupCannotUpdateDefault()
            sg_dict = self._make_security_group_dict(sg)
            kwargs['original_security_group'] = sg_dict
            sg.update_fields(s)
            sg.update()
            sg_dict = self._make_security_group_dict(sg)
            kwargs['security_group'] = sg_dict
            self._registry_notify(
                    resources.SECURITY_GROUP,
                    events.PRECOMMIT_UPDATE,
                    exc_cls=ext_sg.SecurityGroupConflict, **kwargs)

        registry.notify(resources.SECURITY_GROUP, events.AFTER_UPDATE, self,
                        **kwargs)
        return sg_dict

 def treat_devices_removed(self, devices):
     resync = False
     self.sg_agent.remove_devices_filter(devices)
     for device in devices:
         LOG.info(_LI("Attachment %s removed"), device)
         details = None
         try:
             details = self.plugin_rpc.update_device_down(self.context,
                                                          device,
                                                          self.agent_id,
                                                          cfg.CONF.host)
         except Exception:
             LOG.exception(_LE("Error occurred while removing port %s"),
                           device)
             resync = True
         if details and details['exists']:
             LOG.info(_LI("Port %s updated."), device)
         else:
             LOG.debug("Device %s not defined on plugin", device)
         port_id = self._clean_network_ports(device)
         self.ext_manager.delete_port(self.context,
                                      {'device': device,
                                       'port_id': port_id})
         registry.notify(local_resources.PORT_DEVICE, events.AFTER_DELETE,
                         self, context=self.context, device=device,
                         port_id=port_id)
     if self.prevent_arp_spoofing:
         self.mgr.delete_arp_spoofing_protection(devices)
     return resync

    def create_or_update_agent(self, context, agent_state):
        """Registers new agent in the database or updates existing.

        Returns tuple of agent status and state.
        Status is from server point of view: alive, new or revived.
        It could be used by agent to do some sync with the server if needed.
        """
        status = agent_consts.AGENT_ALIVE
        with context.session.begin(subtransactions=True):
            res_keys = ['agent_type', 'binary', 'host', 'topic']
            res = dict((k, agent_state[k]) for k in res_keys)
            if 'availability_zone' in agent_state:
                res['availability_zone'] = agent_state['availability_zone']
            configurations_dict = agent_state.get('configurations', {})
            res['configurations'] = jsonutils.dumps(configurations_dict)
            resource_versions_dict = agent_state.get('resource_versions')
            if resource_versions_dict:
                res['resource_versions'] = jsonutils.dumps(
                    resource_versions_dict)
            res['load'] = self._get_agent_load(agent_state)
            current_time = timeutils.utcnow()
            try:
                agent = self._get_agent_by_type_and_host(
                    context, agent_state['agent_type'], agent_state['host'])
                if not agent.is_active:
                    status = agent_consts.AGENT_REVIVED
                    if 'resource_versions' not in agent_state:
                        # updating agent_state with resource_versions taken
                        # from db so that
                        # _update_local_agent_resource_versions() will call
                        # version_manager and bring it up to date
                        agent_state['resource_versions'] = self._get_dict(
                            agent, 'resource_versions', ignore_missing=True)
                res['heartbeat_timestamp'] = current_time
                if agent_state.get('start_flag'):
                    res['started_at'] = current_time
                greenthread.sleep(0)
                self._log_heartbeat(agent_state, agent, configurations_dict)
                agent.update_fields(res)
                agent.update()
                event_type = events.AFTER_UPDATE
            except agent_exc.AgentNotFoundByTypeHost:
                greenthread.sleep(0)
                res['created_at'] = current_time
                res['started_at'] = current_time
                res['heartbeat_timestamp'] = current_time
                res['admin_state_up'] = cfg.CONF.enable_new_agents
                agent = agent_obj.Agent(context=context, **res)
                greenthread.sleep(0)
                agent.create()
                event_type = events.AFTER_CREATE
                self._log_heartbeat(agent_state, agent, configurations_dict)
                status = agent_consts.AGENT_NEW
            greenthread.sleep(0)

        registry.notify(resources.AGENT, event_type, self, context=context,
                        host=agent_state['host'], plugin=self,
                        agent=agent_state)
        return status, agent_state

 def create_bgp_speaker(self, context, bgp_speaker):
     bgp_speaker = super(BgpPlugin, self).create_bgp_speaker(context,
                                                             bgp_speaker)
     payload = {'plugin': self, 'context': context,
                'bgp_speaker': bgp_speaker}
     registry.notify(dr_resources.BGP_SPEAKER, events.AFTER_CREATE,
                     self, payload=payload)
     return bgp_speaker

 def _process_updated_router(self, router):
     ri = self.router_info[router['id']]
     ri.router = router
     registry.notify(resources.ROUTER, events.BEFORE_UPDATE,
                     self, router=ri)
     ri.process()
     registry.notify(resources.ROUTER, events.AFTER_UPDATE, self, router=ri)
     self.l3_ext_manager.update_router(self.context, router)

 def test_vif_details_bridge_name_handler_registration(self,
                                                       mock_gen_br_name):
     driver.register()
     mock_gen_br_name.return_value = 'fake-trunk-br-name'
     test_trigger = mock.Mock()
     registry.notify(agent_consts.OVS_BRIDGE_NAME, events.BEFORE_READ,
                     test_trigger, **{'port': {'trunk_details':
                                               {'trunk_id': 'foo'}}})
     test_trigger.assert_called_once_with('fake-trunk-br-name')

 def _call_before_port_binding(self, host):
     kwargs = {
         'context': mock.Mock(),
         'port': {
             portbindings.HOST_ID: host
         }
     }
     registry.notify(resources.PORT, events.BEFORE_CREATE, self.ml2_plugin,
                     **kwargs)

 def _clear_router_provider(self, resource, event, trigger, context,
                            router_id, **kwargs):
     """Remove the association between a router and a service provider."""
     drv = self.get_provider_for_router(context, router_id)
     registry.notify(
         resources.ROUTER_CONTROLLER, events.PRECOMMIT_DELETE_ASSOCIATIONS,
         trigger, context=context, router_id=router_id,
         old_driver=drv, new_driver=None, **kwargs)
     self._stm.del_resource_associations(context, [router_id])

    def test_handle_event(self, mock_sg_cb):
        fake_register()
        self.driver_manager.register_driver(FAKE_DRIVER)

        registry.notify(
            resources.SECURITY_GROUP_RULE, events.AFTER_CREATE, mock.ANY)
        mock_sg_cb.assert_called_once_with(
            resources.SECURITY_GROUP_RULE, events.AFTER_CREATE, mock.ANY)
        mock_sg_cb.reset_mock()
        registry.notify('fake_resource', events.AFTER_DELETE, mock.ANY)
        mock_sg_cb.assert_not_called()

 def _registry_notify(self, res, event, id=None, exc_cls=None, **kwargs):
     # NOTE(armax): a callback exception here will prevent the request
     # from being processed. This is a hook point for backend's validation;
     # we raise to propagate the reason for the failure.
     try:
         registry.notify(res, event, self, **kwargs)
     except exceptions.CallbackFailure as e:
         if exc_cls:
             reason = (_('cannot perform %(event)s due to %(reason)s') %
                       {'event': event, 'reason': e})
             raise exc_cls(reason=reason, id=id)