#
# Copyright (C) 2005 Chris Halls <halls@debian.org>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of version 2.1 of the GNU Lesser General Public
# License as published by the Free Software Foundation.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

"""
Fetchers for apt-proxy

These classes implement the network code for fetching files from apt-proxy
network backends
"""

import re, os, string, time, glob, signal, stat, base64, urllib
from twisted.web import static, http
from twisted.internet import protocol, reactor, defer, error, abstract
from twisted.python import failure
from twisted.protocols import policies, ftp

from misc import log


class Fetcher:
    """
    This class manages the selection of a BackendServer and downloading from
    that backend
    
    """
    cacheEntry = None
    fetcher = None   # connection-specific fetcher

    def __init__(self):
        self.backendServer = None
        self.size = None # Size of file notified by fetcher's server
        self.mtime = None # Mtime of file notified by fetcher's server

    def start(self, cacheEntry):
        self.cacheEntry = cacheEntry
        log.debug("fetcher start:" + self.cacheEntry.filename, "fetcher")
        self.backend = cacheEntry.backend
        self.len_received = 0
        self.deferred = defer.Deferred()
        self.start_download()
        return self.deferred

    def activateNextBackendServer(self, fetcher):
        """
        Returns true if another BackendServer was found
        """
        if self.backendServer is None:
            self.backendServer = self.backend.get_first_server()
            if(self.backendServer == None):
                log.err("No backend server found for backend " + self.backend.name, "fetcher")
                return False
        else:
            # Look for the next possible BackendServer
            self.backendServer = self.backend.get_next_server(self.backendServer)

            if(self.backendServer == None):
                # The attempt to retrieve a file from the BackendServer failed.
                log.debug("no more Backends", "fetcher")
                return False
        self.connectToBackend()

    def connectToBackend(self):
        log.debug('Connecting to backend server %s' % (self.backendServer), 'fetcher')
        self.fetcher = self.backendServer.fetcher(self.backendServer)
        d = self.fetcher.connect()
        d.addCallback(self.connected)
        d.addErrback(self.connection_failed)
        #fetcher.apEndTransfer(fetcher_class)

        return True

    def __str__(self):
        return 'Fetcher server=%s file=%s' % (str(self.backendServer), self.cacheEntry.path)

    def start_download(self):
        """
        Begin streaming file
        Serve from cache or through the appropriate Fetcher
        depending on the asociated backend.

        Use post_convert and gzip_convert regular expresions of the Fetcher
        to gzip/gunzip file before and after download.
        """
        log.debug("Downloading: " + self.cacheEntry.file_path, 'Fetcher')
        #init_tempfile()
        if self.backendServer is None:
            self.activateNextBackendServer(self.fetcher)
        elif self.fetcher is None:
            self.connectToBackend()
        else:
            self.download()

    def download_complete(self):
        """
        Download was successful
        """
        log.debug("download complete. Sent:%s bytes" % (self.len_received), "Fetcher")
        if self.fetcher is not None and not self.fetcher.pipelining:
            self.connection_closed(self.fetcher)
        if self.len_received==0:
            self.download_started() # Send status code to clients
        self.cacheEntry.download_data_end()
        self.deferred.callback((True, ""))

    def fail_over(self, reason_code, reason_msg):
        """
        A non-fatal download has occured. Attempt download from next
        backend
        """
        if not self.activateNextBackendServer(self.fetcher):
            self.download_failed(reason_code, reason_msg)

    def download_failed(self, reason_code, reason_msg):
        #self.cacheEntry.download_data_end()
        log.debug("download_failed: (%s) %s " %(reason_code, reason_msg), "Fetcher")
        if self.fetcher is not None and not self.fetcher.pipelining:
            self.connection_closed(self.fetcher)
        self.cacheEntry.download_failure(reason_code, reason_msg)
        self.deferred.callback((False, reason_msg))

    def cancel_download(self):
        if self.fetcher:
            log.debug(
                "telling fetchers to disconnect",'Fetcher')
            self.fetcher.disconnect()
        self.download_failed(None, "Download canceled")

    def data_received(self, data, save=True):
        """
        File Data has been received from the backend server
        @param data: raw data received from server
        @param save: if true, save to disk (rsync saves file itself)
        """
        #log.debug("data_received: %s bytes" % len(data), 'Fetcher');
        if not self.len_received:
            self.download_started(save)
        self.len_received = self.len_received + len(data)
        self.cacheEntry.download_data_received(data)

    def download_started(self, save=True):
        if save:
            self.cacheEntry.init_tempfile()
        self.cacheEntry.download_started(self, self.size, self.mtime)


    def server_size(self, len):
        """
        The server has sent the expected length of the file
        """
        self.size = len
        log.debug("File size: " + str(len), 'Fetcher');

    def server_mtime(self, mtime):
        """
        The server has sent the modification time of the file
        """
        self.mtime = mtime
        log.debug("File mtime: " + str(mtime), 'Fetcher');

    def transfer_complete(self):
        """
        All data has been transferred
        """
        log.debug("Finished receiving data: " + self.cacheEntry.filename, 'Fetcher');
        self.download_complete()

    def connection_failed(self, reason = None):
        """
        A fetcher has failed to connect to the backend server
        """
        msg = '[%s] Connection Failed: %s/%s'%(
            self.backend.name,
            self.backendServer.path, self.cacheEntry.path)

        if reason:
            msg = '%s (%s)'%(msg, reason.getErrorMessage())
            log.debug("Connection Failed: "+str(reason), 'Fetcher')
        log.err(msg)
        self.fail_over(http.SERVICE_UNAVAILABLE, reason)

    def connected(self, result):
        log.debug("Connected to "+ self.backendServer.uri, 'Fetcher')
        self.download()

    def download(self):
        mtime = self.cacheEntry.get_request_mtime()
        log.debug('downloading:%s mtime:%s' % (self.cacheEntry.path, mtime), 'Fetcher')
        self.fetcher.download(self, self.cacheEntry.path, mtime)

    def disconnect(self):
        if self.fetcher is not None:
            log.debug('disconnect %s' % (self.cacheEntry.path), 'Fetcher')
            self.fetcher.disconnect()
            self.fetcher = None

    def connection_closed(self, fetcher):
        """
        A protocol fetcher's connection has closed - we must reopen the connection
        next time
        """
        log.debug("Connection closed for %s, state=%s" %(self.cacheEntry.path, self.cacheEntry.state), 'Fetcher')
        #if self.cacheEntry.state in \
        #   (self.cacheEntry.STATE_CONNECTING, self.cacheEntry.STATE_DOWNLOAD, self.cacheEntry.STATE_SENDFILE):
        #    self.fetcher_internal_error("Backend connection closed")
        if fetcher == self.fetcher:
            self.fetcher = None

    def file_not_found(self):
        log.msg("(%s) file not found: %s" % (self.backendServer.path, self.cacheEntry.path), 'fetcher')
        # TODO - failover?
        self.download_failed(http.NOT_FOUND, "file not found on backend")

    def fetcher_internal_error(self, reason):
        log.msg("(%s) internal error: %s" % (self.backendServer.path, reason), 'fetcher')
        self.download_failed(http.INTERNAL_SERVER_ERROR, reason)

    def send_complete_file(self, filename):
        """
        Send a complete file (used by FileFetcher)
        """
        self.cacheEntry.transfer_file(filename)

    def up_to_date(self):
        """
        Fetcher has determined that our cached file is up to date
        so the file is sent from our cache
        """
        log.msg("(%s) up_to_date" % (self.cacheEntry.path), 'fetcher')
        self.cacheEntry.send_cached_file()
        if not self.fetcher.pipelining:
            self.connection_closed(self.fetcher)
        self.deferred.callback((True, ""))

