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