<s> from oslo_log import helpers as log_helpers <EOL> from oslo_log import log as logging <EOL> import oslo_messaging <EOL> from neutron . common import constants <EOL> from neutron . common import rpc as n_rpc <EOL> from neutron . common import topics <EOL> from neutron import manager <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class DVRServerRpcApi ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , topic ) : <EOL> target = oslo_messaging . Target ( topic = topic , version = '<STR_LIT:1.0>' , <EOL> namespace = constants . RPC_NAMESPACE_DVR ) <EOL> self . client = n_rpc . get_client ( target ) <EOL> @ log_helpers . log_method_call <EOL> def get_dvr_mac_address_by_host ( self , context , host ) : <EOL> cctxt = self . client . prepare ( ) <EOL> return cctxt . call ( context , '<STR_LIT>' , host = host ) <EOL> @ log_helpers . log_method_call <EOL> def get_dvr_mac_address_list ( self , context ) : <EOL> cctxt = self . client . prepare ( ) <EOL> return cctxt . call ( context , '<STR_LIT>' ) <EOL> @ log_helpers . log_method_call <EOL> def get_ports_on_host_by_subnet ( self , context , host , subnet ) : <EOL> """<STR_LIT>""" <EOL> cctxt = self . client . prepare ( ) <EOL> return cctxt . call ( context , '<STR_LIT>' , <EOL> host = host , subnet = subnet ) <EOL> @ log_helpers . log_method_call <EOL> def get_subnet_for_dvr ( self , context , subnet , fixed_ips ) : <EOL> cctxt = self . client . prepare ( ) <EOL> return cctxt . call ( <EOL> context , '<STR_LIT>' , subnet = subnet , fixed_ips = fixed_ips ) <EOL> class DVRServerRpcCallback ( object ) : <EOL> """<STR_LIT>""" <EOL> target = oslo_messaging . Target ( version = '<STR_LIT>' , <EOL> namespace = constants . RPC_NAMESPACE_DVR ) <EOL> @ property <EOL> def plugin ( self ) : <EOL> if not getattr ( self , '<STR_LIT>' , None ) : <EOL> self . _plugin = manager . NeutronManager . get_plugin ( ) <EOL> return self . _plugin <EOL> def get_dvr_mac_address_list ( self , context ) : <EOL> return self . plugin . get_dvr_mac_address_list ( context ) <EOL> def get_dvr_mac_address_by_host ( self , context , ** kwargs ) : <EOL> host = kwargs . get ( '<STR_LIT:host>' ) <EOL> LOG . debug ( "<STR_LIT>" , host ) <EOL> return self . plugin . get_dvr_mac_address_by_host ( context , host ) <EOL> def get_ports_on_host_by_subnet ( self , context , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> host = kwargs . get ( '<STR_LIT:host>' ) <EOL> subnet = kwargs . get ( '<STR_LIT>' ) <EOL> LOG . debug ( "<STR_LIT>" , host ) <EOL> return self . plugin . get_ports_on_host_by_subnet ( context , <EOL> host , subnet ) <EOL> def get_subnet_for_dvr ( self , context , ** kwargs ) : <EOL> fixed_ips = kwargs . get ( '<STR_LIT>' ) <EOL> subnet = kwargs . get ( '<STR_LIT>' ) <EOL> return self . plugin . get_subnet_for_dvr ( <EOL> context , subnet , fixed_ips = fixed_ips ) <EOL> class DVRAgentRpcApiMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> DVR_RPC_VERSION = "<STR_LIT:1.0>" <EOL> def _get_dvr_update_topic ( self ) : <EOL> return topics . get_topic_name ( self . topic , <EOL> topics . DVR , <EOL> topics . UPDATE ) <EOL> def dvr_mac_address_update ( self , context , dvr_macs ) : <EOL> """<STR_LIT>""" <EOL> if not dvr_macs : <EOL> return <EOL> cctxt = self . client . prepare ( topic = self . _get_dvr_update_topic ( ) , <EOL> version = self . DVR_RPC_VERSION , fanout = True ) <EOL> cctxt . cast ( context , '<STR_LIT>' , dvr_macs = dvr_macs ) <EOL> class DVRAgentRpcCallbackMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def dvr_mac_address_update ( self , context , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> dvr_macs = kwargs . get ( '<STR_LIT>' , [ ] ) <EOL> LOG . debug ( "<STR_LIT>" , dvr_macs ) <EOL> self . dvr_agent . dvr_mac_address_update ( dvr_macs ) </s>