def uri_path_to_path(path, check_part):
    # Split into parts and unescape them.
    parts = [urllib.unquote(part) for part in path.split('/')]
    for part in parts:
        if not check_part(part):
            return None
    # Join up the parts.
    return os.sep.join(parts)

def is_valid_local_path_part(part):
    # Deny use of parent directory or characters that are invalid in a
    # path part.
    return not (part == os.pardir
                or '\0' in part
                or os.sep in part
                or (os.altsep and os.altsep in part))

class FileFetcher:
    """
    A Fetcher that simply copies files from disk
    """
    pipelining = True
    def __init__(self, backendServer):
        self.backendServer = backendServer
        self.isConnected = True # Always connected

    def connect(self):
        # We always conect
        return defer.succeed(True)

    def download(self, fetcher, uri, mtime):
        """
        Request download
        %param fetcher: Fetcher class to receive callbacks
        %param uri: URI of file to be downloaded within backend
        %param mtime: Modification time of current file in cache
        """
        self.parent = fetcher
        self.cache_mtime = mtime
        self.request_uri = uri

        path = uri_path_to_path(uri, is_valid_local_path_part)
        if path is None:
            self.parent.file_not_found()
            return
        self.local_file = self.backendServer.uri[len("file://"):] + '/' + path
        if not os.path.exists(self.local_file):
            self.parent.file_not_found()
            return

        # start the transfer
        self.parent.send_complete_file(self.local_file)

    def disconnect(self):
        pass

