def wait_for_first_response(tasks, converters):
    """given a list of unawaited tasks and non-coro result parsers to be called on the results,
    this function returns the 1st result that is returned and converted

    if it is possible for 2 tasks to complete at the same time,
    only the 1st result deteremined by asyncio.wait will be returned

    returns None if none successfully complete
    returns 1st error raised if any occur (probably)
    """
    primed = [wait_for_result(t, c) for t, c in zip(tasks, converters)]
    done, pending = await asyncio.wait(primed, return_when=asyncio.FIRST_COMPLETED)
    for p in pending:
        p.cancel()

    try:
        return done.pop().result()
    except NotImplementedError as e:
        raise e
    except:
        return None 

def _handler(self, websocket: websockets.WebSocketCommonProtocol, *unused_args):
        """Setup the consumer and producer response handlers with asyncio.

        Args:
            websocket: the websocket connection to the client

        """

        asyncio.set_event_loop(self.loop)
        consumer_task = asyncio.ensure_future(self._consumer_handler(websocket))
        producer_task = asyncio.ensure_future(self._producer_handler(websocket))

        done, pending = await asyncio.wait(
            [consumer_task, producer_task], return_when=asyncio.FIRST_COMPLETED
        )

        for task in pending:
            task.cancel() 

def stop(self):
        """Flush all pending data and close all connections to kafka cluster"""
        if self._closed:
            return
        self._closed = True

        # If the sender task is down there is no way for accumulator to flush
        if self._sender is not None and self._sender.sender_task is not None:
            await asyncio.wait([
                self._message_accumulator.close(),
                self._sender.sender_task],
                return_when=asyncio.FIRST_COMPLETED,
                loop=self._loop)

            await self._sender.close()

        await self.client.close()
        log.debug("The Kafka producer has closed.") 

def _push_error_to_user(self, exc):
        """ Most critical errors are not something we can continue execution
        without user action. Well right now we just drop the Consumer, but
        java client would certainly be ok if we just poll another time, maybe
        it will need to rejoin, but not fail with GroupAuthorizationFailedError
        till the end of days...
        XXX: Research if we can't have the same error several times. For
             example if user gets GroupAuthorizationFailedError and adds
             permission for the group, would Consumer work right away or would
             still raise exception a few times?
        """
        exc = copy.copy(exc)
        self._subscription.abort_waiters(exc)
        self._pending_exception = exc
        self._error_consumed_fut = create_future(loop=self._loop)
        return asyncio.wait(
            [self._error_consumed_fut, self._closing],
            return_when=asyncio.FIRST_COMPLETED,
            loop=self._loop
        ) 

def connect(self, timeout: Optional[float] = None) -> None:
        """Establish a connection with the WebSocket server.

        Returns:
            `(True, <chosen-subprotocol>)` if connection accepted.
            `(False, None)` if connection rejected.

        """
        connected = asyncio.Event()
        self._message_processor = asyncio.ensure_future(
            self._process_messages(connected, timeout or self.TIMEOUT)
        )
        await asyncio.wait(
            [connected.wait(), self._message_processor],
            return_when=asyncio.FIRST_COMPLETED,
        )
        if self._message_processor.done():
            # Make sure to raise an exception from the task.
            self._message_processor.result()
            raise RuntimeError(f"Failed to connect to the server: {self._url}!") 

def onConnect (self, **kwargs):
        self.logger.info ('connect', nick=self.nick, uuid='01f7b138-ea53-4609-88e9-61f3eca3e7e7')

        self.send('NICK', nick=self.nick)
        self.send('USER', user=self.nick, realname='https://github.com/PromyLOPh/crocoite')

        # Don't try to join channels until the server has
        # sent the MOTD, or signaled that there's no MOTD.
        done, pending = await asyncio.wait(
            [self.wait('RPL_ENDOFMOTD'), self.wait('ERR_NOMOTD')],
            loop=self.loop, return_when=asyncio.FIRST_COMPLETED)

        # Cancel whichever waiter's event didn't come in.
        for future in pending:
            future.cancel()

        for c in self.channels:
            self.logger.info ('join', channel=c, uuid='367063a5-9069-4025-907c-65ba88af8593')
            self.send ('JOIN', channel=c)
            # no need for NAMES here, server sends this automatically 

