def get(self):
        """Get the results of a search.

        @raise SearchError: Raised if a query could not be resolved (because
            Solr returned an error).
        @return: A C{Deferred} that fires with C{dict} that maps L{Query}
            instances to search results.
        """
        # Raise errors found when resolving special results.
        for result in self._specialResults.values():
            if isinstance(result, SearchError):
                return fail(result)

        deferreds = []
        for query in self._queries:
            deferreds.append(self._index.search(query))
        deferreds = DeferredList(deferreds, consumeErrors=True)

        def unpackValues(values):
            results = dict(self._specialResults)
            for i, (success, value) in enumerate(values):
                query = self._queries[i]
                if not success:
                    # FIXME If there's more than one exception we'll
                    # effectively ignore all but the first one with this
                    # logic.  It would be good if we didn't ignore/hide issues
                    # like this.
                    value.raiseException()
                results[query] = value
            return results

        return deferreds.addCallback(unpackValues)

def send_catch_log_deferred(signal=Any, sender=Anonymous, *arguments, **named):
    """Like send_catch_log but supports returning deferreds on signal handlers.
    Returns a deferred that gets fired once all signal handlers deferreds were
    fired.
    """

    def logerror(failure, recv):
        if dont_log is None or not isinstance(failure.value, dont_log):
            logger.error(
                "Error caught on signal handler: %(receiver)s",
                {"receiver": recv},
                exc_info=failure_to_exc_info(failure),
                extra={"spider": spider},
            )
        return failure

    dont_log = named.pop("dont_log", None)
    spider = named.get("spider", None)
    dfds = []
    for receiver in liveReceivers(getAllReceivers(sender, signal)):
        d = maybeDeferred(robustApply, receiver, signal=signal, sender=sender, *arguments, **named)
        d.addErrback(logerror, receiver)
        d.addBoth(lambda result: (receiver, result))
        dfds.append(d)
    d = DeferredList(dfds)
    d.addCallback(lambda out: [x[1] for x in out])
    return d

    def stop(self):
        log.info("\n")
        log.info('end-of-execution-stopping-consumers')
        # Ask each of our consumers to stop. When a consumer fully stops, it
        # fires the deferred returned from its start() method. We saved all
        # those deferreds away (above, in start()) in self._consumer_d_list,
        # so now we'll use a DeferredList to wait for all of them...
        for consumer in self._consumer_list:
            consumer.stop()
        dl = DeferredList(self._consumer_d_list)

        # Once the consumers are all stopped, then close our client
        def _stop_client(result):
            if isinstance(result, Failure):
                log.error('error', result=result)
            else:
                log.info('all-consumers-stopped', client=self._client)
            self._client.close()
            return result

        dl.addBoth(_stop_client)

        # And once the client is shutdown, stop the reactor
        def _stop_reactor(result):
            reactor.stop()
            return result

        dl.addBoth(_stop_reactor)

 def send_payment_request(self, readTokens, writeTokens):
   """Called by a Circuit object when it wants to actually make a payment
   @param readTokens:  the number of read tokens to pay for at each hop in the circuit
   @type  readTokens:  int
   @param writeTokens:  the number of read tokens to pay for at each hop in the circuit
   @type  writeTokens:  int"""
   assert (readTokens + writeTokens) / Globals.CELLS_PER_PAYMENT, "tried to pay for bad number of cells"
   #make sure our setup is done:
   if not self.setupDone:
     #have we even started?
     if not self.setupStarted:
       self.send_setup_message()
     self.queuedReadTokens += readTokens
     self.queuedWriteTokens += writeTokens
     return
   #dont bother trying to send payments for circuits that are already closed
   if self.circ.is_done():
     return
   #send the payments
   deferreds = []
   for paymentStream in self.paymentStreams.values():
     deferreds.append(paymentStream.send_payment(readTokens, writeTokens))
   paymentsDoneDeferred = DeferredList(deferreds)
   paymentsDoneDeferred.addErrback(self.generic_error_handler)
   addTokensDeferred = Deferred()
   self.inflightReadTokens += readTokens
   self.inflightWriteTokens += writeTokens
   #timeout in case the payment fails.  We will close the circuit in this case.
   event = Scheduler.schedule_once(PaymentStream.PAR_TIMEOUT, self.all_receipts_received, None, addTokensDeferred, readTokens, writeTokens, None)
   paymentsDoneDeferred.addCallback(self.all_receipts_received, addTokensDeferred, readTokens, writeTokens, event)
   addTokensDeferred.addCallback(self._add_tokens_callback, readTokens, writeTokens)
   addTokensDeferred.addErrback(self.generic_error_handler)