class FetcherHttpClient(http.HTTPClient):
    """
    This class represents an Http conncetion to a backend
    server. It is generated by the HttpFetcher class when
    a connection is made to an http server
    """
    def __init__(self, parent):
        self.parent = parent # HttpFetcher
        self.proxy = self.parent.proxy
        self.fetcher = None

    def connectionMade(self):
        """
        Http connection made - inform parent, which will
        trigger callbacks
        """
        self.parent.connected(self)

    def download(self, fetcher, uri, mtime):
        # Request file from backend
        self.log_headers = None
        self.close_on_completion = True
        self.server_mtime = None
        self.server_size = None
        self.fetcher = fetcher
        self.uri = uri
        self.finished = False
        backendServer = self.parent.backendServer
        if self.proxy is None:
            serverpath = backendServer.path
        else:
            serverpath = "http://" + backendServer.host
            if backendServer.port != 80:
                serverpath = serverpath + ":" + str(backendServer.port)
            serverpath = serverpath + "/" + backendServer.path 

        #self.sendCommand(self.request.method, 
        self.sendCommand("GET", serverpath + "/" + uri)

        self.sendHeader('host', backendServer.host)
        if self.proxy is not None and self.proxy.user is not None:
            self.sendHeader('Proxy-Authorization', "Basic " +
                            base64.encodestring(self.proxy.user + ":" + self.proxy.password)[:-1])

        if mtime is not None:
            datetime = http.datetimeToString(mtime)
            self.sendHeader('if-modified-since', datetime)

        self.endHeaders()

    def download_complete(self):
        if self.finished: 
            return
        log.debug("File transfer complete",'http_client')
        self.finished = True
        #if self.close_on_completion:
            #self.fetcher.disconnect()
            #self.parent.connection_closed() # We don't have a persistent connection
            #self.fetcher.disconnect()
            #self.transport.loseConnection()
        self.fetcher.download_complete()

    def handleStatus(self, version, code, message):
        __pychecker__ = 'unusednames=version,message'
        log.debug('handleStatus %s - %s' % (code, message), 'http_client')
        self.http_status = int(code)

        #self.setResponseCode(self.http_status)

    def handleResponse(self, buffer):
        #log.debug('handleResponse, %s bytes' % (len(buffer)), 'http_client')
        log.debug('handleResponse status=%s' % (self.http_status), 'http_client')
        if self.http_status == http.NOT_MODIFIED:
            log.debug("Backend server reported file is not modified: " + self.uri,'http_client')
            self.fetcher.up_to_date()
        elif self.http_status == http.NOT_FOUND:
            log.debug("Not found on backend server",'http_client')
            self.fetcher.file_not_found()
        elif self.http_status == http.OK:
            self.download_complete()
        else:
            log.debug("Unknown status code: %s" % (self.http_status),'http_client')
            self.fetcher.fetcher_internal_error("Unknown status code: %s" % (self.http_status))

    def handleHeader(self, key, value):

        log.debug("Received: " + key + " " + str(value), 'http_client')
        key = string.lower(key)

        if key == 'last-modified':
            self.server_mtime = http.stringToDatetime(value)
            self.fetcher.server_mtime(self.server_mtime)
        elif key == 'content-length':
            self.server_size = int(value)
            self.fetcher.server_size(self.server_size)
        elif key == 'connection':
            if value == "close":
                log.debug('will close on completion', 'http_client')
                self.close_on_completion = True
            elif value == "keep-alive":
                log.debug('will not close on completion', 'http_client')
                self.close_on_completion = False

    #def handleEndHeaders(self):
        #if self.http_status == http.NOT_MODIFIED:
            #log.debug("Backend server reported file is not modified: " + self.uri,'http_client')
            #self.fetcher.up_to_date()
        #elif self.http_status == http.NOT_FOUND:
            #log.debug("Not found on backend server",'http_client')
            #self.fetcher.file_not_found()
        #else:
            #log.debug("Unknown status code: %s" % (self.http_status),'http_client')

    def rawDataReceived(self, data):
        if self.http_status == http.OK:
            self.fetcher.data_received(data)
            #log.debug("Recieved: %s expected: %s" % (self.fetcher.len_received, self.server_size),'http_client')
            if self.server_size is not None:
                if self.fetcher.len_received >= self.server_size:
                    if self.fetcher.len_received == self.server_size:
                        pass
                        #self.download_complete()
                    else:
                        log.err("File transfer overrun! Expected size:%s Received size:%s" % 
                                (self.server_size, self.fetcher.len_received), 'http_client')
                        self.parent.fetcher_internal_error("Data overrun")

