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 run_queue(self):
        ''' If possible run the first test in queue '''

        # Adapted from neubot/rendezvous/client.py

        if not self.queue:
            return
        if self.running:
            return

        #
        # Subscribe BEFORE starting the test, otherwise we
        # may miss the 'testdone' event if the connection
        # to the negotiator service fails, and we will stay
        # stuck forever.
        #
        NOTIFIER.subscribe('testdone', self.test_done)

        # Prevent concurrent tests
        self.running = True

        # Safely run first element in queue
        deferred = Deferred()
        deferred.add_callback(self._do_run_queue)
        deferred.add_errback(self._run_queue_error)
        deferred.callback(self.queue[0])

    def run_queue(self):
        ''' If possible run the first test in queue '''

        # Adapted from neubot/rendezvous/client.py

        if not self.queue:
            return
        if self.running:
            return

        #
        # Subscribe BEFORE starting the test, otherwise we
        # may miss the 'testdone' event if the connection
        # to the negotiator service fails, and we will stay
        # stuck forever.
        #
        NOTIFIER.subscribe('testdone', self.test_done)

        # Prevent concurrent tests
        self.running = True

        # Safely run first element in queue
        try:
            self._do_run_queue()
        except (SystemExit, KeyboardInterrupt):
            raise
        except:
            exc = asyncore.compact_traceback()
            error = str(exc)
            LOG.error('runner_core: catched exception: %s' % error)
            NOTIFIER.publish('testdone')

 def handle_connection_lost(stream):
     ''' Invoked when the connection is lost '''
     final_state = 0
     context = stream.opaque
     if context:
         extra = context.extra
         if extra:
             final_state = extra['final_state']
     if not final_state:
         logging.warning('skype_negotiate: not reached final state')
     NOTIFIER.publish('testdone')  # Tell the runner we're done

 def _api_debug(self, stream, request, query):
     response = Message()
     debuginfo = {}
     NOTIFIER.snap(debuginfo)
     POLLER.snap(debuginfo)
     debuginfo["queue_history"] = QUEUE_HISTORY
     debuginfo["WWW"] = WWW
     stringio = StringIO.StringIO()
     pprint.pprint(debuginfo, stringio)
     stringio.seek(0)
     response.compose(code="200", reason="Ok", body=stringio,
                      mimetype="text/plain")
     stream.send_response(request, response)

    def _api_state(self, stream, request, query):
        dictionary = cgi.parse_qs(query)

        t = None
        if dictionary.has_key("t"):
            t = dictionary["t"][0]
            stale = NOTIFIER.needs_publish(STATECHANGE, t)
            if not stale:
                NOTIFIER.subscribe(STATECHANGE, self._api_state_complete,
                                   (stream, request, query, t), True)
                return

        self._api_state_complete(STATECHANGE, (stream, request, query, t))

    def _api_state(self, stream, request, query):
        ''' Implements /api/state URI '''
        dictionary = cgi.parse_qs(query)

        otime = None
        if "t" in dictionary:
            otime = dictionary["t"][0]
            stale = NOTIFIER.needs_publish(STATECHANGE, otime)
            if not stale:
                NOTIFIER.subscribe(STATECHANGE, self._api_state_complete,
                                   (stream, request, query, otime), True)
                return

        self._api_state_complete(STATECHANGE, (stream, request, query, otime))

 def _api_debug(stream, request, query):
     ''' Implements /api/debug URI '''
     response = Message()
     debuginfo = {}
     NOTIFIER.snap(debuginfo)
     POLLER.snap(debuginfo)
     debuginfo["queue_history"] = QUEUE_HISTORY
     debuginfo["WWWDIR"] = utils_hier.WWWDIR
     gc.collect()
     debuginfo['typestats'] = objgraph.typestats()
     body = pprint.pformat(debuginfo)
     response.compose(code="200", reason="Ok", body=body,
                      mimetype="text/plain")
     stream.send_response(request, response)

    def test_bittorrent_invokation_bad(self):
        ''' Verify run_queue() behavior when bittorrent is invoked
            and there is NOT a URI for bittorrent '''

        #
        # The whole point of this test is to make sure that
        # the callback() is invoked and the "testdone" event
        # has been fired, when we try to run a bittorrent
        # test and we don't have a registered URI for such
        # test.
        #

        # We need to ensure callback() is invoked
        callback = [0]
        def on_callback():
            ''' Register callback() invokation '''
            # pylint: disable=W0613
            callback[0] += 1

        CONFIG.conf['privacy.can_publish'] = 1
        CONFIG.conf['privacy.informed'] = 1
        CONFIG.conf['privacy.can_collect'] = 1
        core = RunnerCore()
        core.queue.append(('bittorrent', on_callback, None))
        core.run_queue()

        # Worked as expected?
        self.assertTrue(callback[0])
        self.assertFalse(NOTIFIER.is_subscribed("testdone"))

    def test_wrong_privacy(self):
        ''' Verify run_queue() behavior when privacy is wrong '''

        #
        # The whole point of this test is to make sure
        # that privacy.complain() is invoked and "testdone"
        # is published when privacy settings are not OK
        # and a test is started.
        #

        # We need to ensure privacy.complain() is invoked
        privacy_complain = [0]
        def on_privacy_complain():
            ''' Register privacy.complain() invokation '''
            privacy_complain[0] += 1

        # Setup (we will restore that later)
        saved_complain = privacy.complain
        privacy.complain = on_privacy_complain

        CONFIG.conf['privacy.informed'] = 0
        core = RunnerCore()
        core.queue.append(('foo', '/', lambda: None))
        core.run_queue()

        # Restore
        privacy.complain = saved_complain

        # Worked as expected?
        self.assertTrue(privacy_complain[0])
        self.assertFalse(NOTIFIER.is_subscribed("testdone"))

    def test_safety_net(self):
        ''' Verify run_queue() safety net works '''

        #
        # The whole point of this test is to make sure
        # that and error is printed and "testdone" is
        # published when a new test is started and the
        # test name is bad.
        #

        # We need to ensure LOG.error() is invoked
        log_error = [0]
        def on_log_error(message, *args):
            ''' Register LOG.error() invokation '''
            # pylint: disable=W0613
            log_error[0] += 1

        # Setup (we will restore that later)
        saved_log_error = LOG.error
        LOG.error = on_log_error

        CONFIG.conf['privacy.can_publish'] = 1
        CONFIG.conf['privacy.informed'] = 1
        CONFIG.conf['privacy.can_collect'] = 1
        core = RunnerCore()
        core.queue.append(('foo', '/', lambda: None))
        core.run_queue()

        # Restore
        LOG.error = saved_log_error

        # Worked as expected?
        self.assertTrue(log_error[0])
        self.assertFalse(NOTIFIER.is_subscribed("testdone"))

