def __init__(self, **kwds):
        """Build a HTTP client or a local client that talks directly to a job manger."""
        if 'pulsar_app' in kwds or 'job_manager' in kwds:
            self.job_manager_interface_class = LocalPulsarInterface
            pulsar_app = kwds.get('pulsar_app', None)
            job_manager = kwds.get('job_manager', None)
            file_cache = kwds.get('file_cache', None)
            self.job_manager_interface_args = dict(
                job_manager=job_manager,
                pulsar_app=pulsar_app,
                file_cache=file_cache,
            )
        else:
            self.job_manager_interface_class = HttpPulsarInterface
            transport_type = kwds.get('transport', None)
            transport_params = dict([(p.replace('transport_', '', 1), v) for p, v in kwds.items() if p.startswith('transport_')])
            transport = get_transport(transport_type, transport_params=transport_params)
            self.job_manager_interface_args = dict(transport=transport)
        cache = kwds.get('cache', None)
        if cache is None:
            cache = _environ_default_int('PULSAR_CACHE_TRANSFERS')
        if cache:
            log.info("Setting Pulsar client class to caching variant.")
            self.client_cacher = ClientCacher(**kwds)
            self.client_class = InputCachingJobClient
            self.extra_client_kwds = {"client_cacher": self.client_cacher}
        else:
            log.info("Setting Pulsar client class to standard, non-caching variant.")
            self.client_class = JobClient
            self.extra_client_kwds = {}
def __init__ ( self , ** kwds ) : if 'pulsar_app' in kwds or 'job_manager' in kwds : self . job_manager_interface_class = LocalPulsarInterface pulsar_app = kwds . get ( 'pulsar_app' , None ) job_manager = kwds . get ( 'job_manager' , None ) file_cache = kwds . get ( 'file_cache' , None ) self . job_manager_interface_args = dict ( job_manager = job_manager , pulsar_app = pulsar_app , file_cache = file_cache , ) else : self . job_manager_interface_class = HttpPulsarInterface transport_type = kwds . get ( 'transport' , None ) transport_params = dict ( [ ( p . replace ( 'transport_' , '' , 1 ) , v ) for p , v in kwds . items ( ) if p . startswith ( 'transport_' ) ] ) transport = get_transport ( transport_type , transport_params = transport_params ) self . job_manager_interface_args = dict ( transport = transport ) cache = kwds . get ( 'cache' , None ) if cache is None : cache = _environ_default_int ( 'PULSAR_CACHE_TRANSFERS' ) if cache : log . info ( "Setting Pulsar client class to caching variant." ) self . client_cacher = ClientCacher ( ** kwds ) self . client_class = InputCachingJobClient self . extra_client_kwds = { "client_cacher" : self . client_cacher } else : log . info ( "Setting Pulsar client class to standard, non-caching variant." ) self . client_class = JobClient self . extra_client_kwds = { }
Build a HTTP client or a local client that talks directly to a job manger.