#     def handleResponse(self, buffer):
#         if self.length == 0:
#             self.setResponseCode(http.NOT_FOUND)
#         # print "length: " + str(self.length), "response:", self.status_code
#         if self.http_status == http.NOT_MODIFIED:
#             self.apDataEnd(self.transfered, False)
#         else:
#             self.apDataEnd(self.transfered, True)

    def lineReceived(self, line):
        """
        log each line as received from server
        """
        if self.log_headers is None:
            self.log_headers = line
        else:
            self.log_headers += ", " + line;
        http.HTTPClient.lineReceived(self, line)

    def sendCommand(self, command, path):
        "log the line and handle it to the base class."
        log.debug(command + ":" + path,'http_client')
        http.HTTPClient.sendCommand(self, command, path)

    def endHeaders(self):
        "log and handle to the base class."
        if self.log_headers != None:
            log.debug(" Headers: " + self.log_headers, 'http_client')
            self.log_headers = None;
        http.HTTPClient.endHeaders(self)

    def sendHeader(self, name, value):
        "log and handle to the base class."
        log.debug(name + " sendHeader:" + value,'http_client')
        http.HTTPClient.sendHeader(self, name, value)
        
    def disconnect(self):
        log.debug("DISCONNECT:",'http_client')
        import traceback
        traceback.print_stack()


class HttpFetcher(protocol.ClientFactory):
    """
    A Fetcher factory that retrieves files via HTTP
    """
    pipelining = False # twisted's HTTP client does not support pipelining
    def __init__(self, backendServer):
        self.backendServer = backendServer
        self.isConnected = False
        self.instance = None

    def connect(self):
        self.connectCallback = defer.Deferred()
        self.proxy = self.backendServer.backend.config.http_proxy
        if self.proxy is None:
            host = self.backendServer.host
            port = self.backendServer.port
        else:
            host = self.proxy.host
            port = self.proxy.port
        self.read_limit = self.backendServer.backend.config.bandwidth_limit
        if self.read_limit is None:
            factory = self
        else:
            # Limit download rate
            factory = policies.ThrottlingFactory(self, readLimit=self.read_limit)
        reactor.connectTCP(host, port, factory, self.backendServer.backend.config.timeout)
        return self.connectCallback

    def buildProtocol(self, addr):
        return FetcherHttpClient(self)

    def connected(self, connection):
        "Connection was made to HTTP backend (callback from HTTP client)"
        self.connection = connection
        self.isConnected = True
        self.connectCallback.callback(None)

    def clientConnectionFailed(self, connector, reason):
        #self.instance.connectionFailed(reason)
        log.debug("clientConnectionFailed reason: %s" % (reason), "http-client")
        self.connectCallback.errback(reason)
    def clientConnectionLost(self, connector, reason):
        log.debug("clientConnectionLost reason=%s" %(reason), "http-client")
        if self.connection is not None and self.connection.fetcher is not None:
            self.connection.fetcher.connection_closed(self)

    def download(self, fetcher, uri, mtime):
        """
        Request download
        %param fetcher: Fetcher class to receive callbacks
        %param uri: URI of file to be downloaded within backend
        %param mtime: Modification time of current file in cache
        """
        self.connection.download(fetcher, uri, mtime)

    def disconnect(self):
        if self.isConnected:
            self.connection.transport.loseConnection()
            self.isConnected = False

# RFC 959 says pathnames must be ASCII and not include CR or LF.
ftp_path_part_re = re.compile(r'[^\r\n\x80-\xFF]+$')
def is_valid_ftp_path_part(part):
    # Also deny use of parent directory, assuming Unix path conventions
    # on the server.
    return part != '..' and ftp_path_part_re.match(part)