def run(poller, conf):
    '''
     This function is invoked when Neubot is already
     running and you want to leverage some functionalities
     of this module.
    '''

    # Make sure the conf makes sense before we go
    config.finalize_conf(conf)

    if conf["bittorrent.listen"]:
        if conf["bittorrent.negotiate"]:

            #
            # We assume that the caller has already started
            # the HTTP server and that it contains our negotiator
            # so here we just bring up the test server.
            #
            server = ServerPeer(poller)
            server.configure(conf)
            server.listen((conf["bittorrent.address"],
                           conf["bittorrent.port"]))

        else:
            server = PeerNeubot(poller)
            server.configure(conf)
            server.listen((conf["bittorrent.address"],
                           conf["bittorrent.port"]))

    else:

        #
        # Make sure there is someone ready to receive the
        # "testdone" event.  If there is noone it is a bug
        # none times out of ten.
        #
        if not NOTIFIER.is_subscribed("testdone"):
            log.oops("The 'testdone' event is not subscribed")

        if conf["bittorrent.negotiate"]:
            client = BitTorrentClient(poller)
            client.configure(conf)

            #
            # The rendezvous client uses this hidden variable
            # to pass us the URI to connect() to (the rendezvous
            # returns an URI, not address and port).
            #
            uri = None
            if "bittorrent._uri" in conf:
                uri = conf["bittorrent._uri"]

            client.connect_uri(uri)

        else:
            client = PeerNeubot(poller)
            client.configure(conf)
            client.connect((conf["bittorrent.address"],
                           conf["bittorrent.port"]))

 def connection_lost(self, stream):
     if NOTIFIER.is_subscribed("testdone"):
         LOG.debug("RendezVous: don't _schedule(): test in progress")
         return
     if self._task:
         LOG.debug("RendezVous: don't _schedule(): we already have a task")
         return
     self._schedule()

