apt_dht/apt_dht.py

   1
   2 from binascii import b2a_hex
   3 from urlparse import urlunparse
   4 import os, re
   5
   6 from twisted.internet import defer
   7 from twisted.web2 import server, http, http_headers
   8 from twisted.python import log
   9
  10 from apt_dht_conf import config
  11 from PeerManager import PeerManager
  12 from HTTPServer import TopLevel
  13 from MirrorManager import MirrorManager
  14 from Hash import HashObject
  15 from db import DB
  16 from util import findMyIPAddr
  17
  18 class AptDHT:
  19     def __init__(self, dht):
  20         log.msg('Initializing the main apt_dht application')
  21         self.db = DB(config.get('DEFAULT', 'cache_dir') + '/.apt-dht.db')
  22         self.dht = dht
  23         self.dht.loadConfig(config, config.get('DEFAULT', 'DHT'))
  24         self.dht.join().addCallbacks(self.joinComplete, self.joinError)
  25         self.http_server = TopLevel(config.get('DEFAULT', 'cache_dir'), self)
  26         self.http_site = server.Site(self.http_server)
  27         self.peers = PeerManager()
  28         self.mirrors = MirrorManager(config.get('DEFAULT', 'cache_dir'), self)
  29         self.my_addr = None
  30
  31     def getSite(self):
  32         return self.http_site
  33
  34     def joinComplete(self, result):
  35         self.my_addr = findMyIPAddr(result, config.getint(config.get('DEFAULT', 'DHT'), 'PORT'))
  36         if not self.my_addr:
  37             raise RuntimeError, "IP address for this machine could not be found"
  38
  39     def joinError(self, failure):
  40         log.msg("joining DHT failed miserably")
  41         log.err(failure)
  42
  43     def check_freshness(self, path, modtime, resp):
  44         log.msg('Checking if %s is still fresh' % path)
  45         d = self.peers.get([path], "HEAD", modtime)
  46         d.addCallback(self.check_freshness_done, path, resp)
  47         return d
  48
  49     def check_freshness_done(self, resp, path, orig_resp):
  50         if resp.code == 304:
  51             log.msg('Still fresh, returning: %s' % path)
  52             return orig_resp
  53         else:
  54             log.msg('Stale, need to redownload: %s' % path)
  55             return self.get_resp(path)
  56
  57     def get_resp(self, path):
  58         d = defer.Deferred()
  59
  60         log.msg('Trying to find hash for %s' % path)
  61         findDefer = self.mirrors.findHash(path)
  62
  63         findDefer.addCallbacks(self.findHash_done, self.findHash_error,
  64                                callbackArgs=(path, d), errbackArgs=(path, d))
  65         findDefer.addErrback(log.err)
  66         return d
  67
  68     def findHash_error(self, failure, path, d):
  69         log.err(failure)
  70         self.findHash_done(HashObject(), path, d)
  71
  72     def findHash_done(self, hash, path, d):
  73         if hash.expected() is None:
  74             log.msg('Hash for %s was not found' % path)
  75             self.lookupHash_done([], hash, path, d)
  76         else:
  77             log.msg('Found hash %s for %s' % (hash.hexexpected(), path))
  78             # Lookup hash from DHT
  79             key = hash.normexpected(bits = config.getint(config.get('DEFAULT', 'DHT'), 'HASH_LENGTH'))
  80             lookupDefer = self.dht.getValue(key)
  81             lookupDefer.addCallback(self.lookupHash_done, hash, path, d)
  82
  83     def lookupHash_done(self, locations, hash, path, d):
  84         if not locations:
  85             log.msg('Peers for %s were not found' % path)
  86             getDefer = self.peers.get([path])
  87             getDefer.addCallback(self.mirrors.save_file, hash, path)
  88             getDefer.addErrback(self.mirrors.save_error, path)
  89             getDefer.addCallbacks(d.callback, d.errback)
  90         else:
  91             log.msg('Found peers for %s: %r' % (path, locations))
  92             # Download from the found peers
  93             getDefer = self.peers.get(locations)
  94             getDefer.addCallback(self.check_response, hash, path)
  95             getDefer.addCallback(self.mirrors.save_file, hash, path)
  96             getDefer.addErrback(self.mirrors.save_error, path)
  97             getDefer.addCallbacks(d.callback, d.errback)
  98
  99     def check_response(self, response, hash, path):
 100         if response.code < 200 or response.code >= 300:
 101             log.msg('Download from peers failed, going to direct download: %s' % path)
 102             getDefer = self.peers.get([path])
 103             return getDefer
 104         return response
 105
 106     def cached_file(self, hash, url, file_path):
 107         assert file_path.startswith(config.get('DEFAULT', 'cache_dir'))
 108         urlpath, newdir = self.db.storeFile(file_path, hash.digest(), config.get('DEFAULT', 'cache_dir'))
 109         log.msg('now avaliable at %s: %s' % (urlpath, url))
 110
 111         if self.my_addr:
 112             site = self.my_addr + ':' + str(config.getint('DEFAULT', 'PORT'))
 113             full_path = urlunparse(('http', site, urlpath, None, None, None))
 114             key = hash.norm(bits = config.getint(config.get('DEFAULT', 'DHT'), 'HASH_LENGTH'))
 115             storeDefer = self.dht.storeValue(key, full_path)
 116             storeDefer.addCallback(self.store_done, full_path)
 117             storeDefer.addErrback(log.err)
 118
 119     def store_done(self, result, path):
 120         log.msg('Added %s to the DHT: %r' % (path, result))
 121