def __init__(self, agent, host):
        # TODO(ihrachys): we can't use RpcCallback here due to
        # inheritance issues
        self.agent = agent
        self.conf = self.agent.conf
        self.root_helper = self.agent.root_helper
        self.host = host
        self.conn = rpc.create_connection(new=True)
        self.context = context.get_admin_context_without_session()
        self.topic = topics.IPSEC_AGENT_TOPIC
        node_topic = '%s.%s' % (self.topic, self.host)

        self.processes = {}
        self.process_status_cache = {}

        self.conn.create_consumer(
            node_topic,
            self.create_rpc_dispatcher(),
            fanout=False)
        self.conn.consume_in_thread()
        self.agent_rpc = IPsecVpnDriverApi(topics.IPSEC_DRIVER_TOPIC, '1.0')
        self.process_status_cache_check = loopingcall.FixedIntervalLoopingCall(
            self.report_status, self.context)
        self.process_status_cache_check.start(
            interval=self.conf.ipsec.ipsec_status_check_interval)

    def __init__(self, server_timeout=None):
        super(NeutronRestProxyV2, self).__init__()
        LOG.info(_('NeutronRestProxy: Starting plugin. Version=%s'),
                 version_string_with_vcs())
        pl_config.register_config()

        # Include the BigSwitch Extensions path in the api_extensions
        neutron_extensions.append_api_extensions_path(extensions.__path__)

        self.add_meta_server_route = cfg.CONF.RESTPROXY.add_meta_server_route

        # init network ctrl connections
        self.servers = servermanager.ServerPool(server_timeout)

        # init dhcp support
        self.topic = topics.PLUGIN
        self.network_scheduler = importutils.import_object(
            cfg.CONF.network_scheduler_driver
        )
        self._dhcp_agent_notifier = dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
        self.agent_notifiers[const.AGENT_TYPE_DHCP] = (
            self._dhcp_agent_notifier
        )
        self.conn = rpc.create_connection(new=True)
        self.callbacks = RpcProxy()
        self.dispatcher = self.callbacks.create_rpc_dispatcher()
        self.conn.create_consumer(self.topic, self.dispatcher,
                                  fanout=False)
        # Consume from all consumers in a thread
        self.conn.consume_in_thread()
        if cfg.CONF.RESTPROXY.sync_data:
            self._send_all_data()

        LOG.debug(_("NeutronRestProxyV2: initialization done"))

    def __init__(self, agent, host):
        self.host = host
        self.conn = 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.conn.create_consumer(
            node_topic,
            self.create_rpc_dispatcher(),
            fanout=False)
        self.conn.consume_in_thread()
        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['rest_mgmt'],
                                                       v['tunnel_ip'],
                                                       v['username'],
                                                       v['password'],
                                                       v['timeout']))
                          for k, v in csrs_found.items()])

    def start(self):
        super(Service, self).start()

        self.conn = rpc.create_connection(new=True)
        LOG.debug("Creating Consumer connection for Service %s" %
                  self.topic)

        dispatcher = rpc_dispatcher.RpcDispatcher([self.manager],
                                                  self.serializer)

        # Share this same connection for these Consumers
        self.conn.create_consumer(self.topic, dispatcher, fanout=False)

        node_topic = '%s.%s' % (self.topic, self.host)
        self.conn.create_consumer(node_topic, dispatcher, fanout=False)

        self.conn.create_consumer(self.topic, dispatcher, fanout=True)

        # Hook to allow the manager to do other initializations after
        # the rpc connection is created.
        if callable(getattr(self.manager, 'initialize_service_hook', None)):
            self.manager.initialize_service_hook(self)

        # Consume from all consumers in a thread
        self.conn.consume_in_thread()

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

    :param dispatcher: The dispatcher 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 = 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, dispatcher, fanout=True)
        if node_name:
            node_topic_name = '%s.%s' % (topic_name, node_name)
            connection.create_consumer(node_topic_name,
                                       dispatcher,
                                       fanout=False)
    connection.consume_in_thread()
    return connection

 def start_rpc_listener(self):
     self.callbacks = rpc.RpcCallbacks(self.notifier, self.type_manager)
     self.topic = topics.PLUGIN
     self.conn = c_rpc.create_connection(new=True)
     self.dispatcher = self.callbacks.create_rpc_dispatcher()
     self.conn.create_consumer(self.topic, self.dispatcher,
                               fanout=False)
     return self.conn.consume_in_thread()

 def setup_rpc(self):
     self.topic = skycloud_constants.PORT_FORWARD_PLUGIN_TOPIC
     self.rpc_api = port_forward_agent_notify_api.AgentNotify
     self.agent_notifiers = {skycloud_constants.PORT_FORWARD_AGENT: self.rpc_api}
     self.conn = rpc.create_connection(new=True)
     self.callbacks = PortForwardPluginRpcCallbacks()
     self.dispatcher = self.callbacks.create_rpc_dispatcher()
     self.conn.create_consumer(self.topic, self.dispatcher, fanout=False)
     self.conn.consume_in_thread()

 def _setup_rpc_dhcp_metadata(self, notifier=None):
     self.topic = topics.PLUGIN
     self.conn = rpc.create_connection(new=True)
     self.dispatcher = nsx_rpc.NSXRpcCallbacks().create_rpc_dispatcher()
     self.conn.create_consumer(self.topic, self.dispatcher, fanout=False)
     self.agent_notifiers[const.AGENT_TYPE_DHCP] = notifier or dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
     self.conn.consume_in_thread()
     self.network_scheduler = importutils.import_object(cfg.CONF.network_scheduler_driver)
     self.supported_extension_aliases.extend(["agent", "dhcp_agent_scheduler"])

 def _setup_rpc(self):
     self.notifier = rpc.AgentNotifierApi(topics.AGENT)
     self.agent_notifiers[const.AGENT_TYPE_DHCP] = dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
     self.callbacks = rpc.RpcCallbacks(self.notifier, self.type_manager)
     self.topic = topics.PLUGIN
     self.conn = c_rpc.create_connection(new=True)
     self.dispatcher = self.callbacks.create_rpc_dispatcher()
     self.conn.create_consumer(self.topic, self.dispatcher, fanout=False)
     self.conn.consume_in_thread()

 def setup_rpc(self):
     # RPC support
     self.topic = topics.L3PLUGIN
     self.conn = rpc.create_connection(new=True)
     self.agent_notifiers.update({q_const.AGENT_TYPE_L3: l3_rpc_agent_api.L3AgentNotify})
     self.callbacks = L3RouterPluginRpcCallbacks()
     self.dispatcher = self.callbacks.create_rpc_dispatcher()
     self.conn.create_consumer(self.topic, self.dispatcher, fanout=False)
     self.conn.consume_in_thread()

 def setup_rpc(self):
     # RPC support
     self.topic = topics.PLUGIN
     self.conn = rpc.create_connection(new=True)
     self.callbacks = MidoRpcCallbacks()
     self.dispatcher = self.callbacks.create_rpc_dispatcher()
     self.conn.create_consumer(self.topic, self.dispatcher,
                               fanout=False)
     # Consume from all consumers in a thread
     self.conn.consume_in_thread()

 def _setup_rpc(self):
     # RPC support
     self.service_topics = {svc_constants.CORE: topics.PLUGIN, svc_constants.L3_ROUTER_NAT: topics.L3PLUGIN}
     self.conn = rpc.create_connection(new=True)
     self.notifier = agent_notifier_api.AgentNotifierApi(topics.AGENT)
     self.callbacks = rpc_callbacks.HyperVRpcCallbacks(self.notifier)
     self.dispatcher = self.callbacks.create_rpc_dispatcher()
     for svc_topic in self.service_topics.values():
         self.conn.create_consumer(svc_topic, self.dispatcher, fanout=False)
     # Consume from all consumers in a thread
     self.conn.consume_in_thread()

 def setup_rpc(self):
     # RPC support
     self.topic = topics.PLUGIN
     self.conn = rpc.create_connection(new=True)
     self.notifier = AgentNotifierApi(topics.AGENT)
     self.callbacks = SdnveRpcCallbacks(self.notifier)
     self.dispatcher = self.callbacks.create_rpc_dispatcher()
     self.conn.create_consumer(self.topic, self.dispatcher,
                               fanout=False)
     # Consume from all consumers in a thread
     self.conn.consume_in_thread()

    def __init__(self, plugin):
        self.agent_rpc = LoadBalancerAgentApi(TOPIC_LOADBALANCER_AGENT)
        self.callbacks = LoadBalancerCallbacks(plugin)

        self.conn = rpc.create_connection(new=True)
        self.conn.create_consumer(TOPIC_PROCESS_ON_HOST, self.callbacks.create_rpc_dispatcher(), fanout=False)
        self.conn.consume_in_thread()
        self.plugin = plugin
        self.plugin.agent_notifiers.update({q_const.AGENT_TYPE_LOADBALANCER: self.agent_rpc})

        self.pool_scheduler = importutils.import_object(cfg.CONF.loadbalancer_pool_scheduler_driver)

 def __init__(self, service_plugin):
     super(CiscoCsrIPsecVPNDriver, self).__init__(service_plugin)
     self.callbacks = CiscoCsrIPsecVpnDriverCallBack(self)
     self.conn = rpc.create_connection(new=True)
     self.conn.create_consumer(
         topics.CISCO_IPSEC_DRIVER_TOPIC,
         self.callbacks.create_rpc_dispatcher(),
         fanout=False)
     self.conn.consume_in_thread()
     self.agent_rpc = CiscoCsrIPsecVpnAgentApi(
         topics.CISCO_IPSEC_AGENT_TOPIC, BASE_IPSEC_VERSION)

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

        self.plugin.agent_callbacks = LoadBalancerCallbacks(self.plugin)
        self.plugin.conn = rpc.create_connection(new=True)
        self.plugin.conn.create_consumer(
            topics.LOADBALANCER_PLUGIN,
            self.plugin.agent_callbacks.create_rpc_dispatcher(),
            fanout=False)
        self.plugin.conn.consume_in_thread()

 def _setup_rpc(self):
     # RPC support
     self.topic = topics.PLUGIN
     self.conn = rpc.create_connection(new=True)
     self.callbacks = LinuxBridgeRpcCallbacks()
     self.dispatcher = self.callbacks.create_rpc_dispatcher()
     self.conn.create_consumer(self.topic, self.dispatcher, fanout=False)
     # Consume from all consumers in a thread
     self.conn.consume_in_thread()
     self.notifier = AgentNotifierApi(topics.AGENT)
     self.agent_notifiers[q_const.AGENT_TYPE_DHCP] = dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
     self.agent_notifiers[q_const.AGENT_TYPE_L3] = l3_rpc_agent_api.L3AgentNotify

 def _setup_rpc_dhcp_metadata(self):
     self.topic = topics.PLUGIN
     self.conn = rpc.create_connection(new=True)
     self.dispatcher = nvp_rpc.NVPRpcCallbacks().create_rpc_dispatcher()
     self.conn.create_consumer(self.topic, self.dispatcher,
                               fanout=False)
     self.agent_notifiers[const.AGENT_TYPE_DHCP] = (
         dhcp_rpc_agent_api.DhcpAgentNotifyAPI())
     self.conn.consume_in_thread()
     self.network_scheduler = importutils.import_object(
         cfg.CONF.network_scheduler_driver
     )

    def __init__(self):
        super(MeteringPlugin, self).__init__()

        self.callbacks = MeteringCallbacks(self)

        self.conn = rpc.create_connection(new=True)
        self.conn.create_consumer(
            topics.METERING_PLUGIN,
            self.callbacks.create_rpc_dispatcher(),
            fanout=False)
        self.conn.consume_in_thread()

        self.meter_rpc = metering_rpc_agent_api.MeteringAgentNotifyAPI()

 def setup_rpc(self):
     # RPC support
     self.topic = topics.PLUGIN
     self.conn = rpc.create_connection(new=True)
     self.notifier = AgentNotifierApi(topics.AGENT)
     self.dhcp_agent_notifier = dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
     self.l3_agent_notifier = l3_rpc_agent_api.L3AgentNotify
     self.callbacks = OVSRpcCallbacks(self.notifier, self.tunnel_type)
     self.dispatcher = self.callbacks.create_rpc_dispatcher()
     self.conn.create_consumer(self.topic, self.dispatcher,
                               fanout=False)
     # Consume from all consumers in a thread
     self.conn.consume_in_thread()