def http_download(destination, url):
    class FileWriter(Protocol):
        def __init__(self, fp, fin):
            self.fp = fp
            self.tmp = fp.temporarySibling(".tmp")
            self.fh = self.tmp.open("w")
            self.fin = fin

        def dataReceived(self, bytes):
            self.fh.write(bytes)

        def connectionLost(self, reason):
            self.fh.close()
            if isinstance(reason.value, ResponseDone):
                self.tmp.moveTo(self.fp)
                self.fin.callback(None)
            else:
                self.fin.errback(reason)

    log.msg("Downloading jquery from {0}".format(url))

    agent = Agent(reactor)

    d = agent.request("GET", url)

    def gotResponse(response):
        finished = Deferred()
        response.deliverBody(FileWriter(destination, finished))
        return finished
    d.addCallback(gotResponse)
    return d

def requestNewPage(headers, url, external_id):
    agent = Agent(reactor)
    request_d = agent.request('GET', url,Headers(headers),None)
    d = defer.Deferred()
    d.addCallback(parseNewPage, external_id)
    request_d.addCallback(downloadPage, d)
    return request_d

def _upload(reactor, url, project, revision, revision_date, benchmark,
            executable, environment, result_value, result_date, std_dev,
            max_value, min_value):
    data = {
        'commitid': str(revision),
        'revision_date': revision_date,
        'project': project,
        'benchmark': benchmark,
        'environment': environment,
        'executable': executable,
        'result_value': str(result_value),
        'result_date': result_date,
        'std_dev': str(std_dev),
        'max': str(max_value),
        'min': str(min_value),
        }
    print('uploading', data)
    agent = Agent(reactor)
    d = agent.request('POST', url, None, StringProducer(urlencode(data)))
    def check(response):
        d = readBody(response)
        def read(body):
            print('body', repr(body))
            if response.code != 200:
                raise Exception("Upload failed: %r" % (response.code,))
        d.addCallback(read)
        return d
    d.addCallback(check)
    return d

    def get(self, base_url, args = {}, username = None, password = None):

        #pool = HTTPConnectionPool(reactor, persistent=True)
        #pool.maxPersistentPerHost = 3
        #pool.retryAutomatically = False

        agent   = Agent(reactor)
        headers = { 'User-Agent': [self._USER_AGENT_] }

        if username:
            authorization            = b64encode(username + ":" + password)
            headers['Authorization'] = [authorization]

        url = self._get_url(base_url, args)

        self.log.info('Requesting URL: %s' % url)

        d_agent  = agent.request(
            'GET',
            url,
            Headers(headers),
            None)

        d_agent.addCallback(self.cb_agent)
        d_agent.addErrback(self.cb_agent_err)

        self.d = defer.Deferred()

        return self.d

    def test_round_trip(self):
        d = Deferred()
        self.worker.deliver_hook = lambda x: d.callback(None)

        self.worker.startWorker()
        self.test_worker.startWorker()

        params = {
            'username': 'user',
            'password': 'pass',
            'owner': 'owner',
            'service': 'service',
            'subservice': 'subservice',
            'call-number': '+27831234567',
            'origin': '12345',
            'messageid': 'message_id',
            'provider': 'provider',
            'tariff': 0,
            'text': 'message content',
        }

        agent = Agent(reactor)
        response = yield agent.request(
            'POST', self.config['url'], Headers({
                'User-Agent': ['Vumi Vas2Net Transport'],
                'Content-Type': ['application/x-www-form-urlencoded'],
            }), StringProducer(urlencode(params)))

        log.msg('Headers', list(response.headers.getAllRawHeaders()))
        self.assertTrue(response.headers.hasHeader('X-Nth-Smsid'))

        yield d

    def _process_queue(self):
        if self.searches.under_processing:
            return False
        if not self.searches.request_queue:
            return False
        self.searches.under_processing = True

        self.searches.errors = {}
        self.searches.tweets = []
        self.searches.got_twt_data = []
        self.searches.processed_request = self.searches.request_queue.pop(0)
        self.tweet_count = 0

        debug_msg(str(self.searches.processed_request['search_spec']))

        auth_sys = AuthProcessor()
        params_urlized = auth_sys.get_search_params(self.searches.processed_request['search_spec'])
        req_headers = auth_sys.get_headers(params_urlized)

        contextFactory = TwtClientContextFactory()
        agent = Agent(reactor, contextFactory)

        d = agent.request(
            'GET',
            'https://api.twitter.com/1.1/search/tweets.json' + '?' + params_urlized,
            Headers(req_headers),
            None)

        borders = TwtResponseBorders(self, self.searches)
        d.addCallback(borders.cbRequest)
        d.addBoth(borders.cbShutdown)

        return True

