def _setup_rpc(self):
     # Setup a rpc server
     self.topic = sfc_topics.SFC_PLUGIN
     self.endpoints = [ovs_sfc_rpc.SfcRpcCallback(self)]
     self.conn = n_rpc.create_connection()
     self.conn.create_consumer(self.topic, self.endpoints, fanout=False)
     self.conn.consume_in_threads()

    def __init__(self, agent, host):
        self.host = host
        self.conn = n_rpc.create_connection(new=True)
        context = ctx.get_admin_context_without_session()
        node_topic = '%s.%s' % (topics.CISCO_IPSEC_AGENT_TOPIC, self.host)

        self.service_state = {}

        self.endpoints = [self]
        self.conn.create_consumer(node_topic, self.endpoints, fanout=False)
        self.conn.consume_in_threads()
        self.agent_rpc = (
            CiscoCsrIPsecVpnDriverApi(topics.CISCO_IPSEC_DRIVER_TOPIC, '1.0'))
        self.periodic_report = loopingcall.FixedIntervalLoopingCall(
            self.report_status, context)
        self.periodic_report.start(
            interval=agent.conf.cisco_csr_ipsec.status_check_interval)

        csrs_found = find_available_csrs_from_config(cfg.CONF.config_file)
        if csrs_found:
            LOG.info(_("Loaded %(num)d Cisco CSR configuration%(plural)s"),
                     {'num': len(csrs_found),
                      'plural': 's'[len(csrs_found) == 1:]})
        else:
            raise SystemExit(_('No Cisco CSR configurations found in: %s') %
                             cfg.CONF.config_file)
        self.csrs = dict([(k, csr_client.CsrRestClient(v))
                          for k, v in csrs_found.items()])

 def _start_rpc_listeners(self):
     self.notifier = ovsvapp_rpc.OVSvAppAgentNotifyAPI(topics.AGENT)
     self.endpoints = [ovsvapp_rpc.OVSvAppServerRpcCallback(self.notifier)]
     self.topic = constants.OVSVAPP
     self.conn = n_rpc.create_connection(new=True)
     self.conn.create_consumer(self.topic, self.endpoints, fanout=False)
     return self.conn.consume_in_threads()

    def __init__(self, vpn_service, host):
        super(VyattaIPSecDriver, self).__init__(vpn_service, host)
        self.vpn_service = vpn_service
        self.host = host

        # register RPC endpoint
        conn = n_rpc.create_connection()
        node_topic = '%s.%s' % (topics.BROCADE_IPSEC_AGENT_TOPIC,
                                self.host)

        endpoints = [self.rpc_endpoint_factory(self)]
        conn.create_consumer(node_topic, endpoints, fanout=False)
        conn.consume_in_threads()

        # initialize agent to server RPC link
        self.server_api = NeutronServerAPI(
            topics.BROCADE_IPSEC_DRIVER_TOPIC)

        # initialize VPN service cache (to keep service state)
        self._svc_cache = list()
        self._router_resources_cache = dict()

        # setup periodic task. All periodic task require fully configured
        # device driver. It will be called asynchronously, and soon, so it
        # should be last, when all configuration is done.
        self._periodic_tasks = periodic = _VyattaPeriodicTasks(self)
        loop = loopingcall.DynamicLoopingCall(periodic)
        loop.start(initial_delay=5)