def poll(self, events, timeout=None,
                   return_when=asyncio.FIRST_COMPLETED):
        """Poll for any of a set of event types to be received for this session.
        """
        awaitables = {}
        for name in events:
            awaitables[self.recv(name)] = name
        done, pending = await asyncio.wait(
            awaitables, timeout=timeout, return_when=return_when)

        if done:
            ev_dicts = []
            for fut in done:
                awaitables.pop(fut)
                ev_dicts.append(fut.result())
            return ev_dicts, awaitables.values()
        else:
            raise asyncio.TimeoutError(
                "None of {} was received in {} seconds"
                .format(events, timeout))

    # call control / 'mod_commands' methods
    # TODO: dynamically add @decorated functions to this class
    # and wrap them using functools.update_wrapper ...? 

def cancellable_wait(self, *awaitables: Awaitable[_R], timeout: float = None) -> _R:
        futures = [asyncio.ensure_future(a, loop=self.loop) for a in awaitables + (self.wait(),)]
        try:
            done, pending = await asyncio.wait(
                futures,
                timeout=timeout,
                return_when=asyncio.FIRST_COMPLETED,
                loop=self.loop,
            )
        except CancelledError:
            for future in futures:
                future.cancel()
            raise
        for task in pending:
            task.cancel()
        await asyncio.wait(pending, return_when=asyncio.ALL_COMPLETED, loop=self.loop,)
        if not done:
            raise TimeoutError()
        if self.triggered_token is not None:
            for task in done:
                task.exception()
            raise OperationCancelled(f'Cancellation requested by {self.triggered_token} token')
        return done.pop().result() 

def ws_handler(self, request):

        ws = web.WebSocketResponse()
        await ws.prepare(request)

        if self.ws is not None:
            await self.ws.close()

        self.ws = ws

        _, unfinished = await asyncio.wait(
            [
                self._websocket_receive(ws),
                self._websocket_send(ws)
            ],
            return_when=asyncio.FIRST_COMPLETED
        )
        for task in unfinished:
            task.cancel()

        return ws 

def waiter(client):
    async def wait_for(*events, return_when=asyncio.FIRST_COMPLETED):
        if not events:
            return
        done, pending = await asyncio.wait(
            [client.wait(event) for event in events],
            loop=client.loop,
            return_when=return_when)

        # Get the result(s) of the completed task(s).
        ret = [future.result() for future in done]

        # Cancel any events that didn't come in.
        for future in pending:
            future.cancel()

        # Return list of completed event names.
        return ret
    return wait_for


# taken from :ref:`Patterns` 

def stop(self):
        """Flush all pending data and close all connections to kafka cluster"""
        if self._closed:
            return
        self._closed = True

        # If the sender task is down there is no way for accumulator to flush
        if self._sender is not None and self._sender.sender_task is not None:
            await asyncio.wait([
                self._message_accumulator.close(),
                self._sender.sender_task],
                return_when=asyncio.FIRST_COMPLETED,
                loop=self._loop)

            await self._sender.close()

        await self.client.close()
        log.debug("The Kafka producer has closed.") 

def _push_error_to_user(self, exc):
        """ Most critical errors are not something we can continue execution
        without user action. Well right now we just drop the Consumer, but
        java client would certainly be ok if we just poll another time, maybe
        it will need to rejoin, but not fail with GroupAuthorizationFailedError
        till the end of days...
        XXX: Research if we can't have the same error several times. For
             example if user gets GroupAuthorizationFailedError and adds
             permission for the group, would Consumer work right away or would
             still raise exception a few times?
        """
        exc = copy.copy(exc)
        self._subscription.abort_waiters(exc)
        self._pending_exception = exc
        self._error_consumed_fut = create_future(loop=self._loop)
        return asyncio.wait(
            [self._error_consumed_fut, self._closing],
            return_when=asyncio.FIRST_COMPLETED,
            loop=self._loop
        ) 

def daemon():
    try_init_cgroup()

    async with VJ4Session(config['server_url']) as session:
        while True:
            try:
                await session.login_if_needed(config['uname'], config['password'])
                done, pending = await wait([do_judge(session), do_noop(session)],
                                           return_when=FIRST_COMPLETED)
                for task in pending:
                    task.cancel()
                await gather(*done)
            except Exception as e:
                logger.exception(e)
            logger.info('Retrying after %d seconds', RETRY_DELAY_SEC)
            await sleep(RETRY_DELAY_SEC) 