class FtpFetcher(protocol.Protocol):
    """
    This is the secuence here:

        -Start and connect the FTPClient
        -Ask for mtime
        -Ask for size
        -if couldn't get the size
            -try to get it by listing
        -get all that juicy data
        
    NOTE: Twisted's FTPClient code uses it's own timeouts here and there,
    so the timeout specified for the backend may not always be used
    """
    
    pipelining = True
    def __init__(self, backendServer):
        self.backendServer = backendServer
        self.isConnected = False
        self.instance = None
        self.ftpclient = None

    def connect(self):
        """
        Establish connection to ftp server specified by backendServer
        """
        self.connectCallback = defer.Deferred()
        if not self.backendServer.username:
            creator = protocol.ClientCreator(reactor, ftp.FTPClient, passive=0)
        else:
            creator = protocol.ClientCreator(reactor, ftp.FTPClient, request.backendServer.username,
                                    request.backendServer.password, passive=0)
        d = creator.connectTCP(self.backendServer.host, self.backendServer.port,
                               self.backendServer.backend.config.timeout)
        d.addCallback(self.controlConnectionMade)
        d.addErrback(self.clientConnectionFailed)
        return self.connectCallback

    def controlConnectionMade(self, ftpclient):
        self.ftpclient = ftpclient

        if(self.backendServer.backend.config.passive_ftp):
            log.debug('Got control connection, using passive ftp', 'ftp_client')
            self.ftpclient.passive = 1
        else:
            log.debug('Got control connection, using active ftp', 'ftp_client')
            self.ftpclient.passive = 0

        if log.isEnabled('ftp_client'):
            self.ftpclient.debug = 1
        self.connectCallback.callback(None)

    def clientConnectionFailed(self, reason):
        #self.instance.connectionFailed(reason)
        log.debug("clientConnectionFailed reason: %s" % (reason), "ftp_client")
        self.connectCallback.errback(reason)

    def download(self, fetcher, uri, mtime):
        """
        Request download
        %param fetcher: Fetcher class to receive callbacks
        %param uri: URI of file to be downloaded within backend
        %param mtime: Modification time of current file in cache
        """
        self.parent = fetcher
        self.cache_mtime = mtime
        self.request_uri = uri
        path = uri_path_to_path(uri, is_valid_ftp_path_part)
        if path is None:
            self.parent.file_not_found()
            return
        self.remote_file = (self.parent.backendServer.path + '/' 
                            + path)
        self.ftpFetchMtime()

    def ftpFetchMtime(self):
        "Get the modification time from the server."
        d = self.ftpclient.queueStringCommand('MDTM ' + self.remote_file)
        d.addCallback(self.ftpMtimeResult)
        d.addErrback(self.ftpMtimeFailed)

    def ftpMtimeResult(self, msgs):
        """
        Got an answer to the mtime request.

        Someone should check that this is timezone independent.
        """
        code, msg = msgs[0].split()
        if code == '213':
            time_tuple=time.strptime(msg[:14], "%Y%m%d%H%M%S")
            #replace day light savings with -1 (current)
            time_tuple = time_tuple[:8] + (-1,)
            #correct the result to GMT
            mtime = time.mktime(time_tuple) - time.altzone
            self.parent.server_mtime(mtime)

            if (self.cache_mtime
                    and self.cache_mtime >= mtime):
                self.parent.up_to_date()
                return
        self.ftpFetchSize()

    def ftpMtimeFailed(self, msgs):
        if msgs.check(ftp.CommandFailed):
            code = msgs.getErrorMessage()[2:5]
            log.debug("ftp fetch of Mtime failed: %s code:%s" % (msgs.getErrorMessage(), code), 'ftp_client')
            if code == '550':
                # Not found
                self.parent.file_not_found()
                return
        log.debug("ftp fetch of Mtime for %s unknown failure: %s" % (self.remote_file, msgs), 'ftp_client')
        self.ftpFetchSize()

    def ftpFetchSize(self):
        "Get the size of the file from the server"

        d = self.ftpclient.queueStringCommand('SIZE ' + self.remote_file)
        d.addCallback(self.ftpSizeResult)
        d.addErrback(self.ftpSizeFailed)

    def ftpSizeResult(self, msgs):
        code, msg = msgs[0].split()
        if code == '213':
            size = int(msg)
            self.parent.server_size(size)
            self.ftpFetchFile()
        else:
            self.ftpSizeFailed()
    def ftpSizeFailed(self, msgs):
        log.debug("ftp size failed: %s" % (msgs), 'ftp_client')
        self.ftpFetchList()

    def ftpFetchList(self):
        "If ftpFetchSize didn't work try to get the size with a list command."
        self.filelist = ftp.FTPFileListProtocol()
        d = self.ftpclient.list(self.remote_file, self.filelist)
        d.addCallback(self.ftpListResult)
        d.addErrback(self.ftpListFailed)

    def ftpListResult(self, msg):
        __pychecker__ = 'unusednames=msg'
        if len(self.filelist.files)== 0:
            log.debug("Not found on backend server",'ftp_client')
            self.parent.file_not_found()
            return
        file = self.filelist.files[0]
        self.parent.server_size(file['size'])
        fetcher.ftpFetchFile()

    def ftpListFailed(self, msgs):
        log.debug("ftp list failed: %s" % (msgs), 'ftp_client')
        self.parent.fetcher_internal_error("Could not list directory")

    def ftpFetchFile(self):
        "And finally, we ask for the file."
        log.debug('ftpFetchFile: ' + self.remote_file, 'ftp_client')
        d = self.ftpclient.retrieveFile(self.remote_file, self)
        d.addCallback(self.ftpFetchResult)
        d.addErrback(self.ftpFetchFailed)
    def ftpFetchResult(self, msg):
        self.parent.download_complete()
    def ftpFetchFailed(self, msgs):
        log.debug("ftp fetch failed: %s" % (msgs), 'ftp_client')
        self.parent.file_not_found()

    def dataReceived(self, data):
        self.parent.data_received(data)

    def disconnect(self):
        if self.ftpclient is not None:
            log.debug('disconnecting', 'ftp_client')
            self.ftpclient.quit()
            self.ftpclient.transport.loseConnection()
            self.ftpclient = None

    def connectionLost(self, reason=None):
        """
        Maybe we should do some recovery here, I don't know, but the Deferred
        should be enough.
        """
        log.debug("lost connection: %s"%(reason),'ftp_client')

