def associate_floatingip_router(self, context, floatingip_id, router_id):
     with context.session.begin(subtransactions=True):
         floatingip_db = self._get_floatingip(context,
                                              floatingip_id)
         floating_network_id = floatingip_db['floating_network_id']
         gw_portdb = None
         try:
             port_qry = context.elevated().session.query(models_v2.Port)
             gw_portdb = port_qry.filter_by(
                 network_id=floating_network_id,
                 device_id=router_id,
                 device_owner=DEVICE_OWNER_ROUTER_GW).one()
         except exc.NoResultFound:
             raise uosfloatingip.GWPortForFloatingIPNotFound(id=router_id)
         tenant_id = floatingip_db.tenant_id
         _ctx = n_context.Context('', tenant_id)
         body = {'floatingip': {'port_id': gw_portdb['id']}}
         payload = {'id': floatingip_db.id}
         payload.update(body)
         self._notifier = n_rpc.get_notifier('network')
         _notifer = n_rpc.get_notifier('network')
         _notifer.info(_ctx, 'floatingip.update.start', payload)
         result = self.update_floatingip(context, floatingip_id, body)
         _notifer.info(_ctx, 'floatingip.update.end',
                      {'floatingip': result})
         return result

 def _notify_new_addresses(self, context, new_addresses):
     for addr in new_addresses:
         payload = dict(used_by_tenant_id=addr["used_by_tenant_id"],
                        ip_block_id=addr["subnet_id"],
                        ip_address=addr["address_readable"],
                        device_ids=[p["device_id"] for p in addr["ports"]],
                        created_at=addr["created_at"])
         n_rpc.get_notifier("network").info(context,
                                            "ip_block.address.create",
                                            payload)

 def deallocate_ip_address(self, context, address):
     address["deallocated"] = 1
     payload = dict(used_by_tenant_id=address["used_by_tenant_id"],
                    ip_block_id=address["subnet_id"],
                    ip_address=address["address_readable"],
                    device_ids=[p["device_id"] for p in address["ports"]],
                    created_at=address["created_at"],
                    deleted_at=timeutils.utcnow())
     n_rpc.get_notifier("network").info(context,
                                        "ip_block.address.delete",
                                        payload)

    def update_floatingip_ratelimit(self, request, id, body):
        context = request.context
        router = self._get_service_plugin(constants.L3_ROUTER_NAT)
        # NOTE(gongysh) to fix the privilege problem
        router.get_floatingip(context, id)
        try:
            payload = body.copy()
            rate_limit = payload.get('rate_limit')
            rate_limit = int(rate_limit)
        except (AttributeError, ValueError, TypeError):
            msg = _("Invalid format: %s") % request.body
            raise qexception.BadRequest(resource='body', msg=msg)
        payload['id'] = id
        _notifier = n_rpc.get_notifier('network')
        _notifier.info(context, 'floatingip.update_ratelimit.start', payload)
        with context.session.begin(subtransactions=True):
            try:
                fip_qry = context.session.query(l3_db.FloatingIP)
                floating_ip = fip_qry.filter_by(id=id).one()
                floating_ip.update({'rate_limit': body['rate_limit']})
            except sa_exc.NoResultFound:
                raise l3.FloatingIPNotFound(floatingip_id=id)

        router_id = floating_ip['router_id']
        if router_id:
            router.l3_rpc_notifier.routers_updated(context, [router_id])
        result = router._make_floatingip_dict(floating_ip)
        _notifier.info(context, 'floatingip.update_ratelimit.end',
                       {'floatingip': result})
        return result

 def update_floatingip_registerno(self, request, id, body):
     context = request.context
     router = self._get_service_plugin(constants.L3_ROUTER_NAT)
     # NOTE(gongysh) to fix the privilege problem
     router.get_floatingip(context, id)
     _notifier = n_rpc.get_notifier('network')
     payload = body.copy()
     registerno = payload.get('uos_registerno', None)
     if registerno is None:
         msg = _("Invalid format: %s") % request.body
         raise qexception.BadRequest(resource='body', msg=msg)
     payload['id'] = id
     _notifier.info(context, 'floatingip.update_registerno.start', payload)
     # NOTE(gongysh) do we need to validate it?
     with context.session.begin(subtransactions=True):
         try:
             fip_qry = context.session.query(l3_db.FloatingIP)
             floating_ip = fip_qry.filter_by(id=id).one()
         except sa_exc.NoResultFound:
             raise l3.FloatingIPNotFound(floatingip_id=id)
         floating_ip.update({'uos_registerno': registerno.strip()})
     result = router._make_floatingip_dict(floating_ip)
     _notifier.info(context, 'floatingip.update_registerno.end',
                    {'floatingip': result})
     return result

    def remove_router_interface(self, context, router_id, interface_info):
        if not interface_info:
            msg = _("Either subnet_id or port_id must be specified")
            raise n_exc.BadRequest(resource='router', msg=msg)
        port_id = interface_info.get('port_id')
        subnet_id = interface_info.get('subnet_id')
        device_owner = self._get_device_owner(context, router_id)
        if port_id:
            port, subnet = self._remove_interface_by_port(context, router_id,
                                                          port_id, subnet_id,
                                                          device_owner)
        elif subnet_id:
            port, subnet = self._remove_interface_by_subnet(
                context, router_id, subnet_id, device_owner)

        self.l3_rpc_notifier.routers_updated(
            context, [router_id], 'remove_router_interface')
        info = {'id': router_id,
                'tenant_id': port['tenant_id'],
                'port_id': port['id'],
                'subnet_id': subnet['id']}
        notifier = n_rpc.get_notifier('network')
        notifier.info(
            context, 'router.interface.delete', {'router_interface': info})
        return info

    def notify_router_interface_action(self, context, router_interface_info, routers, action):
        l3_method = "%s_router_interface" % action
        self.l3_cfg_rpc_notifier.routers_updated(context, routers, l3_method)

        mapping = {"add": "create", "remove": "delete"}
        notifier = n_rpc.get_notifier("network")
        router_event = "router.interface.%s" % mapping[action]
        notifier.info(context, router_event, {"router_interface": router_interface_info})

