<s> import logging <EOL> import miniupnpc <EOL> from lbrynet . core . PTCWallet import PTCWallet <EOL> from lbrynet . core . BlobManager import DiskBlobManager , TempBlobManager <EOL> from lbrynet . dht import node <EOL> from lbrynet . core . PeerManager import PeerManager <EOL> from lbrynet . core . RateLimiter import RateLimiter <EOL> from lbrynet . core . client . DHTPeerFinder import DHTPeerFinder <EOL> from lbrynet . core . HashAnnouncer import DummyHashAnnouncer <EOL> from lbrynet . core . server . DHTHashAnnouncer import DHTHashAnnouncer <EOL> from lbrynet . core . utils import generate_id <EOL> from lbrynet . core . PaymentRateManager import BasePaymentRateManager <EOL> from twisted . internet import threads , defer <EOL> log = logging . getLogger ( __name__ ) <EOL> class LBRYSession ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , blob_data_payment_rate , db_dir = None , lbryid = None , peer_manager = None , dht_node_port = None , <EOL> known_dht_nodes = None , peer_finder = None , hash_announcer = None , <EOL> blob_dir = None , blob_manager = None , peer_port = None , use_upnp = True , <EOL> rate_limiter = None , wallet = None ) : <EOL> """<STR_LIT>""" <EOL> self . db_dir = db_dir <EOL> self . lbryid = lbryid <EOL> self . peer_manager = peer_manager <EOL> self . dht_node_port = dht_node_port <EOL> self . known_dht_nodes = known_dht_nodes <EOL> if self . known_dht_nodes is None : <EOL> self . known_dht_nodes = [ ] <EOL> self . peer_finder = peer_finder <EOL> self . hash_announcer = hash_announcer <EOL> self . blob_dir = blob_dir <EOL> self . blob_manager = blob_manager <EOL> self . peer_port = peer_port <EOL> self . use_upnp = use_upnp <EOL> self . rate_limiter = rate_limiter <EOL> self . external_ip = '<STR_LIT:127.0.0.1>' <EOL> self . upnp_redirects = [ ] <EOL> self . wallet = wallet <EOL> self . dht_node = None <EOL> self . base_payment_rate_manager = BasePaymentRateManager ( blob_data_payment_rate ) <EOL> def setup ( self ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( "<STR_LIT>" ) <EOL> if self . lbryid is None : <EOL> self . lbryid = generate_id ( ) <EOL> if self . wallet is None : <EOL> self . wallet = PTCWallet ( self . db_dir ) <EOL> if self . peer_manager is None : <EOL> self . peer_manager = PeerManager ( ) <EOL> if self . use_upnp is True : <EOL> d = self . _try_upnp ( ) <EOL> else : <EOL> d = defer . succeed ( True ) <EOL> if self . peer_finder is None : <EOL> d . addCallback ( lambda _ : self . _setup_dht ( ) ) <EOL> else : <EOL> if self . hash_announcer is None and self . peer_port is not None : <EOL> log . warning ( "<STR_LIT>" ) <EOL> self . hash_announcer = DummyHashAnnouncer ( ) <EOL> d . addCallback ( lambda _ : self . _setup_other_components ( ) ) <EOL> return d <EOL> def shut_down ( self ) : <EOL> """<STR_LIT>""" <EOL> ds = [ ] <EOL> if self . dht_node is not None : <EOL> ds . append ( defer . maybeDeferred ( self . dht_node . stop ) ) <EOL> if self . rate_limiter is not None : <EOL> ds . append ( defer . maybeDeferred ( self . rate_limiter . stop ) ) <EOL> if self . peer_finder is not None : <EOL> ds . append ( defer . maybeDeferred ( self . peer_finder . stop ) ) <EOL> if self . hash_announcer is not None : <EOL> ds . append ( defer . maybeDeferred ( self . hash_announcer . stop ) ) <EOL> if self . wallet is not None : <EOL> ds . append ( defer . maybeDeferred ( self . wallet . stop ) ) <EOL> if self . blob_manager is not None : <EOL> ds . append ( defer . maybeDeferred ( self . blob_manager . stop ) ) <EOL> if self . use_upnp is True : <EOL> ds . append ( defer . maybeDeferred ( self . _unset_upnp ) ) <EOL> return defer . DeferredList ( ds ) <EOL> def _try_upnp ( self ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> def threaded_try_upnp ( ) : <EOL> if self . use_upnp is False : <EOL> log . debug ( "<STR_LIT>" ) <EOL> return False <EOL> u = miniupnpc . UPnP ( ) <EOL> num_devices_found = u . discover ( ) <EOL> if num_devices_found > <NUM_LIT:0> : <EOL> u . selectigd ( ) <EOL> external_ip = u . externalipaddress ( ) <EOL> if external_ip != '<STR_LIT>' : <EOL> self . external_ip = external_ip <EOL> if self . peer_port is not None : <EOL> if u . getspecificportmapping ( self . peer_port , '<STR_LIT>' ) is None : <EOL> u . addportmapping ( self . peer_port , '<STR_LIT>' , u . lanaddr , self . peer_port , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . upnp_redirects . append ( ( self . peer_port , '<STR_LIT>' ) ) <EOL> log . info ( "<STR_LIT>" , self . peer_port ) <EOL> else : <EOL> log . warning ( "<STR_LIT>" , self . peer_port ) <EOL> if self . dht_node_port is not None : <EOL> if u . getspecificportmapping ( self . dht_node_port , '<STR_LIT>' ) is None : <EOL> u . addportmapping ( self . dht_node_port , '<STR_LIT>' , u . lanaddr , self . dht_node_port , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . upnp_redirects . append ( ( self . dht_node_port , '<STR_LIT>' ) ) <EOL> log . info ( "<STR_LIT>" , self . dht_node_port ) <EOL> else : <EOL> log . warning ( "<STR_LIT>" , self . dht_node_port ) <EOL> return True <EOL> return False <EOL> def upnp_failed ( err ) : <EOL> log . warning ( "<STR_LIT>" , err . getErrorMessage ( ) ) <EOL> return False <EOL> d = threads . deferToThread ( threaded_try_upnp ) <EOL> d . addErrback ( upnp_failed ) <EOL> return d <EOL> def _setup_dht ( self ) : <EOL> from twisted . internet import reactor <EOL> log . debug ( "<STR_LIT>" ) <EOL> def match_port ( h , p ) : <EOL> return h , p <EOL> def join_resolved_addresses ( result ) : <EOL> addresses = [ ] <EOL> for success , value in result : <EOL> if success is True : <EOL> addresses . append ( value ) <EOL> return addresses <EOL> def start_dht ( addresses ) : <EOL> self . dht_node . joinNetwork ( addresses ) <EOL> self . peer_finder . run_manage_loop ( ) <EOL> self . hash_announcer . run_manage_loop ( ) <EOL> return True <EOL> ds = [ ] <EOL> for host , port in self . known_dht_nodes : <EOL> d = reactor . resolve ( host ) <EOL> d . addCallback ( match_port , port ) <EOL> ds . append ( d ) <EOL> self . dht_node = node . Node ( udpPort = self . dht_node_port , lbryid = self . lbryid , <EOL> externalIP = self . external_ip ) <EOL> self . peer_finder = DHTPeerFinder ( self . dht_node , self . peer_manager ) <EOL> if self . hash_announcer is None : <EOL> self . hash_announcer = DHTHashAnnouncer ( self . dht_node , self . peer_port ) <EOL> dl = defer . DeferredList ( ds ) <EOL> dl . addCallback ( join_resolved_addresses ) <EOL> dl . addCallback ( start_dht ) <EOL> return dl <EOL> def _setup_other_components ( self ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> if self . rate_limiter is None : <EOL> self . rate_limiter = RateLimiter ( ) <EOL> if self . blob_manager is None : <EOL> if self . blob_dir is None : <EOL> self . blob_manager = TempBlobManager ( self . hash_announcer ) <EOL> else : <EOL> self . blob_manager = DiskBlobManager ( self . hash_announcer , self . blob_dir , self . db_dir ) <EOL> self . rate_limiter . start ( ) <EOL> d1 = self . blob_manager . setup ( ) <EOL> d2 = self . wallet . start ( ) <EOL> dl = defer . DeferredList ( [ d1 , d2 ] , fireOnOneErrback = True , consumeErrors = True ) <EOL> dl . addErrback ( lambda err : err . value . subFailure ) <EOL> return dl <EOL> def _unset_upnp ( self ) : <EOL> def threaded_unset_upnp ( ) : <EOL> u = miniupnpc . UPnP ( ) <EOL> num_devices_found = u . discover ( ) <EOL> if num_devices_found > <NUM_LIT:0> : <EOL> u . selectigd ( ) <EOL> for port , protocol in self . upnp_redirects : <EOL> if u . getspecificportmapping ( port , protocol ) is None : <EOL> log . warning ( "<STR_LIT>" , protocol , port ) <EOL> else : <EOL> u . deleteportmapping ( port , protocol ) <EOL> log . info ( "<STR_LIT>" , protocol , port ) <EOL> self . upnp_redirects = [ ] <EOL> d = threads . deferToThread ( threaded_unset_upnp ) <EOL> d . addErrback ( lambda err : str ( err ) ) <EOL> return d </s>