class GzipFetcher(Fetcher, protocol.ProcessProtocol):
    """
    This is a fake Fetcher, it uses the real Fetcher from the request's
    backend via LoopbackRequest to get the data and gzip's or gunzip's as
    needed.

    NOTE: We use the serve_cached=0 parameter to Request.fetch so if
    it is cached it doesn't get uselessly read, we just get it from the cache.
    """
    post_convert = re.compile(r"^Should not match anything$")
    gzip_convert = post_convert

    exe = '/bin/gzip'
    def activate(self, request, postconverting=0):
        log.debug("FetcherGzip request:" + str(request.uri) + " postconvert:" + str(postconverting), 'gzip')
        Fetcher.activate(self, request)
        if not request.apFetcher:
            return

        self.args = (self.exe, '-c', '-9', '-n')
        if(log.isEnabled('gzip',9)):
            self.args += ('-v',)

        if request.uri[-3:] == '.gz':
            host_uri = request.uri[:-3]
        else:
            host_uri = request.uri+'.gz'
            self.args += ('-d',)
        self.host_file = self.factory.config.cache_dir + host_uri
        self.args += (self.host_file,)

        running = self.factory.runningFetchers
        if not postconverting or running.has_key(host_uri):
            #Make sure that the file is there
            loop = LoopbackRequest(request, self.host_transfer_done)
            loop.uri = host_uri
            loop.local_file = self.host_file
            loop.process()
            self.loop_req = loop
            loop.serve_if_cached=0
            if running.has_key(host_uri):
                #the file is on it's way, wait for it.
                running[host_uri].insert_request(loop)
            else:
                #we are not postconverting, so we need to fetch the host file.
                loop.fetch(serve_cached=0)
        else:
            #The file should be there already.
            self.loop_req = None
            self.host_transfer_done()

    def host_transfer_done(self):
        """
        Called by our LoopbackRequest when the real Fetcher calls
        finish() on it.

        If everything went well, check mtimes and only do the work if needed.

        If posible arrange things so the target file gets the same mtime as
        the host file.
        """
        log.debug('transfer done', 'gzip')
        if self.loop_req and self.loop_req.code != http.OK:
            self.setResponseCode(self.loop_req.code,
                                 self.loop_req.code_message)
            self.apDataReceived("")
            self.apDataEnd("")
            return

        if os.path.exists(self.host_file):
            self.local_mtime = os.stat(self.host_file)[stat.ST_MTIME]
        old_mtime = None
        if os.path.exists(self.local_file):
            old_mtime = os.stat(self.local_file)[stat.ST_MTIME]
        if self.local_mtime == old_mtime:
            self.apEndCached()
        else:
            log.debug("Starting process: " + self.exe + " " + str(self.args), 'gzip')
            self.process = reactor.spawnProcess(self, self.exe, self.args)

    def outReceived(self, data):
        self.setResponseCode(http.OK)
        self.apDataReceived(data)

    def errReceived(self, data):
        log.debug('gzip: ' + data,'gzip')

    def loseConnection(self):
        """
        This is a bad workaround Process.loseConnection not doing it's
        job right.
        The problem only happends when we try to finish the process
        while decompresing.
        """
        if hasattr(self, 'process') and self.process.pid:
            try:
                os.kill(self.process.pid, signal.SIGTERM)
                self.process.connectionLost()
            except exceptions.OSError, Error:
                import errno
                (Errno, Errstr) = Error
                if Errno != errno.ESRCH:
                    log.debug('Passing OSError exception '+Errstr)
                    raise 
                else:
                    log.debug('Threw away exception OSError no such process')

    def processEnded(self, reason=None):
        __pychecker__ = 'unusednames=reason'
        log.debug("Status: %d" %(self.process.status),'gzip')
        if self.process.status != 0:
            self.setResponseCode(http.NOT_FOUND)

        self.apDataReceived("")
        self.apDataEnd(self.transfered)