def notify_tag_action(context, action, parent, parent_id, tags=None):
    notifier = n_rpc.get_notifier('network')
    tag_event = 'tag.%s' % action
    # TODO(hichihara): Add 'updated_at' into payload
    payload = {'parent_resource': parent,
               'parent_resource_id': parent_id}
    if tags is not None:
        payload['tags'] = tags
    notifier.info(context, tag_event, payload)

    def __init__(
        self,
        plugin,
        collection,
        resource,
        attr_info,
        allow_bulk=False,
        member_actions=None,
        parent=None,
        allow_pagination=False,
        allow_sorting=False,
    ):
        if member_actions is None:
            member_actions = []
        self._plugin = plugin
        self._collection = collection.replace("-", "_")
        self._resource = resource.replace("-", "_")
        self._attr_info = attr_info
        self._allow_bulk = allow_bulk
        self._allow_pagination = allow_pagination
        self._allow_sorting = allow_sorting
        self._native_bulk = self._is_native_bulk_supported()
        self._native_pagination = self._is_native_pagination_supported()
        self._native_sorting = self._is_native_sorting_supported()
        self._policy_attrs = [name for (name, info) in self._attr_info.items() if info.get("required_by_policy")]
        self._notifier = n_rpc.get_notifier("network")
        # use plugin's dhcp notifier, if this is already instantiated
        agent_notifiers = getattr(plugin, "agent_notifiers", {})
        self._dhcp_agent_notifier = (
            agent_notifiers.get(const.AGENT_TYPE_DHCP) or dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
        )
        if cfg.CONF.notify_nova_on_port_data_changes:
            from neutron.notifiers import nova

            self._nova_notifier = nova.Notifier()
        self._member_actions = member_actions
        self._primary_key = self._get_primary_key()
        if self._allow_pagination and self._native_pagination:
            # Native pagination need native sorting support
            if not self._native_sorting:
                raise exceptions.Invalid(_("Native pagination depend on native sorting"))
            if not self._allow_sorting:
                LOG.info(_("Allow sorting is enabled because native " "pagination requires native sorting"))
                self._allow_sorting = True

        if parent:
            self._parent_id_name = "%s_id" % parent["member_name"]
            parent_part = "_%s" % parent["member_name"]
        else:
            self._parent_id_name = None
            parent_part = ""
        self._plugin_handlers = {
            self.LIST: "get%s_%s" % (parent_part, self._collection),
            self.SHOW: "get%s_%s" % (parent_part, self._resource),
        }
        for action in [self.CREATE, self.UPDATE, self.DELETE]:
            self._plugin_handlers[action] = "%s%s_%s" % (action, parent_part, self._resource)

    def notify_router_interface_action(
            self, context, router_interface_info, routers, action):
        l3_method = '%s_router_interface' % action
        self.l3_cfg_rpc_notifier.routers_updated(context, routers, l3_method)

        mapping = {'add': 'create', 'remove': 'delete'}
        notifier = n_rpc.get_notifier('network')
        router_event = 'router.interface.%s' % mapping[action]
        notifier.info(context, router_event,
                      {'router_interface': router_interface_info})

    def test_get_notifier_null_publisher(self):
        rpc.NOTIFIER = mock.Mock()
        mock_prep = mock.Mock()
        mock_prep.return_value = 'notifier'
        rpc.NOTIFIER.prepare = mock_prep

        notifier = rpc.get_notifier('service', host='bar')

        mock_prep.assert_called_once_with(publisher_id='service.bar')
        self.assertEqual('notifier', notifier)