def _run(self):
        first_completed = concurrent.FIRST_COMPLETED

        if self._get_max_tasks() < 1:
            raise RuntimeError("Executor has no workers")

        try:
            while not self.goal(self.learner):
                futures = self._get_futures()
                done, _ = concurrent.wait(futures, return_when=first_completed)
                self._process_futures(done)
        finally:
            remaining = self._remove_unfinished()
            if remaining:
                concurrent.wait(remaining)
            self._cleanup() 

def _run(self):
        first_completed = asyncio.FIRST_COMPLETED

        if self._get_max_tasks() < 1:
            raise RuntimeError("Executor has no workers")

        try:
            while not self.goal(self.learner):
                futures = self._get_futures()
                done, _ = await asyncio.wait(
                    futures, return_when=first_completed, loop=self.ioloop
                )
                self._process_futures(done)
        finally:
            remaining = self._remove_unfinished()
            if remaining:
                await asyncio.wait(remaining)
            self._cleanup() 

def mqtt_connected(func):
    """
        MQTTClient coroutines decorator which will wait until connection before calling the decorated method.
        :param func: coroutine to be called once connected
        :return: coroutine result
    """
    @asyncio.coroutine
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        if not self._connected_state.is_set():
            base_logger.warning("Client not connected, waiting for it")
            _, pending = yield from asyncio.wait([self._connected_state.wait(), self._no_more_connections.wait()], return_when=asyncio.FIRST_COMPLETED)
            for t in pending:
                t.cancel()
            if self._no_more_connections.is_set():
                raise ClientException("Will not reconnect")
        return (yield from func(self, *args, **kwargs))
    return wrapper 

def __call__(self, receive: Callable, send: Callable) -> None:
        request = self._create_request_from_scope(send)
        receiver_task = asyncio.ensure_future(self.handle_messages(request, receive))
        handler_task = asyncio.ensure_future(self.handle_request(request, send))
        done, pending = await asyncio.wait(
            [handler_task, receiver_task], return_when=asyncio.FIRST_COMPLETED
        )
        await _cancel_tasks(pending)
        _raise_exceptions(done) 

def __call__(self, receive: Callable, send: Callable) -> None:
        websocket = self._create_websocket_from_scope(send)
        receiver_task = asyncio.ensure_future(self.handle_messages(receive))
        handler_task = asyncio.ensure_future(self.handle_websocket(websocket, send))
        done, pending = await asyncio.wait(
            [handler_task, receiver_task], return_when=asyncio.FIRST_COMPLETED
        )
        await _cancel_tasks(pending)
        _raise_exceptions(done) 

def process(self, queue, workflow):
        try:
            while queue.__futures__:
                done, _ = await wait(queue.__futures__,
                                     return_when=FIRST_COMPLETED)
                queue.progress(done)
            return workflow.result()
        except CancelledError:
            for task in queue.__futures__:
                task.cancel()
            await gather(*queue.__futures__)
            raise 

def _connect(self):
        await self.launch_shards()

        while True:
            pollers = [shard.get_future() for shard in self.shards.values()]
            done, _ = await asyncio.wait(pollers, return_when=asyncio.FIRST_COMPLETED)
            for f in done:
                # we wanna re-raise to the main Client.connect handler if applicable
                f.result() 

def process(self):

        try:

            if await self.connect():

                self._listener_task = asyncio.ensure_future(self._ws.receive())
                self._sender_task = asyncio.ensure_future(self._outgoing_queue.get())

                main_tasks = [self._listener_task, self._sender_task]

                done, pending = await asyncio.wait(main_tasks, return_when=asyncio.FIRST_COMPLETED)

                if self._listener_task in done:
                    self.logger.debug('[transport] receiving :  {}'.format(self._listener_task.result()))
                    await self._incoming_queue.put(self._listener_task.result())
                else:
                    # self.logger.debug("listener task pended")
                    self._listener_task.cancel()

                if self._sender_task in done:
                    message = self._sender_task.result()
                    self._ws.send_str(str(message))
                    self.logger.debug('[transport] sending :  {}'.format(message))
                elif self._sender_task:
                    # self.logger.debug("sender tast pended")
                    self._sender_task.cancel()

        except Exception as ex:
            self.logger.error(ex, extra={"tag": "err"})
            traceback.print_exc() 

