def do_negotiate(self, stream, request, nodelay=False):
        session = TRACKER.session_negotiate(request["authorization"])
        if not request["authorization"]:
            request["authorization"] = session.identifier

        #
        # XXX make sure we track ALSO the first connection of the
        # session (which is assigned an identifier in session_negotiate)
        # or, should this connection fail, we would not be able to
        # propagate quickly this information because unregister_connection
        # would not find an entry in self.connections{}.
        #
        if session.negotiations == 1:
            TRACKER.register_connection(stream, request["authorization"])
            nodelay = True

        if not session.active:
            if not nodelay:
                NOTIFIER.subscribe(RENEGOTIATE, self._do_renegotiate,
                          (stream, request), True)
                return

        m1 = compat.SpeedtestNegotiate_Response()
        m1.authorization = session.identifier
        m1.unchoked = session.active
        m1.queuePos = session.queuepos
        m1.publicAddress = stream.peername[0]
        s = marshal.marshal_object(m1, "text/xml")

        stringio = StringIO.StringIO(s)
        response = Message()
        response.compose(code="200", reason="Ok",
         body=stringio, mimetype="application/xml")
        stream.send_response(request, response)

 def _on_internal_error(self, stream, request):
     """ Generate 500 Internal Server Error page """
     logging.error("Internal error while serving response", exc_info=1)
     response = Message()
     response.compose(code="500", reason="Internal Server Error", body="500 Internal Server Error", keepalive=0)
     stream.send_response(request, response)
     stream.close()

 def _api_exit(self, stream, request, query):
     POLLER.sched(0, POLLER.break_loop)
     response = Message()
     stringio = StringIO.StringIO("See you, space cowboy\n")
     response.compose(code="200", reason="Ok", body=stringio,
                      mimetype="text/plain", keepalive=False)
     stream.send_response(request, response)

    def _do_negotiate(self, baton):
        ''' Respond to a /negotiate request '''
        stream, request, position = baton

        module = request.uri.replace('/negotiate/', '')
        module = self.modules[module]
        request_body = json.load(request.body)

        parallelism = CONFIG['negotiate.parallelism']
        unchoked = int(position < parallelism)
        response_body = {
                         'queue_pos': position,
                         'real_address': stream.peername[0],
                         'unchoked': unchoked,
                        }
        if unchoked:
            extra = module.unchoke(stream, request_body)
            if not 'authorization' in extra:
                raise RuntimeError('Negotiate API violation')
            extra.update(response_body)
            response_body = extra
        else:
            response_body['authorization'] = ''

        response = Message()
        response.compose(code='200', reason='Ok',
                         body=json.dumps(response_body),
                         keepalive=True,
                         mimetype='application/json')
        stream.send_response(request, response)

def log_api(stream, request, query):
    ''' Implements /api/log '''

    # Get logs and options
    logs = LOG.listify()
    options = cgi.parse_qs(query)

    # Reverse logs on request
    if utils.intify(options.get('reversed', ['0'])[0]):
        logs = reversed(logs)

    # Filter according to verbosity
    if utils.intify(options.get('verbosity', ['1'])[0]) < 2:
        logs = [ log for log in logs if log['severity'] != 'DEBUG' ]
    if utils.intify(options.get('verbosity', ['1'])[0]) < 1:
        logs = [ log for log in logs if log['severity'] != 'INFO' ]

    # Human-readable output?
    if utils.intify(options.get('debug', ['0'])[0]):
        logs = [ '%(timestamp)d [%(severity)s]\t%(message)s\r\n' % log
                 for log in logs ]
        body = ''.join(logs).encode('utf-8')
        mimetype = 'text/plain; encoding=utf-8'
    else:
        body = json.dumps(logs)
        mimetype = 'application/json'

    # Compose and send response
    response = Message()
    response.compose(code='200', reason='Ok', body=body, mimetype=mimetype)
    stream.send_response(request, response)

    def _api_config(self, stream, request, query):
        response = Message()

        indent, mimetype, sort_keys = None, "application/json", False
        dictionary = cgi.parse_qs(query)
        if "debug" in dictionary and utils.intify(dictionary["debug"][0]):
            indent, mimetype, sort_keys = 4, "text/plain", True

        if request.method == "POST":
            s = request.body.read()
            updates = qs_to_dictionary(s)
            privacy.check(updates)

            # Very low barrier to prevent damage from kiddies
            if "agent.interval" in updates:
                interval = int(updates["agent.interval"])
                if interval < 1380 and interval != 0:
                    raise ConfigError("Bad agent.interval")

            CONFIG.merge_api(updates, DATABASE.connection())
            STATE.update("config", updates)
            # Empty JSON b/c '204 No Content' is treated as an error
            s = "{}"
        else:
            s = json.dumps(CONFIG.conf, sort_keys=sort_keys, indent=indent)

        stringio = StringIO.StringIO(s)
        response.compose(code="200", reason="Ok", body=stringio,
                         mimetype=mimetype)
        stream.send_response(request, response)