def delete_subnet(context, id):
    """Delete a subnet.

    : param context: neutron api request context
    : param id: UUID representing the subnet to delete.
    """
    LOG.info("delete_subnet %s for tenant %s" % (id, context.tenant_id))
    with context.session.begin():
        subnet = db_api.subnet_find(context, id=id, scope=db_api.ONE)
        if not subnet:
            raise exceptions.SubnetNotFound(subnet_id=id)

        payload = dict(tenant_id=subnet["tenant_id"],
                       ip_block_id=subnet["id"],
                       created_at=subnet["created_at"],
                       deleted_at=timeutils.utcnow())

        _delete_subnet(context, subnet)

        n_rpc.get_notifier("network").info(context, "ip_block.delete", payload)

def do_notify(context, event_type, payload):
    """Generic Notifier.

    Parameters:
        - `context`: session context
        - `event_type`: the event type to report, i.e. ip.usage
        - `payload`: dict containing the payload to send
    """
    LOG.debug('IP_BILL: notifying {}'.format(payload))

    notifier = n_rpc.get_notifier('network')
    notifier.info(context, event_type, payload)

 def __init__(
     self,
     plugin,
     collection,
     resource,
     attr_info,
     allow_bulk=False,
     member_actions=None,
     parent=None,
     allow_pagination=False,
     allow_sorting=False,
 ):
     if member_actions is None:
         member_actions = []
     self._plugin = plugin
     self._collection = collection.replace("-", "_")
     self._resource = resource.replace("-", "_")
     self._attr_info = attr_info
     self._allow_bulk = allow_bulk
     self._allow_pagination = allow_pagination
     self._allow_sorting = allow_sorting
     self._native_bulk = self._is_native_bulk_supported()
     self._native_pagination = self._is_native_pagination_supported()
     self._native_sorting = self._is_native_sorting_supported()
     self._policy_attrs = [name for (name, info) in self._attr_info.items() if info.get("required_by_policy")]
     self._notifier = n_rpc.get_notifier("network")
     self._member_actions = member_actions
     self._primary_key = self._get_primary_key()
     if self._allow_pagination and self._native_pagination:
         # Native pagination need native sorting support
         if not self._native_sorting:
             raise exceptions.Invalid(_("Native pagination depend on native sorting"))
         if not self._allow_sorting:
             LOG.info(_LI("Allow sorting is enabled because native " "pagination requires native sorting"))
             self._allow_sorting = True
     self.parent = parent
     if parent:
         self._parent_id_name = "%s_id" % parent["member_name"]
         parent_part = "_%s" % parent["member_name"]
     else:
         self._parent_id_name = None
         parent_part = ""
     self._plugin_handlers = {
         self.LIST: "get%s_%s" % (parent_part, self._collection),
         self.SHOW: "get%s_%s" % (parent_part, self._resource),
     }
     for action in [self.CREATE, self.UPDATE, self.DELETE]:
         self._plugin_handlers[action] = "%s%s_%s" % (action, parent_part, self._resource)

    def notify_router_interface_action(
            self, context, router_id, tenant_id, port_id, subnet_id, action):
        l3_method = '%s_router_interface' % action
        self.l3_rpc_notifier.routers_updated(context, [router_id], l3_method)

        mapping = {'add': 'create', 'remove': 'delete'}
        info = {
            'id': router_id,
            'tenant_id': tenant_id,
            'port_id': port_id,
            'subnet_id': subnet_id
        }
        notifier = n_rpc.get_notifier('network')
        router_event = 'router.interface.%s' % mapping[action]
        notifier.info(context, router_event, {'router_interface': info})
        return info

    def __init__(self, plugin, collection, resource, attr_info,
                 allow_bulk=False, member_actions=None, parent=None,
                 allow_pagination=False, allow_sorting=False):
        if member_actions is None:
            member_actions = []
        self._plugin = plugin
        self._collection = collection.replace('-', '_')
        self._resource = resource.replace('-', '_')
        self._attr_info = attr_info
        self._allow_bulk = allow_bulk
        self._allow_pagination = allow_pagination
        self._allow_sorting = allow_sorting
        self._native_bulk = self._is_native_bulk_supported()
        self._native_pagination = self._is_native_pagination_supported()
        self._native_sorting = self._is_native_sorting_supported()
        self._policy_attrs = [name for (name, info) in self._attr_info.items()
                              if info.get('required_by_policy')]
        self._notifier = n_rpc.get_notifier('network')
        if cfg.CONF.notify_nova_on_port_data_changes:
            from neutron.notifiers import nova
            self._nova_notifier = nova.Notifier()
        self._member_actions = member_actions
        self._primary_key = self._get_primary_key()
        if self._allow_pagination and self._native_pagination:
            # Native pagination need native sorting support
            if not self._native_sorting:
                raise exceptions.Invalid(
                    _("Native pagination depend on native sorting")
                )
            if not self._allow_sorting:
                LOG.info(_LI("Allow sorting is enabled because native "
                             "pagination requires native sorting"))
                self._allow_sorting = True

        if parent:
            self._parent_id_name = '%s_id' % parent['member_name']
            parent_part = '_%s' % parent['member_name']
        else:
            self._parent_id_name = None
            parent_part = ''
        self._plugin_handlers = {
            self.LIST: 'get%s_%s' % (parent_part, self._collection),
            self.SHOW: 'get%s_%s' % (parent_part, self._resource)
        }
        for action in [self.CREATE, self.UPDATE, self.DELETE]:
            self._plugin_handlers[action] = '%s%s_%s' % (action, parent_part,
                                                         self._resource)

    def _metering_notification(self):
        for label_id, info in self.metering_infos.items():
            data = {'label_id': label_id,
                    'tenant_id': self.label_tenant_id.get(label_id),
                    'pkts': info['pkts'],
                    'bytes': info['bytes'],
                    'time': info['time'],
                    'first_update': info['first_update'],
                    'last_update': info['last_update'],
                    'host': self.host}

            LOG.debug("Send metering report: %s", data)
            notifier = n_rpc.get_notifier('metering')
            notifier.info(self.context, 'l3.meter', data)
            info['pkts'] = 0
            info['bytes'] = 0
            info['time'] = 0

