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
  39     def getSite(self):
  40         return self.http_site
  41
  42     def joinComplete(self, result):
  43         self.my_addr = findMyIPAddr(result, config.getint(config.get('DEFAULT', 'DHT'), 'PORT'))
  44         if not self.my_addr:
  45             raise RuntimeError, "IP address for this machine could not be found"
  46
  47     def joinError(self, failure):
  48         log.msg("joining DHT failed miserably")
  49         log.err(failure)
  50         raise RuntimeError, "IP address for this machine could not be found"
  51
  52     def check_freshness(self, path, modtime, resp):
  53         log.msg('Checking if %s is still fresh' % path)
  54         d = self.peers.get([path], "HEAD", modtime)
  55         d.addCallback(self.check_freshness_done, path, resp)
  56         return d
  57
  58     def check_freshness_done(self, resp, path, orig_resp):
  59         if resp.code == 304:
  60             log.msg('Still fresh, returning: %s' % path)
  61             return orig_resp
  62         else:
  63             log.msg('Stale, need to redownload: %s' % path)
  64             return self.get_resp(path)
  65
  66     def get_resp(self, path):
  67         d = defer.Deferred()
  68
  69         log.msg('Trying to find hash for %s' % path)
  70         findDefer = self.mirrors.findHash(path)
  71
  72         findDefer.addCallbacks(self.findHash_done, self.findHash_error,
  73                                callbackArgs=(path, d), errbackArgs=(path, d))
  74         findDefer.addErrback(log.err)
  75         return d
  76
  77     def findHash_error(self, failure, path, d):
  78         log.err(failure)
  79         self.findHash_done(HashObject(), path, d)
  80
  81     def findHash_done(self, hash, path, d):
  82         if hash.expected() is None:
  83             log.msg('Hash for %s was not found' % path)
  84             self.lookupHash_done([], hash, path, d)
  85         else:
  86             log.msg('Found hash %s for %s' % (hash.hexexpected(), path))
  87             # Lookup hash from DHT
  88             key = hash.normexpected(bits = config.getint(config.get('DEFAULT', 'DHT'), 'HASH_LENGTH'))
  89             lookupDefer = self.dht.getValue(key)
  90             lookupDefer.addCallback(self.lookupHash_done, hash, path, d)
  91
  92     def lookupHash_done(self, locations, hash, path, d):
  93         if not locations:
  94             log.msg('Peers for %s were not found' % path)
  95             getDefer = self.peers.get([path])
  96             getDefer.addCallback(self.cache.save_file, hash, path)
  97             getDefer.addErrback(self.cache.save_error, path)
  98             getDefer.addCallbacks(d.callback, d.errback)
  99         else:
 100             log.msg('Found peers for %s: %r' % (path, locations))
 101             # Download from the found peers
 102             getDefer = self.peers.get(locations)
 103             getDefer.addCallback(self.check_response, hash, path)
 104             getDefer.addCallback(self.cache.save_file, hash, path)
 105             getDefer.addErrback(self.cache.save_error, path)
 106             getDefer.addCallbacks(d.callback, d.errback)
 107
 108     def check_response(self, response, hash, path):
 109         if response.code < 200 or response.code >= 300:
 110             log.msg('Download from peers failed, going to direct download: %s' % path)
 111             getDefer = self.peers.get([path])
 112             return getDefer
 113         return response
 114
 115     def new_cached_file(self, url, file_path, hash, urlpath):
 116         self.mirrors.updatedFile(url, file_path)
 117
 118         if self.my_addr and hash:
 119             site = self.my_addr + ':' + str(config.getint('DEFAULT', 'PORT'))
 120             full_path = urlunparse(('http', site, urlpath, None, None, None))
 121             key = hash.norm(bits = config.getint(config.get('DEFAULT', 'DHT'), 'HASH_LENGTH'))
 122             storeDefer = self.dht.storeValue(key, full_path)
 123             storeDefer.addCallback(self.store_done, full_path)
 124             storeDefer.addErrback(log.err)
 125
 126     def store_done(self, result, path):
 127         log.msg('Added %s to the DHT: %r' % (path, result))
 128