def check_timeouts():
    # the server we should poke, defined at the top of this file
    global TARGET_TIMEOUT_SERVER
    
    sessions = Session.objects.get_timedout_sessions()
    
    for session in sessions:
        agent = Agent(reactor)
        
        print "Timing out session: %s (%d)" % (session.task.name, session.id)

        payload_dict = {
            'patient': session.patient.id,
            'session': session.id
            }

        payload = "&".join(map(lambda x: "%s=%s" % (x, payload_dict[x]), payload_dict))

        d = agent.request(
            'POST',
            TARGET_TIMEOUT_SERVER,
            Headers({
                    "Content-Type": ["application/x-www-form-urlencoded;charset=utf-8"],
                    "Content-Length": [str(len(payload))]
                    }),
            StringProducer(payload))

        d.addCallback(session_timeout_finished, sessionid=session.id)
        d.addErrback(session_timeout_errored, sessionid=session.id)
        
    # run again in a bit
    reactor.callLater(30, check_timeouts)

    def testCreateCommitProvision(self):

        agent = Agent(reactor)

        header = Headers({'User-Agent': ['OpenNSA Test Client'], 'Host': ['localhost'] } )

        payload = { "source" : "aruba:topology:ps?vlan=1783",
                    "destination" : "aruba:topology:bon?vlan=1783",
                    "auto_commit" : False
                }
        payload_data = json.dumps(payload)

        create_url = 'http://localhost:%i%s' % (self.PORT, rest.PATH)
        producer = FileBodyProducer(StringIO(payload_data))

        resp = yield agent.request('POST', create_url, header, producer)

        self.failUnlessEqual(resp.code, 201, 'Service did not return created')
        if not resp.headers.hasHeader('location'):
            self.fail('No location header in create response')

        conn_url = 'http://localhost:%i%s' % (self.PORT, resp.headers.getRawHeaders('location')[0])

        # so... the connection will not necesarely have moved into reserveheld or all sub-connections might not even be in place yet
        # we cannot really commit until we are in created and ReserveHeld
        # the clock doesn't really do anything here (not scheduling related)

        yield task.deferLater(reactor, 0.1, self._createCommitProvisionCB, agent, conn_url, header)

    def httpsRequest(self, url, headers={}, method='GET', data=None):
        headers['Content-Type'] = ['application/json']
        if url.startswith('https'):
            agent = Agent(reactor, WebClientContextFactory())
        else:
            agent = Agent(reactor)

        if data:
            data = StringProducer(data)

        request = yield agent.request(
            method,
            url,
            Headers(headers),
            data
        )

        if request.length:
            d = defer.Deferred()
            request.deliverBody(BodyReceiver(d))
            body = yield d

            defer.returnValue(json.loads(body.read()))
        else:
            defer.returnValue(None)

def get_page(url, method="GET", payload=None, headers=None):
    """Downloads the page from given URL, using asynchronous networking"""
    agent = Agent(reactor)

    producer = None
    if payload:
        producer = StringProducer(payload)

    _headers = {"User-Agent": [settings.USER_AGENT]}
    if headers:
        for key, value in headers.items():
            _headers[key] = [value]

    response = (yield agent.request(method, str(url), Headers(_headers), producer))

    # for h in response.headers.getAllRawHeaders():
    #    print h

    try:
        finished = defer.Deferred()
        (yield response).deliverBody(ResponseCruncher(finished))
    except:
        raise Exception("Downloading page '%s' failed" % url)

    defer.returnValue((yield finished))