def main(args):

    config.register_descriptions()
    common.main("bittorrent", "Neubot BitTorrent module", args)
    conf = CONFIG.copy()
    config.finalize_conf(conf)

    if conf["bittorrent.listen"]:

        #
        # If we need to negotiate and we're runing
        # standalone we also need to bring up the
        # global HTTP server.
        #
        if conf["bittorrent.negotiate"]:
            HTTP_SERVER.configure(conf)
            HTTP_SERVER.listen((conf["bittorrent.address"],
                               conf["bittorrent.negotiate.port"]))
            conf["negotiate.listen"] = True
            negotiate.run(POLLER, conf)

        #
        # Drop privileges after listen() so we can
        # bind() to privileged ports
        #
        if conf["bittorrent.daemonize"]:
            system.change_dir()
            system.go_background()
            LOG.redirect()

        system.drop_privileges(LOG.error)

    else:
        #
        # When we're connecting to a remote host to perform a test
        # we want Neubot to quit at the end of the test.  When this
        # happens the test code publishes the "testdone" event, so
        # here we prepare to intercept the event and break our main
        # loop.
        #
        NOTIFIER.subscribe("testdone", lambda event, ctx: POLLER.break_loop())

    run(POLLER, conf)

    POLLER.loop()

    def test_bittorrent_invokation_good(self):
        ''' Verify run_queue() behavior when bittorrent is invoked
            and there is a URI for bittorrent '''

        #
        # The whole point of this test is to make sure that
        # bittorrent.run() is invoked when privacy is OK and
        # we have a negotiate URI.  We also want to check that
        # the "testdone" event is subscribed after run_queue(),
        # i.e. that someone is waiting for the event that
        # signals the end of the test.
        #

        # We need to ensure bittorrent.run() is invoked
        bittorrent_run = [0]
        def on_bittorrent_run(poller, conf):
            ''' Register bittorrent.run() invokation '''
            # pylint: disable=W0613
            bittorrent_run[0] += 1

        # Setup (we will restore that later)
        saved_run = bittorrent.run
        bittorrent.run = on_bittorrent_run
        RUNNER_TESTS.update({'bittorrent': '/'})

        CONFIG.conf['privacy.can_publish'] = 1
        CONFIG.conf['privacy.informed'] = 1
        CONFIG.conf['privacy.can_collect'] = 1
        core = RunnerCore()
        core.queue.append(('bittorrent', Deferred(), None))
        core.run_queue()

        # Restore
        bittorrent.run = saved_run
        RUNNER_TESTS.update({})

        # Worked as expected?
        self.assertTrue(bittorrent_run[0])
        self.assertTrue(NOTIFIER.is_subscribed("testdone"))

        #
        # Clear the "testdone" because otherwise it will
        # screw up other tests and we don't want that
        #
        NOTIFIER.publish("testdone")

    def run_queue(self):
        ''' If possible run the first test in queue '''

        # Adapted from neubot/rendezvous/client.py

        if not self.queue:
            return
        if self.running:
            return

        #
        # Subscribe BEFORE starting the test, otherwise we
        # may miss the 'testdone' event if the connection
        # to the negotiator service fails, and we will stay
        # stuck forever.
        #
        NOTIFIER.subscribe('testdone', self.test_done)

        # Prevent concurrent tests
        self.running = True

        # Make a copy of current settings
        conf = CONFIG.copy()

        # Make sure we abide to M-Lab policy
        if privacy.count_valid(conf, 'privacy.') != 3:
            privacy.complain()
            NOTIFIER.publish('testdone')

        # Run speedtest
        elif self.queue[0][0] == 'speedtest':
            conf['speedtest.client.uri'] =  self.queue[0][1]
            client = ClientSpeedtest(POLLER)
            client.configure(conf)
            client.connect_uri()

        # Run bittorrent
        elif self.queue[0][0] == 'bittorrent':
            conf['bittorrent._uri'] =  self.queue[0][1]
            bittorrent.run(POLLER, conf)

        # Safety net
        else:
            LOG.error('Asked to run an unknown test')
            NOTIFIER.publish('testdone')

 def connection_failed(self, connector, exception):
     logging.info("dash: connect() failed: test done")
     NOTIFIER.publish("testdone")

 def connection_lost(self, stream):
     logging.info("dash: negotiate connection closed: test done")
     NOTIFIER.publish("testdone")
     self.client = None
     self.stream = None

    def _maintain_database(self, *args, **kwargs):

        POLLER.sched(INTERVAL, self._maintain_database)

        if (self._use_database and not NOTIFIER.is_subscribed("testdone")):
            self._writeback()

 def connection_lost(self, stream):
     ''' Invoked when the connection is closed or lost '''
     NOTIFIER.publish('testdone')