def start():
    """
    
    """
    global _StartingDeferred
    if _StartingDeferred:
        lg.warn('driver.start already called')
        return _StartingDeferred
    if _Debug:
        lg.out(_DebugLevel - 6, 'driver.start')
    dl = []
    for name in boot_up_order():
        svc = services().get(name, None)
        if not svc:
            raise ServiceNotFound(name)
        if not svc.enabled():
            continue
        if svc.state == 'ON':
            continue
        d = Deferred()
        dl.append(d)
        svc.automat('start', d)
    if len(dl) == 0:
        return succeed(1)
    _StartingDeferred = DeferredList(dl)
    _StartingDeferred.addCallback(on_started_all_services)
    return _StartingDeferred

    def process_cluster_info(self, info, cluster, callback):
        """
        process data received from ganeti.
        """
        print '%s:' % cluster.hostname
        infos = json.loads(info)
        self.timer.tick('info fetched from ganeti     ')
        updated = Counter()
        base = cluster.nodes.all()
        mtimes = base.values_list('hostname', 'id', 'mtime')

        data = {}
        for hostname, id, mtime in mtimes:
            data[hostname] = (id, float(mtime) if mtime else None)
        self.timer.tick('mtimes fetched from db       ')

        deferreds = [self.update_node(cluster, info, data, updated) for info in infos]
        deferred_list = DeferredList(deferreds)

        # batch update the cache updated time for all Nodes in this cluster. This
        # will set the last updated time for both Nodes that were modified and for
        # those that weren't.  even if it wasn't modified we want the last
        # updated time to be up to date.
        #
        # XXX don't bother checking to see whether this query needs to run.  With
        # normal usage it will almost always need to
        def update_timestamps(result):
            print '    updated: %s out of %s' % (updated, len(infos))
            base.update(cached=datetime.now())
            self.timer.tick('records or timestamps updated')
        deferred_list.addCallback(update_timestamps)
        deferred_list.addCallback(callback)

        return deferred_list

	def client_connected( protocol ):
		proxy = Proxy( Test_Stub( protocol ), Math_Stub( protocol ))

		request = EchoRequest()
		request.text = "Hello world!"
		echoed = proxy.Test.Echo( request )
		echoed.addCallback( print_response )

		request = PingRequest()
		pinged = proxy.Test.Ping( request )
		pinged.addCallback( print_response )

		request = MathBinaryOperationRequest()
		request.first = 2;
		request.second = 2;
		mathAddd = proxy.Math.Add( request )
		mathAddd.addCallback( print_response )

		mathMultiplyd = proxy.Math.Multiply( request )
		mathMultiplyd.addCallback( print_response )

		dl = DeferredList( [ echoed, pinged, mathAddd, mathMultiplyd ] )
		dl.addCallback( client_finished )

		return dl

 def test_complex2(self, runtime):
     def check(ls):
         if (2 == runtime.id) or (1 == runtime.id):
             self.assertEquals(ls[0][1], "V1")
             self.assertEquals(ls[1][1], "V1")
             self.assertEquals(ls[2][1], "V1")
             self.assertEquals(ls[3][1], "V2")
         else:
             self.assertEquals(ls[0][1], "V1")
             self.assertEquals(ls[1][1], "V1")
             self.assertEquals(ls[2][1], "V1")
             self.assertEquals(ls[3][1], "V2")
             self.assertEquals(ls[4][1], "V2")
     field = self.Zp
     results = []
     results += runtime.broadcast(runtime.players.keys(), runtime.players.keys(), "V1")
     if runtime.id in [1, 2]:
         v = runtime.broadcast([1, 2], [3], "V2")
         if isinstance(v, list):
             results += v
         else:
             results.append(v)
     else:
         results += runtime.broadcast([1, 2], [3])
     if 3 == runtime.id:
         results += [runtime.broadcast([3], runtime.players.keys(), str(7))]
     else:
         results += [runtime.broadcast([3], runtime.players.keys())]
     dls = DeferredList(results)
     runtime.schedule_callback(dls, check)
     dls.addErrback(runtime.error_handler)
     return dls

    def test_send_two_senders_in_parallel(self, runtime):
        """Test of send a value."""
        self.Zp = GF(6277101735386680763835789423176059013767194773182842284081)

        def check(ls):
            for s, x in ls:
                self.assertEquals(int(x), 42)
            return ls

        value = 42

        receivers = [2, 3]
        if 1 == runtime.id:
            d1 = runtime.broadcast([1], receivers, str(value))
        else:
            d1 = runtime.broadcast([1], receivers)

        if 2 == runtime.id:
            d2 = runtime.broadcast([2], [3], str(value))
        else:
            d2 = runtime.broadcast([2], [3])

        ds = [d1]
        if [] != d2:
            ds.append(d2)
        dls = DeferredList(ds)
        dls.addCallback(check)
        return dls

 def run(self):
     jobs, self._jobs = self._jobs[:], []
     jobs_done = DeferredList(jobs)
     jobs_done.addBoth(lambda ignore: self._thread_pool.stop())
     jobs_done.addBoth(lambda ignore: reactor.stop())
     reactor.callWhenRunning(self._thread_pool.start)
     reactor.run(self._install_signal_handlers)

    def subscribe(self):

        def _logFailure(failure):
            log.debug("reported {message}", message=failure.getErrorMessage())
            return failure

        def _logGrantedQoS(value):
            log.debug("response {value!r}", value=value)
            return True

        def _logAll(*args):
            log.debug("all subscriptions complete args={args!r}",args=args)

        d1 = self.protocol.subscribe("foo/bar/baz1", 2 )
        d1.addCallbacks(_logGrantedQoS, _logFailure)

        d2 = self.protocol.subscribe("foo/bar/baz2", 2 )
        d2.addCallbacks(_logGrantedQoS, _logFailure)

        d3 = self.protocol.subscribe("foo/bar/baz3", 2 )
        d3.addCallbacks(_logGrantedQoS, _logFailure)

        dlist = DeferredList([d1,d2,d3], consumeErrors=True)
        dlist.addCallback(_logAll)
        return dlist