def create_consumers(endpoints, prefix, topic_details):
    """Create agent RPC consumers.

    :param endpoints: The list of endpoints to process the incoming messages.
    :param prefix: Common prefix for the plugin/agent message queues.
    :param topic_details: A list of topics. Each topic has a name, an
                          operation, and an optional host param keying the
                          subscription to topic.host for plugin calls.

    :returns: A common Connection.
    """

    connection = n_rpc.create_connection(new=True)
    for details in topic_details:
        topic, operation, node_name = itertools.islice(
            itertools.chain(details, [None]), 3)

        topic_name = topics.get_topic_name(prefix, topic, operation)
        connection.create_consumer(topic_name, endpoints, fanout=True)
        if node_name:
            node_topic_name = '%s.%s' % (topic_name, node_name)
            connection.create_consumer(node_topic_name,
                                       endpoints,
                                       fanout=False)
    connection.consume_in_threads()
    return connection

 def start_rpc_listeners(self):
     """Configure all listeners here"""
     self._setup_rpc()
     self.conn = n_rpc.create_connection()
     self.conn.create_consumer(CONF.QUARK_ASYNC.topic, self.endpoints,
                               fanout=False)
     return self.conn.consume_in_threads()

 def _start_rpc_listeners(self):
     self.notifier = rk_rpc.RkAgentNotifyAPI(topics.AGENT)
     self.endpoints = [rk_rpc.RkServerRpcCallback(self.notifier)]
     self.topic = rk_const.RATEKEEPER
     self.conn = common_rpc.create_connection(new=True)
     self.conn.create_consumer(self.topic, self.endpoints, fanout=False)
     return self.conn.consume_in_threads()

 def __init__(self):
     # Used to provide trunk lookups for the agent.
     registry.provide(trunk_by_port_provider, resources.TRUNK)
     self._connection = n_rpc.create_connection()
     self._connection.create_consumer(
         constants.TRUNK_BASE_TOPIC, [self], fanout=False)
     self._connection.consume_in_threads()

    def start_rpc_listeners(self):
        self.endpoints = [FirewallCallbacks(self)]

        self.conn = n_rpc.create_connection()
        self.conn.create_consumer(
            topics.FIREWALL_PLUGIN, self.endpoints, fanout=False)
        return self.conn.consume_in_threads()

    def __init__(self, agent, host):
        self.agent = agent
        self.conf = self.agent.conf
        self.root_helper = self.agent.root_helper
        self.context = context.get_admin_context_without_session()
        self.host = host
        self.processes = {}
        self.process_status_cache = {}

        self.updated_vpnservices = set()
        self.deleted_vpnservices = set()
        self.updated_vpnusers = set()

        self.conn = q_rpc.create_connection(new=True)
        self.topic = topics.PPTP_AGENT_TOPIC
        node_topic = '%s.%s' % (self.topic, self.host)
        self.conn.create_consumer(
            node_topic,
            self.create_rpc_dispatcher(),
            fanout=False)

        fan_topic = '%s' % (self.topic)
        self.conn.create_consumer(
            fan_topic,
            self.create_rpc_dispatcher(),
            fanout=True)
        self.agent_rpc = PptpVpnDriverApi(topics.PPTP_DRIVER_TOPIC, '1.0')
        self.conn.consume_in_threads()

 def start_rpc_listeners(self):
     self.conn = n_rpc.create_connection(new=True)
     self.conn.create_consumer(topics.PLUGIN, self.endpoints,
                               fanout=False)
     self.conn.create_consumer(topics.L3PLUGIN, self.endpoints,
                               fanout=False)
     self.conn.consume_in_threads()

 def _setup_rpc(self):
     # RPC support
     self.service_topics = {
         svc_constants.CORE: topics.PLUGIN,
         svc_constants.L3_ROUTER_NAT: topics.L3PLUGIN
     }
     self.rpc_context = n_context.ContextBase(
         'neutron', 'neutron', is_admin=False)
     self.conn = n_rpc.create_connection()
     self.endpoints = [
         BridgeRpcCallbacks(),
         securitygroups_rpc.SecurityGroupServerRpcCallback(),
         dhcp_rpc.DhcpRpcCallback(),
         l3_rpc.L3RpcCallback(),
         agents_db.AgentExtRpcCallback(),
         metadata_rpc.MetadataRpcCallback()
     ]
     for svc_topic in self.service_topics.values():
         self.conn.create_consumer(svc_topic, self.endpoints, fanout=False)
     # Consume from all consumers in threads
     self.conn.consume_in_threads()
     self.notifier = AgentNotifierApi(topics.AGENT)
     self.agent_notifiers[n_const.AGENT_TYPE_DHCP] = (
         dhcp_rpc_agent_api.DhcpAgentNotifyAPI())
     self.agent_notifiers[n_const.AGENT_TYPE_L3] = (
         l3_rpc_agent_api.L3AgentNotifyAPI())

 def _setup_opflex_rpc_listeners(self):
     self.opflex_endpoints = [o_rpc.GBPServerRpcCallback(self)]
     self.opflex_topic = o_rpc.TOPIC_OPFLEX
     self.opflex_conn = n_rpc.create_connection(new=True)
     self.opflex_conn.create_consumer(
         self.opflex_topic, self.opflex_endpoints, fanout=False)
     self.opflex_conn.consume_in_threads()

    def setup_rpc(self):
        self.service_topics = {svc_constants.CORE: topics.PLUGIN,
                               svc_constants.L3_ROUTER_NAT: topics.L3PLUGIN}
        self.conn = n_rpc.create_connection(new=True)
        self.notifier = NECPluginV2AgentNotifierApi(topics.AGENT)
        self.agent_notifiers[const.AGENT_TYPE_DHCP] = (
            dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
        )
        self.agent_notifiers[const.AGENT_TYPE_L3] = (
            nec_router.L3AgentNotifyAPI()
        )

        # NOTE: callback_sg is referred to from the sg unit test.
        self.callback_sg = securitygroups_rpc.SecurityGroupServerRpcCallback()
        self.endpoints = [
            NECPluginV2RPCCallbacks(self.safe_reference),
            dhcp_rpc.DhcpRpcCallback(),
            l3_rpc.L3RpcCallback(),
            self.callback_sg,
            agents_db.AgentExtRpcCallback(),
            metadata_rpc.MetadataRpcCallback()]
        for svc_topic in self.service_topics.values():
            self.conn.create_consumer(svc_topic, self.endpoints, fanout=False)
        # Consume from all consumers in threads
        self.conn.consume_in_threads()

 def start_rpc_listeners(self):
     self._setup_rpc()
     self.conn = n_rpc.create_connection(new=True)
     self.conn.create_consumer(topics.PLUGIN, self.endpoints, fanout=False)
     self.conn.create_consumer(topics.L3PLUGIN, self.endpoints, fanout=False)
     self.conn.create_consumer(topics.REPORTS, [agents_db.AgentExtRpcCallback()], fanout=False)
     return self.conn.consume_in_threads()

 def create_rpc_conn(self):
     self.endpoints = [CiscoCsrIPsecVpnDriverCallBack(self)]
     self.conn = n_rpc.create_connection(new=True)
     self.conn.create_consumer(
         topics.CISCO_IPSEC_DRIVER_TOPIC, self.endpoints, fanout=False)
     self.conn.consume_in_threads()
     self.agent_rpc = CiscoCsrIPsecVpnAgentApi(
         topics.CISCO_IPSEC_AGENT_TOPIC, BASE_IPSEC_VERSION, self)

    def start_rpc_listeners(self):
        # other agent based plugin driver might already set callbacks on plugin
        if hasattr(self.plugin, "agent_callbacks"):
            return

        self.conn = n_rpc.create_connection()
        self.conn.create_consumer(lb_const.LOADBALANCER_PLUGINV2, self.agent_endpoints, fanout=False)
        return self.conn.consume_in_threads()

 def _setup_rpc(self):
     self.service_topics = {svc_constants.CORE: topics.PLUGIN, svc_constants.L3_ROUTER_NAT: topics.L3PLUGIN}
     self.conn = n_rpc.create_connection(new=True)
     self.notifier = AgentNotifierApi(topics.AGENT)
     self.endpoints = [RyuRpcCallbacks(self.ofp_api_host)]
     for svc_topic in self.service_topics.values():
         self.conn.create_consumer(svc_topic, self.endpoints, fanout=False)
     self.conn.consume_in_threads()

 def start_rpc_listeners(self):
     self.endpoints = [rpc.RpcCallbacks(self.notifier, self.type_manager),
                       agents_db.AgentExtRpcCallback()]
     self.topic = topics.PLUGIN
     self.conn = n_rpc.create_connection(new=True)
     self.conn.create_consumer(self.topic, self.endpoints,
                               fanout=False)
     return self.conn.consume_in_threads()

 def _init_rpc_listeners(self):
     endpoints = [resources_rpc.ResourcesPushRpcCallback()]
     self._connection = n_rpc.create_connection()
     for rtype in self.rcache.resource_types:
         registry_rpc.register(self.resource_change_handler, rtype)
         topic = resources_rpc.resource_type_versioned_topic(rtype)
         self._connection.create_consumer(topic, endpoints, fanout=True)
     self._connection.consume_in_threads()