def main():
    config.init(sys.argv[1:])
    config.setup_logging()

    cxt = context.get_admin_context()
    plugin = manager.NeutronManager.get_plugin()
    notifier = n_rpc.get_notifier('network')
    for network in plugin.get_networks(cxt):
        notifier.info(cxt, 'network.exists', {'network': network})
    for subnet in plugin.get_subnets(cxt):
        notifier.info(cxt, 'subnet.exists', {'subnet': subnet})
    for port in plugin.get_ports(cxt):
        notifier.info(cxt, 'port.exists', {'port': port})
    for router in plugin.get_routers(cxt):
        notifier.info(cxt, 'router.exists', {'router': router})
    for floatingip in plugin.get_floatingips(cxt):
        notifier.info(cxt, 'floatingip.exists', {'floatingip': floatingip})

 def collect_stats_for_monitor(self):
     for loadbalancer_id, driver_name in self.instance_mapping.items():
         driver = self.device_drivers[driver_name]
         tenant_id = self.label_tenant_id.get(loadbalancer_id, None)
         try:
             stats = driver.get_stats(loadbalancer_id)
             if stats and self.conf.statistic_notification_enable: 
                 stats.pop('members', None)
                 stats['tenant_id'] = tenant_id
                 notifier = n_rpc.get_notifier('loadbalancer')
                 notifier.info(self.context, 'loadbalancer.meter', stats)
                 LOG.info(_('collect_stats_for_monitor'
                          'loadbalancer %(id)s stats %(stats)s .'),
                          {'id':loadbalancer_id,'stats':stats})
         except Exception:
             LOG.exception(_('Error updating statistics on loadbalancer %s'),
                           loadbalancer_id)

def main():
    config.init(sys.argv[1:])
    config.setup_logging()

    cxt = context.get_admin_context()
    plugin = manager.NeutronManager.get_plugin()
    l3_plugin = manager.NeutronManager.get_service_plugins().get(constants.L3_ROUTER_NAT)
    notifier = n_rpc.get_notifier("network")
    for network in plugin.get_networks(cxt):
        notifier.info(cxt, "network.exists", {"network": network})
    for subnet in plugin.get_subnets(cxt):
        notifier.info(cxt, "subnet.exists", {"subnet": subnet})
    for port in plugin.get_ports(cxt):
        notifier.info(cxt, "port.exists", {"port": port})
    for router in l3_plugin.get_routers(cxt):
        notifier.info(cxt, "router.exists", {"router": router})
    for floatingip in l3_plugin.get_floatingips(cxt):
        notifier.info(cxt, "floatingip.exists", {"floatingip": floatingip})