def report_crash(request):
    def finishRequest(_):
        # TODO: this should return a meaningful error code
        request.write(
            '<?xml version="1.0" encoding="UTF-8"?><result>0</result>')

    xml = request.args.get('xmlstring', '')[0]
    crashes = BeautifulSoup(xml).findAll('crash')

    deferreds = []
    for crashXML in crashes:
        crash = db.Crash(
            applicationname=crashXML.applicationname.text,
            bundleidentifier=crashXML.bundleidentifier.text,
            contact=crashXML.contact.text,
            description=crashXML.description.text,
            log=crashXML.log.text,
            platform=crashXML.platform.text,
            senderversion=crashXML.senderversion.text,
            systemversion=crashXML.systemversion.text,
            user=crashXML.userid.text,
            version=crashXML.version.text
        )
        deferreds.append(crash.save())
    deferredList = DeferredList(deferreds)
    return deferredList.addCallback(finishRequest)

 def _checkBodies(self, responses, callback, *params):
   deferreds = [Deferred() for r in responses]
   for i, (s, r) in enumerate(responses):
     r.deliverBody(PrinterClient(deferreds[i]))
   dl = DeferredList(deferreds)
   dl.addCallback(callback, *params)
   return dl

 def enqueue(self):
     now = int(time.time())
     # Compare the heap min timestamp with now().
     # If it's time for the item to be queued, pop it, update the
     # timestamp and add it back to the heap for the next go round.
     queue_items = []
     if self.amqp_queue_size < 100000:
         queue_items_a = queue_items.append
         LOGGER.debug("%s:%s" % (self.heap[0][0], now))
         while self.heap[0][0] < now and len(queue_items) < 1000:
             job = heappop(self.heap)
             uuid = UUID(bytes=job[1][0])
             if not uuid.hex in self.unscheduled_items:
                 queue_items_a(job[1][0])
                 new_job = (now + job[1][1], job[1])
                 heappush(self.heap, new_job)
             else:
                 self.unscheduled_items.remove(uuid.hex)
     else:
         LOGGER.critical('AMQP queue is at or beyond max limit (%d/100000)'
             % self.amqp_queue_size)
     # add items to amqp
     if queue_items:
         LOGGER.info('Found %d new uuids, adding them to the queue'
             % len(queue_items))
         msgs = [Content(uuid) for uuid in queue_items]
         deferreds = [self.chan.basic_publish(
             exchange=self.amqp_exchange, content=msg) for msg in msgs]
         d = DeferredList(deferreds, consumeErrors=True)
         d.addCallbacks(self._addToQueueComplete, self._addToQueueErr)
     else:
         self.enqueueCallLater = reactor.callLater(1, self.enqueue)

    def tearDown(self):

        deferreds = []        
        deferreds.append(self.spider.shutdown())
        d = DeferredList(deferreds)
        d.addCallback(self._tearDownCallback)
        return d 

    def provision(self, request):
        """
        Provision the device with credentials from a cloud controller.
        """
        cors.config_cors(request)
        body = json.loads(request.content.read().decode('utf-8'))
        routerId = body['routerId']
        apitoken = body['apitoken']
        pdserver = body['pdserver']
        wampRouter = body['wampRouter']

        changed = False
        if routerId != nexus.core.info.pdid \
            or pdserver != nexus.core.info.pdserver \
            or wampRouter != nexus.core.info.wampRouter:
            if pdserver and wampRouter:
                nexus.core.provision(routerId, pdserver, wampRouter)
            else:
                nexus.core.provision(routerId)
            changed = True

        if apitoken != nexus.core.getKey('apitoken'):
            nexus.core.saveKey(apitoken, 'apitoken')
            changed = True

        if changed:
            PDServerRequest.resetToken()
            nexus.core.jwt_valid = False

            def set_update_fetcher(session):
                session.set_update_fetcher(self.update_fetcher)

            @inlineCallbacks
            def start_polling(result):
                yield self.update_fetcher.start_polling()

            def send_response(result):
                response = dict()
                response['provisioned'] = True
                response['httpConnected'] = nexus.core.jwt_valid
                response['wampConnected'] = nexus.core.wamp_connected
                request.setHeader('Content-Type', 'application/json')
                return json.dumps(response)

            wampDeferred = nexus.core.connect(WampSession)
            wampDeferred.addCallback(set_update_fetcher)

            httpDeferred = sendStateReport()
            httpDeferred.addCallback(start_polling)

            identDeferred = sendNodeIdentity()

            dl = DeferredList([wampDeferred, httpDeferred, identDeferred],
                    consumeErrors=True)
            dl.addBoth(send_response)
            reactor.callLater(6, dl.cancel)
            return dl
        else:
            return json.dumps({'success': False,
                               'message': 'No change on the provision parameters'})

	def connect(self):
		self.servers = []
		serverDeferreds = []
		for connection_info in self.connection_list:
			try:
				if type(connection_info) == dict:
					def addServer(res):
						self.servers.append(res)
						return res
					d = redis.Connection(**connection_info)
					d.addCallback(addServer)
					serverDeferreds.append(d)
				else:
					server = connection_info
					self.servers.append(server)
			except Exception as e:
				raise Warning(str(e))

		def checkQuorun(res):
			self.quorum = (len(self.connection_list) // 2) + 1
			if len(self.servers) < self.quorum:
				raise CannotObtainLock(
					"Failed to connect to the majority of redis servers")
			return res
		dl = DeferredList(serverDeferreds)
		dl.addCallback(checkQuorun)
		return dl

    def _configure_ssh(self, deployment):
        """
        :return: A ``Deferred`` which fires when all nodes have been configured
            with ssh keys.
        """
        self.ssh_configuration.create_keypair()
        results = []
        for node in deployment.nodes:
            results.append(
                deferToThread(
                    self.ssh_configuration.configure_ssh,
                    node.hostname, self.ssh_port
                )
            )
        d = DeferredList(results, fireOnOneErrback=True, consumeErrors=True)

        # Exit with ssh's output if it failed for some reason:
        def got_failure(failure):
            if failure.value.subFailure.check(CalledProcessError):
                raise SystemExit(
                    b"Error connecting to cluster node: " +
                    failure.value.subFailure.value.output)
            else:
                return failure

        d.addErrback(got_failure)
        return d

 def tearDown(self):
     LOGGER.removeHandler(self.logging_handler)
     a = self.mini_web_server.shutdown()
     b = self.pg.clearCache()
     d = DeferredList([a, b])
     d.addCallback(self._tearDownCallback)
     return d

    def cleanUp(self, wasClean, code, reason):
        """Thorough clean-up method to cancel all remaining deferreds, and send
        connection metrics in"""
        self.ps.metrics.increment("client.socket.disconnect", tags=self.base_tags)
        elapsed = (ms_time() - self.ps.connected_at) / 1000.0
        self.ps.metrics.timing("client.socket.lifespan", duration=elapsed, tags=self.base_tags)

        # Cleanup our client entry
        if self.ps.uaid and self.ap_settings.clients.get(self.ps.uaid) == self:
            del self.ap_settings.clients[self.ps.uaid]

        # Cancel any outstanding deferreds that weren't already called
        for d in self.ps._callbacks:
            if not d.called:
                d.cancel()

        # Attempt to deliver any notifications not originating from storage
        if self.ps.direct_updates:
            defers = []
            if self.ps.use_webpush:
                for notifs in self.ps.direct_updates.values():
                    notifs = filter(lambda x: x.ttl != 0, notifs)
                    defers.extend(map(self._save_webpush_notif, notifs))
            else:
                for chid, version in self.ps.direct_updates.items():
                    defers.append(self._save_simple_notif(chid, version))

            # Tag on the notifier once everything has been stored
            dl = DeferredList(defers)
            dl.addBoth(self._lookup_node)

        # Delete and remove remaining dicts and lists
        del self.ps.direct_updates
        del self.ps.updates_sent