def http_get(uri):
    """
    Performs a GET request
    :param uri: The URL to perform a GET request to
    :return: A deferred firing the body of the response.
    :raises HttpError: When the HTTP response code is not OK (i.e. not the HTTP Code 200)
    """
    def _on_response(response):
        if response.code == http.OK:
            return readBody(response)
        if response.code == http.FOUND:
            # Check if location header contains magnet link
            location_headers = response.headers.getRawHeaders("location")
            if not location_headers:
                return fail(Failure(RuntimeError("HTTP redirect response does not contain location header")))
            new_uri = location_headers[0]
            if new_uri.startswith('magnet'):
                _, infohash, _ = parse_magnetlink(new_uri)
                if infohash:
                    return succeed(new_uri)
            return http_get(new_uri)
        raise HttpError(response)

    try:
        contextFactory = WebClientContextFactory()
        agent = Agent(reactor, contextFactory)
        headers = Headers({'User-Agent': ['Tribler ' + version_id]})
        deferred = agent.request('GET', uri, headers, None)
        deferred.addCallback(_on_response)
        return deferred
    except:
        return fail()

    def _do_request(self, method, document, **params):
        url = "%s/%s" % (self.API_SERVER, document)
        headers = self._get_headers()

        if method == "GET":
            # FIXME: Get rid of this
            if document in ["bugreport.json", "tefrequest.json", "version.json"]:
                url += "?q=" + urllib.quote(json.dumps(params))
            else:
                url += "?" + urllib.urlencode(params)
            producer = None
        elif method == "POST":
            producer = StringProducer(urllib.urlencode(params))
            headers["Content-Type"] = ["application/x-www-form-urlencoded"]
        else:
            raise AssertionError(method)

        log.info("Requsting %s %s %r" % (method, url, headers))

        agent = Agent(reactor)
        d = agent.request(method, url, Headers(headers), producer)

        def dataReceived(response):
            finished = Deferred()
            response.deliverBody(JsonDownloader(finished))
            return finished

        d.addCallback(dataReceived)
        return d

def check_schedule():
    tasks = ScheduledTask.objects.get_due_tasks()
    
    for sched_task in tasks:
        agent = Agent(reactor)
        
        print "Executing task: ", sched_task.task.name
        
        payload = "user=%d&task=%d&arguments=%s" % (sched_task.user.id, sched_task.task.id, json.dumps(sched_task.arguments))
        print payload
        d = agent.request(
            'POST',
            # ullr?
            #'http://ullr:8001/taskmanager/exec',
            'http://localhost:8001/taskmanager/exec',
            Headers({
                    "Content-Type": ["application/x-www-form-urlencoded;charset=utf-8"],
                    "Content-Length": [str(len(payload))]
                    }),
            StringProducer(payload))

        d.addCallback(task_finished, sched_taskid=sched_task.id)
        d.addErrback(task_errored, sched_taskid=sched_task.id)
        
    # run again in a bit
    reactor.callLater(5, check_schedule)

    def _make_short_url(self, long_url):
        self.request_in_flight = True

        api_uri = "https://git.io/"
        encoded = urllib.urlencode({"url": long_url})
        body_producer = StringProducer(encoded)

        agent = Agent(reactor)
        d = agent.request('POST', api_uri, bodyProducer=body_producer)

        def onRequestComplete(data):
            self._onRequestComplete()
            return data

        def onResponse(response):
            if response.code != 201:
                onRequestComplete(None)
                return long_url

            self._onRequestComplete()
            return response.headers.getRawHeaders("Location")[-1]
        d.addCallback(onResponse)

        def onError(failure):
            return long_url
        d.addErrback(onError)
        d.addErrback(onRequestComplete)

        return d