def api_results(stream, request, query):
    ''' Provide results for queried tests '''
    since, until = -1, -1
    test = ''

    dictionary = cgi.parse_qs(query)

    if dictionary.has_key("test"):
        test = str(dictionary["test"][0])
    if dictionary.has_key("since"):
        since = int(dictionary["since"][0])
    if dictionary.has_key("until"):
        until = int(dictionary["until"][0])

    if test == 'bittorrent':
        table = table_bittorrent
    elif test == 'speedtest':
        table = table_speedtest
    else:
        raise NotImplementedTest("Test '%s' is not implemented" % test)

    indent, mimetype, sort_keys = None, "application/json", False
    if "debug" in dictionary and utils.intify(dictionary["debug"][0]):
        indent, mimetype, sort_keys = 4, "text/plain", True

    response = Message()
    lst = table.listify(DATABASE.connection(), since, until)
    body = json.dumps(lst, indent=indent, sort_keys=sort_keys)
    response.compose(code="200", reason="Ok", body=body, mimetype=mimetype)
    stream.send_response(request, response)

    def start_transaction(self, stream=None):

        #
        # XXX This is complexity at the wrong level of abstraction
        # because the HTTP client should manage more than one connections
        # and we should just pass it HTTP messages and receive events
        # back.
        #

        if not stream:
            endpoint = (self.conf.get("api.client.address", "127.0.0.1"),
                        int(self.conf.get("api.client.port", "9774")))
            self.connect(endpoint)

        else:

            uri = "http://%s:%s/api/state?t=%d" % (
              self.conf.get("api.client.address", "127.0.0.1"),
              self.conf.get("api.client.port", "9774"),
              self.timestamp)

            request = Message()
            request.compose(method="GET", uri=uri)

            stream.send_request(request)

 def _on_internal_error(self, stream, request):
     LOG.exception()
     response = Message()
     response.compose(code="500", reason="Internal Server Error",
                      body="500 Internal Server Error", keepalive=0)
     stream.send_response(request, response)
     stream.close()

