def RequestAvatars(self, contacts):
        def simpleIterate(contacts):
            if len(contacts) > 0:
                for handle_id in contacts:
                    handle = self.handle(telepathy.HANDLE_TYPE_CONTACT, handle_id)

                    d = getPage(str(self.avatars_urls[handle.name]['avatar']), timeout=20)
                    d.addCallback(self.on_fetch_avatars_ok, handle)
                    d.addErrback(self.on_fetch_avatars_failed, handle)

                    yield d
        coiterate(simpleIterate(contacts))

    def chat(self, container):
        if container.message.startswith("/"):
            coiterate(
                self.transport.write(make_packet("chat", message=line))
                for line in self.factory.run_command(container.message[1:])
            )
        else:
            message = "<%s> %s" % (self.username, container.message)
            print message

            packet = make_packet("chat", message=message)
            self.factory.broadcast(packet)

 def handle_conflicts(self, ids):
     self.info("Detected %d conflicts", len(ids))
     if ids:
         return itask.coiterate(
             (self.conflict_cb(doc_id) for doc_id in ids))
     else:
         self.resolve_alert(ALERT_NAME, 'ok')

    def enable_cache(self, size):
        """
        Set the permanent cache size.

        Changing the size of the cache sets off a series of events which will
        empty or fill the cache to make it the proper size.

        For reference, 3 is a large-enough size to completely satisfy the
        Notchian client's login demands. 10 is enough to completely fill the
        Notchian client's chunk buffer.

        :param int size: The taxicab radius of the cache, in chunks
        """

        log.msg("Setting cache size to %d..." % size)

        self.permanent_cache = set()
        def assign(chunk):
            self.permanent_cache.add(chunk)

        x = self.spawn[0] // 16
        z = self.spawn[2] // 16

        rx = xrange(x - size, x + size)
        rz = xrange(z - size, z + size)
        d = coiterate(self.request_chunk(x, z).addCallback(assign)
            for x, z in product(rx, rz))
        d.addCallback(lambda chaff: log.msg("Cache size is now %d" % size))

    def post_tube_offer(self, tube, tube_conn):
        service = tube.props[CHANNEL_TYPE_DBUS_TUBE + ".ServiceName"]
        if service == BUS_NAME:
            self.coherence.dbus.add_to_connection(tube_conn, OBJECT_PATH)
            self.coherence_tube = tube_conn
        elif service == DEVICE_IFACE:
            self.device_tube = tube_conn
        elif service == SERVICE_IFACE:
            self.service_tube = tube_conn

        if not self.announce_done and None not in (self.coherence_tube,
                                                   self.device_tube,
                                                   self.service_tube):
            self.announce_done = True

            def iterate(devices):
                for device in devices:
                    yield self._register_device(device)

            def done(result):
                bus = self.coherence.dbus.bus
                bus.add_signal_receiver(self._media_server_found,
                                        "UPnP_ControlPoint_MediaServer_detected")
                bus.add_signal_receiver(self._media_server_removed,
                                        "UPnP_ControlPoint_MediaServer_removed")

            dfr = task.coiterate(iterate(self.coherence.dbus.devices.values()))
            dfr.addCallback(lambda gen: done)

def iterateInReactor(i, delay=None):
    """
    Cooperatively iterate over the given iterator.

    @see: L{twisted.internet.task.coiterate}.
    """
    return coiterate(i)

    def retrain(self):
        """
        Force all L{iquotient.IHamFilter}s to forget their trained state,
        then retrain them based on L{exmess.Message}s with C{trained} set to
        C{True}, then reclassify all messages.

        This should only be called in the batch process.
        """
        filters = list(self.store.powerupsFor(iquotient.IHamFilter))
        for f in filters:
            f.forgetTraining()

        sq = MailboxSelector(self.store)
        sq.setLimit(5000)
        sq.refineByStatus(TRAINED_STATUS)
        work = iter(list(sq))


        # XXX This really should use in-database state, otherwise a restart in
        # the middle will muck things up.
        def go():
            for msg in work:
                for f in filters:
                    f.train(msg._spam, msg)
                yield None
            self.reclassify()
        return coiterate(go())

    def parse_data(self, root):

        def iterate(root):
            for item in root.findall('./movieinfo'):
                trailer = self._parse_into_trailer(item)
                yield trailer

        return task.coiterate(iterate(root))

 def send_messages():
     def message_iterator():
         for i in range(count):
             content = body + "-%d" % i
             msg = Content(content)
             msg["delivery mode"] = 2
             chan.basic_publish(exchange="chatservice", content=msg, routing_key="txamqp_chatroom")
             print "Sending message: %s" % content
             yield None
     return task.coiterate(message_iterator())

    def _list_files(self):
        """
        Get a detailed listing of the current directory
        """

        file_list = FTPFileListProtocol()
        d = self.list('.', file_list)
        d.addCallback(lambda ignore: task.coiterate(self._retrieve_files(file_list)))
        d.addErrback(self._list_files_errback)
        return d

    def _consume_iterator(self, iterator):
        results = []
        errors = []

        def collect_result(d):
            return d.addCallbacks(lambda result: results.append(result), lambda f: errors.append(f))

        work = imap(collect_result, iterator)
        ds = [task.coiterate(work) for i in range(0,10)]
        return defer.gatherResults(ds).addCallback(lambda ign: (results, errors))

 def remote_handlePublicMessage(self, protocol, user, channel, message, encoding, max_line_length):
     try:
         if message.startswith("!"):
             return handleCommand(protocol, user, channel, message[1:], encoding, max_line_length)
         else:
             callback = functools.partial(protocol.callRemote, "msg", channel)
             handler = MessageHandler(
                 self.reactor, self.good_urls, self.bad_urls, message, callback, encoding, max_line_length
             )
             return task.coiterate(iter(handler))
     except Exception:
         log.err()