def getMarket(card):
    cached = globals()['gMarket_Cached']
    if card.marketComments in cached and card.marketReviews in cached:
        d1 = defer.Deferred()
        d1.addCallback(lambda x: cached[card.marketComments])
        d1.callback(None)
        d2 = defer.Deferred()
        d2.addCallback(lambda x: cached[card.marketReviews])
        d2.callback(None)
        return defer.DeferredList((d1,d2))

    agent = Agent(reactor)
    headers = {}
    for k,v in standard_headers.items():
        if k == 'User-Agent':
            headers.update({'User-Agent':[standard_user_agents[randint(0,len(standard_user_agents)-1)]]})
        else:
            headers.update({k:v})

    c = defer.Deferred()
    c.addCallback(parseMarket, url=card.marketComments)
    r = defer.Deferred()
    r.addCallback(parseMarket, url=card.marketReviews)


    request_comments = agent.request('GET', str(card.marketComments),Headers(),None)
    request_comments.addCallback(downloadPage, c)
    request_reviews = agent.request('GET', str(card.marketReviews),Headers(),None)
    request_reviews.addCallback(downloadPage, r)
    return defer.DeferredList((c,r))

 def _make_agents(self, auth_files):
     """
     Configure the web clients that:
     * perform backchannel CAS ticket validation
     * proxy the target site
     """
     self.connectionPool = HTTPConnectionPool(self.reactor)
     if auth_files is None or len(auth_files) == 0:
         agent = Agent(self.reactor, pool=self.connectionPool)
     else:
         extra_ca_certs = []
         for ca_cert in auth_files:
             with open(ca_cert, "rb") as f:
                 data = f.read()
             cert = crypto.load_certificate(crypto.FILETYPE_PEM, data)
             del data
             extra_ca_certs.append(cert)
         policy = CustomPolicyForHTTPS(extra_ca_certs)
         agent = Agent(self.reactor, contextFactory=policy, pool=self.connectionPool)
     if self.proxy_client_endpoint_s is not None:
         self.proxyConnectionPool = HTTPConnectionPool(self.reactor)
         self.proxy_agent = Agent.usingEndpointFactory(
             self.reactor,
             WebClientEndpointFactory(self.reactor, self.proxy_client_endpoint_s),
             pool=self.proxyConnectionPool)
     else:
         self.proxy_agent = agent
     if self.cas_client_endpoint_s is not None:
         self.casConnectionPool = HTTPConnectionPool(self.reactor)
         self.cas_agent = Agent.usingEndpointFactory(
             self.reactor,
             WebClientEndpointFactory(self.reactor, self.cas_client_endpoint_s),
             pool=self.casConnectionPool) 
     else:
         self.cas_agent = agent

    def render_GET(self, request):
        def fail(failure):
            request.write('we failed %s' % failure)
            request.finish()

        def return_body(body):
            """Called when we have a full response"""
            response = {'web-request': body}
            response = json.dumps(response, ensure_ascii=False, encoding='utf-8').encode('utf-8')
            request.write(response)
            request.finish()

        def get_body(result):
            # now that we have the body, 
            # we can return the result, using ready body
            # which is also a async operation.
            d2 = readBody(result)  # get the  contents of the page.
            d2.addCallback(return_body)
            d2.addErrback(fail)

        # setup the deferred/callback for the first asynchronous 
        # call...
        agent = Agent(reactor)
        d1 = agent.request('GET', 'http://example.com/')
        d1.addCallback(get_body)
        d1.addErrback(fail)

        return server.NOT_DONE_YET

    def __init__(self, auth_url, username, api_key, pool=None, proxy=None,
                 extra_headers=None, verbose=False):
        self.auth_url = auth_url
        self.username = username
        self.api_key = api_key
        self.storage_url = None
        self.auth_token = None
        self.pool = pool

        if proxy:
            if ":" in proxy:
                addr, port = proxy.rsplit(":", 1)
                port = int(port)
            else:
                addr, port = proxy, 8000

            endpoint = TCP4ClientEndpoint(reactor, addr, port)
            self.agent = ProxyAgent(endpoint, pool=self.pool)
        else:
            contextFactory = WebClientContextFactory()
            contextFactory.noisy = False
            self.agent = Agent(reactor, contextFactory, pool=self.pool)

        self.extra_headers = extra_headers
        self.verbose = verbose

class Music:
    def __init__(self):
        self.recenttime=0
        self.player = Agent(reactor)
        self.timePath = networking.musicPlayer.path("time")
        
    def current_time(self):
        """return deferred which gets called with the current
        time. This gets called really often"""
        d = self.player.request("GET", self.timePath)
        d.addCallback(self._timeReturned)
        return d

    def _timeReturned(self, response):
        done = Deferred()
        done.addCallback(self._bodyReceived)
        response.deliverBody(GatherJson(done))
        return done

    def _bodyReceived(self, data):
        if 't' in data:
            dispatcher.send("input time", val=data['t'])
        if 'song' in data and data['song']:
            dispatcher.send("current_player_song", song=URIRef(data['song']))
        return data['t'] # pass along to the real receiver
    
    def playOrPause(self, t=None):
        if t is None:
            # could be better
            self.current_time().addCallback(lambda t: self.playOrPause(t))
        else:
            self.player.request("POST",
                                networking.musicPlayer.path("seekPlayOrPause"),
                                bodyProducer=StringProducer(json.dumps({"t" : t})))

def request(method, url, data=None, session_id=None):
    agent = Agent(reactor)
    headers = {
        'Cookie': ['XSRF-TOKEN=antani;'],
        'X-XSRF-TOKEN': ['antani']
    }
    if session_id:
        headers['X-Session'] = [str(session_id)]

    bodyProducer = None
    if data:
        bodyProducer = StringProducer(json.dumps(data))
    
    try:
        response = yield agent.request(method, str(base_url + url), 
                                       Headers(headers), bodyProducer)
    except Exception as exc:
        failed(exc, method, url, data)
        raise exc

    try:
        content_length = response.headers.getRawHeaders('content-length')
    except IndexError:
        content_length = None

    finished = defer.Deferred()
    response.deliverBody(BodyReceiver(finished, content_length))

    response_body = yield finished
    try:
        d = json.loads(response_body)
    except Exception as exc:
        failed(exc, method, url, data, response, response_body)

    defer.returnValue(d)