def test_wait(task):
    done, pending = await asyncio.wait([task], return_when=asyncio.FIRST_COMPLETED)
    for d in done:
        d.exception()
    print("test_wait done")


# Test group 1
# task = asyncio.get_event_loop().create_task(start())
# asyncio.get_event_loop().create_task(test_cancel(task))
# asyncio.get_event_loop().create_task(test_cancel(task))

# Test group 2 

def launch(self):
        event_loop = asyncio.get_event_loop()

        # TODO remove this line once we stop supporting Python 3.5
        asyncio.set_event_loop(event_loop)

        try:
            self._websocket = event_loop.run_until_complete(
                websockets.connect(self._uri))
        except ConnectionRefusedError:
            logger.error('Could not connect to server')
            return

        consumer_task = asyncio.ensure_future(self._consumer_handler())
        tasks = [
            asyncio.ensure_future(self._producer_handler(
                producer_wrapper.producer, producer_wrapper.source_name))
            for producer_wrapper in self.producer_wrappers
        ]
        tasks.append(consumer_task)

        _, pending = event_loop.run_until_complete(asyncio.wait(
                tasks, return_when=asyncio.FIRST_COMPLETED))
        for task in pending:
            task.cancel()
        logger.info('Disconnected From Server') 

def connect(self):
        """Connect Slack RTM."""
        logger = logging.getLogger(f'{__name__}.Bot.connect')
        sleep = 0

        while True:
            try:
                rtm = await self.call('rtm.start')
            except Exception as e:
                logger.exception(e)
                await asyncio.sleep((sleep + 1) * 10)
                sleep += 1
                continue

            if not rtm.body['ok']:
                await asyncio.sleep((sleep + 1) * 10)
                sleep += 1
                continue
            else:
                sleep = 0

            await self.queue.put(create_event('chatterbox_system_start', {}))
            while not self.is_ready:
                await asyncio.sleep(0.01)
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.ws_connect(rtm.body['url']) as ws:
                        await asyncio.wait(
                            (self.ping(ws), self.receive(ws),),
                            return_when=asyncio.FIRST_COMPLETED,
                        )

                raise BotReconnect()
            except BotReconnect:
                logger.info('BotReconnect raised. I will reconnect to rtm.')
                continue
            except:  # noqa
                logger.exception('Unexpected Exception raised')
                continue 

def process(self):
        # start websocket connection
        ws = self.ws = web.WebSocketResponse()
        await ws.prepare(self.request)

        # session was interrupted
        if self.session.interrupted:
            await self.ws.send_str(close_frame(1002, "Connection interrupted"))

        elif self.session.state == STATE_CLOSED:
            await self.ws.send_str(close_frame(3000, "Go away!"))

        else:
            try:
                await self.manager.acquire(self.session)
            except Exception:  # should use specific exception
                await self.ws.send_str(close_frame(3000, "Go away!"))
                await ws.close()
                return ws
            server = ensure_future(self.server(ws, self.session))
            client = ensure_future(self.client(ws, self.session))
            try:
                await asyncio.wait(
                    (server, client), return_when=asyncio.FIRST_COMPLETED
                )
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                await self.session._remote_close(exc)
            finally:
                await self.manager.release(self.session)
                if not server.done():
                    server.cancel()
                if not client.done():
                    client.cancel()

        return ws 

def process(self):
        # start websocket connection
        ws = self.ws = web.WebSocketResponse(autoping=False)
        await ws.prepare(self.request)

        try:
            await self.manager.acquire(self.session)
        except Exception:  # should use specific exception
            await ws.close(message="Go away!")
            return ws

        server = ensure_future(self.server(ws, self.session))
        client = ensure_future(self.client(ws, self.session))
        try:
            await asyncio.wait((server, client), return_when=asyncio.FIRST_COMPLETED)
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            await self.session._remote_close(exc)
        finally:
            await self.manager.release(self.session)
            if not server.done():
                server.cancel()
            if not client.done():
                client.cancel()

        return ws 