class RsyncFetcher(protocol.ProcessProtocol):
    """
    Fetch a file using the rsync protocol
    rsync is run as an external process
    """
    rsyncCommand = '/usr/bin/rsync'
    pipelining = False
    def __init__(self, backendServer):
        self.backendServer = backendServer
        self.rsyncProcess = None

    def connect(self):
        # We can't connect seperately so just return true
        return defer.succeed(True)

    def download(self, fetcher, uri, mtime):
        """
        Request download
        %param fetcher: Fetcher class to receive callbacks
        %param uri: URI of file to be downloaded within backend
        %param mtime: Modification time of current file in cache
        """
        self.rsyncTempFile = None # Temporary filename that rsync streams to
        self.bytes_sent = 0 #Number of bytes sent to client already
        self.parent = fetcher
        self.cache_mtime = mtime
        self.request_uri = uri
        self.cache_path = fetcher.cacheEntry.cache_path
        self.file_path = fetcher.cacheEntry.file_path   # Absolute path of file
        self.cache_dir = fetcher.cacheEntry.filedir
        self.remote_file = (self.backendServer.path + '/' 
                            + uri)

        # Change /path/to/FILE -> /path/to/.FILE.* to match rsync tempfile
        self.globpattern = re.sub(r'/([^/]*)$', r'/.\1.*', self.file_path)

        for file in glob.glob(self.globpattern):
          log.msg('Deleting stale tempfile:' + file, 'rsyncFetcher')
          unlink(file)

        # rsync needs the destination directory in place, so create it if necessary
        if not os.path.exists(self.cache_dir):
            os.makedirs(self.cache_dir)

        if self.backendServer.port:
            portspec = ':' + str(self.backendServer.port)
        else:
            portspec = ''

        uri = 'rsync://'+ self.backendServer.host + portspec \
              +self.backendServer.path+'/' + self.request_uri

        args = [self.rsyncCommand, '--partial', '--progress', '--times',
                '--timeout=%s' %(self.backendServer.backend.config.timeout)]
        if log.isEnabled('rsync',9):
            args.append('--verbose')
        else:
            args.append('--quiet')
        bwlimit = self.backendServer.backend.config.bandwidth_limit
        if bwlimit:
            bwlimit = bwlimit / 1000 # rsync wants kbps
            if bwlimit < 1: 
                bwlimit = 1
            args.append('--bwlimit=%d' % (bwlimit))
        args.extend([uri, '.'])
        log.debug('rsync command: (%s) %s' %(self.cache_dir, string.join(args,' ')), 'rsyncFetcher')
        self.rsyncProcess = reactor.spawnProcess(self, self.rsyncCommand, args, None,
                                            self.cache_dir)

    def findRsyncTempFile(self):
        """
        Look for temporary file created by rsync during streaming
        """
        files = glob.glob(self.globpattern)

        if len(files)==1:
            self.rsyncTempFile = files[0]
            log.debug('tempfile: ' + self.rsyncTempFile, 'rsync_client')
        elif not files:
            # No file created yet
            pass
        else:
            log.err('found more than one tempfile, abort rsync')
            self.parent.fetcher_internal_error("Found more than one rsync temporary file")

    #def connectionMade(self):
    #    pass

    def outReceived(self, data):
        "Data received from rsync process to stdout"
        for s in string.split(data, '\n'):
            if len(s):
                log.debug('rsync: ' + s, 'rsync_client')
        #self.apDataReceived(data)
        if not self.rsyncTempFile:
            self.findRsyncTempFile()
            # Got tempfile?
            #if self.rsyncTempFile:
            #    self.setResponseCode(http.OK)
        if self.rsyncTempFile:
            self.sendData()

    def errReceived(self, data):
        "Data received from rsync process to stderr"
        for s in string.split(data, '\n'):
            if len(s):
                log.err('rsync error: ' + s, 'rsync_client')

    def sendData(self):
        f = None
        if self.rsyncTempFile:
            try:
                f = open(self.rsyncTempFile, 'rb')
            except IOError:
                return
        else:
            # Tempfile has gone, stream main file
            log.debug("sendData open dest (sent: %s bytes)"% (self.bytes_sent), 'rsync_client')
            f = open(self.file_path, 'rb')

        if f:
            f.seek(self.bytes_sent)
            data = f.read(abstract.FileDescriptor.bufferSize)
            #log.debug("sendData got " + str(len(data)))
            f.close()
            if data:
                self.parent.data_received(data, save=False)
                self.bytes_sent = self.bytes_sent + len(data)
                reactor.callLater(0, self.sendData)
            elif not self.rsyncTempFile:
                # Finished reading final file
                log.debug("sendData complete. Bytes sent: %s" %(self.bytes_sent))
                # Tell clients, but data is already saved by rsync so don't
                # write file again
                self.parent.download_complete()
                #self.parent.connection_closed() # We don't have a persistent connection

    def processEnded(self, status_object):
        __pychecker__ = 'unusednames=reason'
        self.rsyncTempFile = None
        self.rsyncProcess = None

        r = status_object.trap(error.ProcessTerminated, error.ProcessDone)
        if r == error.ProcessDone:
            log.debug("rsync process complete", 'rsync_client')
            # File received.  Send to clients.
            self.parent.server_mtime(os.stat(self.file_path)[stat.ST_MTIME])
            reactor.callLater(0, self.sendData)
        elif r == error.ProcessTerminated:
            log.debug("Status: %s" %(status_object.value.exitCode)
                      ,'rsync_client')
            exitcode = status_object.value.exitCode
            if exitcode == 10:
                # Host not found
                self.parent.connection_failed('rsync connection to %s failed'
                                                % (self.backendServer.host))
            elif exitcode == 23:
                self.parent.file_not_found()
            else:
                self.parent.fetcher_internal_error("Error in rsync")

    def disconnect(self):
        "Kill rsync process"
        if self.rsyncProcess and self.rsyncProcess.pid:
            log.debug("disconnect: killing rsync child pid " + 
                      str(self.rsyncProcess.pid), 'rsync_client')
            os.kill(self.rsyncProcess.pid, signal.SIGTERM)
            self.transport.loseConnection()