def api_results(stream, request, query):
    ''' Populates www/results.html page '''

    dictionary = cgi.parse_qs(query)
    test = CONFIG['www_default_test_to_show']
    if 'test' in dictionary:
        test = str(dictionary['test'][0])

    # Read the directory each time, so you don't need to restart the daemon
    # after you have changed the description of a test.
    available_tests = {}
    for filename in os.listdir(TESTDIR):
        if filename.endswith('.json'):
            index = filename.rfind('.json')
            if index == -1:
                raise RuntimeError('api_results: internal error')
            name = filename[:index]
            available_tests[name] = filename
    if not test in available_tests:
        raise NotImplementedTest('Test not implemented')

    # Allow power users to customize results.html heavily, by creating JSON
    # descriptions with local modifications.
    filepath = utils_path.append(TESTDIR, available_tests[test], False)
    if not filepath:
        raise RuntimeError("api_results: append() path failed")
    localfilepath = filepath + '.local'
    if os.path.isfile(localfilepath):
        filep = open(localfilepath, 'rb')
    else:
        filep = open(filepath, 'rb')
    response_body = json.loads(filep.read())
    filep.close()

    # Add extra information needed to populate results.html selection that
    # allows to select which test results must be shown.
    response_body['available_tests'] = available_tests.keys()
    response_body['selected_test'] = test

    descrpath = filepath.replace('.json', '.html')
    if os.path.isfile(descrpath):
        filep = open(descrpath, 'rb')
        response_body['description'] = filep.read()
        filep.close()

    # Provide the web user interface some settings it needs, but only if they
    # were not already provided by the `.local` file.
    for variable in COPY_CONFIG_VARIABLES:
        if not variable in response_body:
            response_body[variable] = CONFIG[variable]

    # Note: DO NOT sort keys here: order MUST be preserved
    indent, mimetype = None, 'application/json'
    if 'debug' in dictionary and utils.intify(dictionary['debug'][0]):
        indent, mimetype = 4, 'text/plain'

    response = Message()
    body = json.dumps(response_body, indent=indent)
    response.compose(code='200', reason='Ok', body=body, mimetype=mimetype)
    stream.send_response(request, response)

 def connection_ready(self, stream):
     request = Message()
     request.compose(method="GET", pathquery="/speedtest/negotiate",
       host=self.host_header)
     request["authorization"] = self.conf.get(
       "speedtest.client.authorization", "")
     stream.send_request(request)

    def connection_ready(self, stream):
        request = Message()

        #
        # With version 2, we upload bytes using chunked transfer
        # encoding for TARGET seconds.
        #
        if self.conf['version'] == 2:
            body = BytegenSpeedtest(TARGET)
            request.compose(method='POST', chunked=body,
              pathquery='/speedtest/upload', host=self.host_header)
            request['authorization'] = self.conf[
              'speedtest.client.authorization']
            stream.send_request(request)
            self.ticks[stream] = utils.ticks()
            self.bytes[stream] = stream.bytes_sent_tot
            return

        request.compose(method="POST", body=RandomBody(ESTIMATE["upload"]),
          pathquery="/speedtest/upload", host=self.host_header)
        request["authorization"] = self.conf.get(
          "speedtest.client.authorization", "")
        self.ticks[stream] = utils.ticks()
        self.bytes[stream] = stream.bytes_sent_tot
        stream.send_request(request)