def _reset_committed_routine(self):
        """ Group coordinator will reset committed points to UNKNOWN_OFFSET
        if no commit is found for group. In the NoGroup mode we need to force
        it after each assignment
        """
        event_waiter = None
        try:
            while True:
                if self._subscription.subscription is None:
                    await self._subscription.wait_for_subscription()
                    continue

                assignment = self._subscription.subscription.assignment
                if assignment is None:
                    await self._subscription.wait_for_assignment()
                    continue

                commit_refresh_needed = assignment.commit_refresh_needed
                commit_refresh_needed.clear()

                for tp in assignment.requesting_committed():
                    tp_state = assignment.state_value(tp)
                    tp_state.update_committed(
                        OffsetAndMetadata(UNKNOWN_OFFSET, ""))

                event_waiter = ensure_future(
                    commit_refresh_needed.wait(), loop=self._loop)

                await asyncio.wait(
                    [assignment.unassign_future, event_waiter],
                    return_when=asyncio.FIRST_COMPLETED,
                    loop=self._loop)

        except asyncio.CancelledError:
            pass

        # Just to make sure we properly close started tasks we cancel
        # event.wait() task
        if event_waiter is not None and not event_waiter.done():
            event_waiter.cancel()
            event_waiter = None 

def _commit_refresh_routine(self, assignment):
        """ Task that will do a commit cache refresh if someone is waiting for
        it.
        """
        retry_backoff_ms = self._retry_backoff_ms / 1000
        commit_refresh_needed = assignment.commit_refresh_needed
        event_waiter = None
        try:
            while assignment.active:
                commit_refresh_needed.clear()
                success = await self._maybe_refresh_commit_offsets(
                    assignment)

                wait_futures = [assignment.unassign_future]
                if not success:
                    timeout = retry_backoff_ms
                else:
                    timeout = None
                    event_waiter = ensure_future(
                        commit_refresh_needed.wait(), loop=self._loop)
                    wait_futures.append(event_waiter)

                await asyncio.wait(
                    wait_futures,
                    timeout=timeout,
                    return_when=asyncio.FIRST_COMPLETED,
                    loop=self._loop)
        except asyncio.CancelledError:
            pass
        except Exception:
            # Reset event to continue in case user fixes the problem
            commit_refresh_needed.set()
            raise

        # Just to make sure we properly close started tasks we cancel
        # event.wait() task
        if event_waiter is not None and not event_waiter.done():
            event_waiter.cancel()
            event_waiter = None 

def seek_to_end(self, *partitions):
        """Seek to the most recent available offset for partitions.

        Arguments:
            *partitions: Optionally provide specific TopicPartitions, otherwise
                default to all assigned partitions.

        Raises:
            IllegalStateError: If any partition is not currently assigned
            TypeError: If partitions are not instances of TopicPartition

        .. versionadded:: 0.3.0

        """
        if not all([isinstance(p, TopicPartition) for p in partitions]):
            raise TypeError('partitions must be TopicPartition instances')

        if not partitions:
            partitions = self._subscription.assigned_partitions()
            assert partitions, 'No partitions are currently assigned'
        else:
            not_assigned = (
                set(partitions) - self._subscription.assigned_partitions()
            )
            if not_assigned:
                raise IllegalStateError(
                    "Partitions {} are not assigned".format(not_assigned))

        for tp in partitions:
            log.debug("Seeking to end of partition %s", tp)
        fut = self._fetcher.request_offset_reset(
            partitions, OffsetResetStrategy.LATEST)
        assignment = self._subscription.subscription.assignment
        await asyncio.wait(
            [fut, assignment.unassign_future],
            timeout=self._request_timeout_ms / 1000,
            return_when=asyncio.FIRST_COMPLETED,
            loop=self._loop
        )
        self._coordinator.check_errors()
        return fut.done() 

def next_request(self) -> bytes:
        getter = self._loop.create_task(self._msgs.get())
        await asyncio.wait(
            [getter, self._con_lost_fut],
            return_when=asyncio.FIRST_COMPLETED)

        if self._con_lost_fut.done():
            getter.cancel()
            return self._con_lost_fut.result()

        return getter.result()