class DownloadQueue:
    """
    This class manages a list of files to download and schedules downloads
    """
    closeTimeout = 5 # Time to close fetcher connections after last download (seconds)
    def __init__(self, parent = None):
        """
        Initialise download queue
        @param parent Class to notify (see downloadQueueEmpty) when queue is empty [class, data]
        """
        #import traceback
        #traceback.print_stack()
        self.queue = [] # List of cacheEntry classes waiting
        self.activeFile = None
        self.fetcher = None
        self.timeoutCB = None
        if parent is not None:
            self.parent, self.parentId = parent
        else:
            self.parent = None

    def addFile(self, cacheEntry):
        """
        Add a file to the queue and start downloading if necessary
        @param cacheEntry Cache entry of file to download
        @return Deferred that is triggered when file has been downloaded
        """
        if len(self.queue) == 0 and self.timeoutCB is not None:
            self.timeoutCB.cancel()
            self.timeoutCB = None
        self.queue.append(cacheEntry)
        if self.activeFile is None:
            self.startNextDownload()
        else:
            log.debug("queue file " + cacheEntry.cache_path, 'DownloadQueue')

    def downloadFinished(self, result):
        success, message = result
        if success:
            log.debug("download complete: %s" % (self.activeFile.cache_path), 'DownloadQueue')
        else:
            log.debug("download failed: %s" % (message), 'DownloadQueue')
        self.activeFile = None
        self.startNextDownload()

    def startNextDownload(self):
        while len(self.queue)>0:
            self.activeFile = self.queue[0]
            self.queue = self.queue[1:]

            if self.activeFile.state != self.activeFile.STATE_NEW:
                log.debug("active download skipped (%s)" % (self.activeFile.cache_path), 'DownloadQueue')
                self.activeFile = None
                continue # Go to next file

            log.debug("start next download (%s)" % (self.activeFile.cache_path), 'DownloadQueue')

            if self.fetcher is not None:
                if self.fetcher.backendServer.backend != self.activeFile.backend:
                    log.debug("old:%s new:%s" %(self.fetcher.backendServer.backend,self.activeFile.backend) 
                              , 'DownloadQueue')
                    log.debug("changing backend server", 'DownloadQueue')
                    self.fetcher.disconnect()
                    self.fetcher = Fetcher()
                else:
                    log.debug("keeping backend server", 'DownloadQueue')
            else:
                log.debug("creating new fetcher", 'DownloadQueue')
                self.fetcher = Fetcher()
            d = self.fetcher.start(self.activeFile)
            d.addCallback(self.downloadFinished)
            return

        # Download queue was empty
        #twisted.internet.base.DelayedCall.debug = True
        log.debug("download queue is empty", 'DownloadQueue')
        if self.closeTimeout and self.fetcher is not None:
            self.timeoutCB = reactor.callLater(self.closeTimeout, self.closeFetcher)
        else:
            self.closeFetcher()

    def closeFetcher(self):
        "Close active fetcher - called after queue has been empty for closeTimeout seconds"

        self.timeoutCB = None
        if self.fetcher is not None:
            log.debug("closing fetcher [%s]" % (self.fetcher.backendServer), 'DownloadQueue')
            self.fetcher.disconnect()
            self.fetcher = None

            if self.parent is not None:
                self.parent.downloadQueueEmpty(self, self.parentId)

    def stop(self):
        log.debug("queue stop", 'DownloadQueue')
        if self.timeoutCB is not None:
            self.timeoutCB.cancel()
        self.closeFetcher()

class DownloadQueuePerClient:
    """
    DownloadQueue that creates several queues, one per client
    """
    def __init__(self):
        self.queues = {}

    def addFile(self, cacheEntry):
        # Add queue entries for all clients. The client
        # queue that is ready first will start the download
        for req in cacheEntry.requests:
            clientId = req.getFileno()

            if self.queues.has_key(clientId):
                q = self.queues[clientId]
            else:
                q = DownloadQueue([self, clientId])
                self.queues[clientId] = q
                log.debug("Adding new queue for client id %s" % (clientId), 'DownloadQueuePerClient')

            q.addFile(cacheEntry)

    def downloadQueueEmpty(self, queue, data):
        """
        DownloadQueue notifies that it is empty
        """
        log.debug("Removing queue for client id %s" % (data), 'DownloadQueuePerClient')
        del self.queues[data]

    def stop(self):
        for q in self.queues.values():
            q.stop()