def api_data(stream, request, query):
    ''' Get data stored on the local database '''
    since, until = -1, -1
    test = ''

    dictionary = cgi.parse_qs(query)

    if "test" in dictionary:
        test = str(dictionary["test"][0])
    if "since" in dictionary:
        since = int(dictionary["since"][0])
    if "until" in dictionary:
        until = int(dictionary["until"][0])

    if test == 'bittorrent':
        table = table_bittorrent
    elif test == 'speedtest':
        table = table_speedtest
    elif test == 'raw':
        table = table_raw
    else:
        raise NotImplementedTest("Test not implemented")

    indent, mimetype, sort_keys = None, "application/json", False
    if "debug" in dictionary and utils.intify(dictionary["debug"][0]):
        indent, mimetype, sort_keys = 4, "text/plain", True

    response = Message()
    lst = table.listify(DATABASE.connection(), since, until)
    body = json.dumps(lst, indent=indent, sort_keys=sort_keys)
    response.compose(code="200", reason="Ok", body=body, mimetype=mimetype)
    stream.send_response(request, response)

 def send_response(self, m):
     response = Message()
     response.compose(code=m["code"], reason=m["reason"],
       keepalive=m["keepalive"], mimetype=m["mimetype"],
       body=m["response_body"])
     m["stream"].send_response(m["request"], response)
     if not m["keepalive"]:
         m["stream"].close()

 def _serve_request(self, stream, request):
     path, query = urlparse.urlsplit(request.uri)[2:4]
     if path in self._dispatch:
         self._dispatch[path](stream, request, query)
     else:
         response = Message()
         response.compose(code="404", reason="Not Found",
                 body=StringIO.StringIO("404 Not Found"))
         stream.send_response(request, response)

    def process_request(self, stream, request):
        ''' Process a /collect or /negotiate HTTP request '''

        #
        # We always pass upstream the collect request.  If it is
        # not authorized the module does not have the identifier in
        # its global table and will raise a KeyError.
        # Here we always keepalive=False so the HTTP layer closes
        # the connection and we are notified that the queue should
        # be changed.
        #
        if request.uri.startswith('/collect/'):
            module = request.uri.replace('/collect/', '')
            module = self.modules[module]
            request_body = json.load(request.body)

            response_body = module.collect_legacy(stream, request_body, request)
            response_body = json.dumps(response_body)

            response = Message()
            response.compose(code='200', reason='Ok', body=response_body,
                             keepalive=False, mimetype='application/json')
            stream.send_response(request, response)

        #
        # The first time we see a stream, we decide whether to
        # accept or drop it, depending on the length of the
        # queue.  The decision whether to accept or not depends
        # on the current queue length and follows the Random
        # Early Discard algorithm.  When we accept it, we also
        # register a function to be called when the stream is
        # closed so that we can update the queue.  And we
        # immediately send a response.
        # When it's not the first time we see a stream, we just
        # take note that we owe it a response.  But we won't
        # respond until its queue position changes.
        #
        elif request.uri.startswith('/negotiate/'):
            if not stream in self.known:
                position = len(self.queue)
                min_thresh = CONFIG['negotiate.min_thresh']
                max_thresh = CONFIG['negotiate.max_thresh']
                if random.random() < float(position - min_thresh) / (
                                       max_thresh - min_thresh):
                    stream.close()
                    return
                self.queue.append(stream)
                self.known.add(stream)
                stream.atclose(self._update_queue)
                self._do_negotiate((stream, request, position))
            else:
                stream.opaque = request

        # For robustness
        else:
            raise RuntimeError('Unexpected URI')

    def process_request(self, stream, request):
        """ Process the incoming HTTP request """

        if request.uri.startswith("/dash/download"):

            context = stream.opaque
            context.count += 1
            if context.count > DASH_MAXIMUM_REPETITIONS:
                raise RuntimeError("dash: too many repetitions")

            #
            # Parse the "/dash/download/<size>" optional RESTful
            # parameter of the request.
            #
            # If such parameter is not parseable into an integer,
            # we let the error propagate, i.e., the poller will
            # automatically close the stream socket.
            #
            body_size = DASH_DEFAULT_BODY_SIZE
            resource_size = request.uri.replace("/dash/download", "")
            if resource_size.startswith("/"):
                resource_size = resource_size[1:]
            if resource_size:
                body_size = int(resource_size)

            if body_size < 0:
                raise RuntimeError("dash: negative body size")
            if body_size > DASH_MAXIMUM_BODY_SIZE:
                body_size = DASH_MAXIMUM_BODY_SIZE

            #
            # XXX We don't have a quick solution for generating
            # and sending many random bytes from Python.
            #
            # Or, better, we have a couple of ideas, but they
            # have not been implemented into Neubot yet.
            #
            pattern = request["Authorization"]
            if not pattern:
                pattern = "deadbeef"
            body = pattern * ((body_size / len(pattern)) + 1)
            if len(body) > body_size:
                body = body[:body_size]

            response = Message()
            response.compose(code="200", reason="Ok", body=body,
                             mimetype="video/mp4")

            stream.set_timeout(15)

            stream.send_response(request, response)

        else:
            # For robustness
            raise RuntimeError("dash: unexpected URI")

 def _api_index(self, stream, request, query):
     '''
      Redirect either to /index.html or /privacy.html depending on
      whether the user has already set privacy permissions or not
     '''
     response = Message()
     if not privacy.allowed_to_run():
         response.compose_redirect(stream, '/privacy.html')
     else:
         response.compose_redirect(stream, '/index.html')
     stream.send_response(request, response)

 def process_request(self, stream, request):
     try:
         self._serve_request(stream, request)
     except ConfigError, error:
         reason = re.sub(r"[\0-\31]", "", str(error))
         reason = re.sub(r"[\x7f-\xff]", "", reason)
         LOG.exception(func=LOG.info)
         response = Message()
         response.compose(code="500", reason=reason,
                 body=StringIO.StringIO(reason))
         stream.send_response(request, response)

 def _serve_request(self, stream, request):
     ''' Serve incoming request '''
     request_uri = urllib.unquote(request.uri)
     path, query = urlparse.urlsplit(request_uri)[2:4]
     if path in self._dispatch:
         self._dispatch[path](stream, request, query)
     else:
         response = Message()
         response.compose(code="404", reason="Not Found",
                 body="404 Not Found")
         stream.send_response(request, response)