def run(command, arguments, max_processes=None, stdout=sys.stdout):
    if max_processes is None:
        max_processes = multiprocessing.cpu_count()

    processes = (
        VergeProcess.spawn(format_command(command, argument), stdout=stdout)
        for argument in arguments
    )

    return defer.gatherResults(
        coiterate(processes) for _ in xrange(max_processes)
    )

 def validate(self, value):
     if not value:
         return super(Tuple, self).validate(None)
     def driver():
         for (f,v) in zip(self.fields, value):
             yield defer.maybeDeferred(f.validate, v).addCallback(result.append)
     # Map the items to their validated versions.
     result = []
     d = task.coiterate(driver())
     # Call the super class with the result.
     d.addCallback(lambda ignore: super(Tuple, self).validate(tuple(result)))
     return d

    def get_children(self, start=0, request_count=0):
        tracks = []

        def query_db():
            rows = self.get_tracks(request_count)
            for row in rows:
                track = self.db_to_didl(row)
                tracks.append(track)
                yield track

        dfr = task.coiterate(query_db())
        dfr.addCallback(lambda gen: tracks)
        return dfr

    def enable_cache(self):
        """
        Start up a rudimentary permanent cache.
        """

        self.permanent_cache = set()
        def assign(chunk):
            self.permanent_cache.add(chunk)

        rx = xrange(self.spawn[0] - 3, self.spawn[0] + 3)
        rz = xrange(self.spawn[2] - 3, self.spawn[2] + 3)
        d = coiterate(assign(self.load_chunk(x, z)) for x, z in product(rx, rz))
        d.addCallback(lambda chaff: log.msg("Cache is warmed up!"))

    def get_devices_async(self,dbus_async_cb,dbus_async_err_cb):
        infos = []

        def iterate_devices(devices):
            for device in devices:
                infos.append(device.get_info())
                yield infos

        def done(generator):
            dbus_async_cb(dbus.Array(infos, signature='v', variant_level=2))

        devices = self.devices.copy().values()
        dfr = task.coiterate(iterate_devices(devices))
        dfr.addCallbacks(done, lambda failure: dbus_async_err_cb(failure.value))

    def update_chunks(self):
        print "Sending chunks..."
        x, chaff, z, chaff = split_coords(self.player.location.x,
            self.player.location.z)

        new = set(product(xrange(x - 10, x + 10), xrange(z - 10, z + 10)))
        old = set(self.chunks.iterkeys())
        added = new - old
        discarded = old - new

        # Perhaps some explanation is in order.
        # The generator expressions are stored in the protocol instance. If we
        # need to cancel them, we can call their close() method, which causes
        # them to become inert. This is incredibly important because we want
        # to cancel all previously pending chunk changes when a new set of
        # chunk changes is requested.
        # The coiterate() function iterates over the iterable it is fed,
        # without tying up the reactor, by yielding after each iteration. The
        # inner part of the generator expression generates all of the chunks
        # around the currently needed chunk, and it sorts them by distance to
        # the current chunk. The end result is that we load chunks one-by-one,
        # nearest to furthest, without stalling other clients.
        if self.chunk_generators:
            for generator in self.chunk_generators:
                generator.close()

        self.chunk_generators = [
            (
                self.enable_chunk(i, j) for i, j in
                sorted(added, key=lambda t: (t[0] - x)**2 + (t[1] - z)**2)
            ),
            (self.disable_chunk(i, j) for i, j in discarded)
        ]

        for generator in self.chunk_generators:
            coiterate(generator)

    def get_videos(self):
        videos = []

        def query_db():
            source_id = self.get_local_video_library_id()
            q = "select * from CoreTracks where TrackID in " \
                "(select distinct(TrackID) from CoreTracks where " \
                "PrimarySourceID=?)"
            for row in self.db.sql_execute(q, source_id):
                video = Video(row, self.db, source_id)
                videos.append(video)
                yield video

        dfr = task.coiterate(query_db())
        dfr.addCallback(lambda gen: videos)
        return dfr

    def get_artists(self):
        artists = []

        def query_db():
            source_id = self.get_local_music_library_id()
            q = "select * from CoreArtists where ArtistID in " \
                "(select distinct(ArtistID) from CoreTracks where " \
                "PrimarySourceID=?) order by Name"
            for row in self.db.sql_execute(q, source_id):
                artist = Artist(row, self.db, source_id)
                artists.append(artist)
                yield artist

        dfr = task.coiterate(query_db())
        dfr.addCallback(lambda gen: artists)
        return dfr