def downstream(self):
        try:
            self.upstream_buffer_task = \
                    asyncio.ensure_future(self.consume_upstream_buffer())
            async for msg in self.up_conn:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    await self.down_conn.send_str(msg.data)
                    if self.downstream_cb is not None:
                        await asyncio.shield(self.downstream_cb(msg.data))
                if msg.type == aiohttp.WSMsgType.BINARY:
                    await self.down_conn.send_bytes(msg.data)
                    if self.downstream_cb is not None:
                        await asyncio.shield(self.downstream_cb(msg.data))
                elif msg.type == aiohttp.WSMsgType.CLOSED:
                    break
                elif msg.type == aiohttp.WSMsgType.ERROR:
                    break
            # here, server gracefully disconnected
        except asyncio.CancelledError:
            raise
        except Exception:
            log.exception('unexpected error')
        finally:
            await self.close_upstream() 

def _redirect_async(self, redirect, auth):
        """Redirect the client endpoint using a Link DETACH redirect
        response.

        :param redirect: The Link DETACH redirect details.
        :type redirect: ~uamqp.errors.LinkRedirect
        :param auth: Authentication credentials to the redirected endpoint.
        :type auth: ~uamqp.authentication.common.AMQPAuth
        """
        # pylint: disable=protected-access
        if not self._connection._cbs:
            _logger.info("Closing non-CBS session.")
            await asyncio.shield(self._session.destroy_async(), loop=self.loop)
        self._session = None
        self._auth = auth
        self._hostname = self._remote_address.hostname
        await self._connection.redirect_async(redirect, auth)
        await self._build_session_async() 

def send_offsets_to_transaction(self, offsets, group_id):
        self._ensure_transactional()

        if not self._txn_manager.is_in_transaction():
            raise IllegalOperation("Not in the middle of a transaction")

        if not group_id or not isinstance(group_id, str):
            raise ValueError(group_id)

        # validate `offsets` structure
        formatted_offsets = commit_structure_validate(offsets)

        log.debug(
            "Begin adding offsets %s for consumer group %s to transaction",
            formatted_offsets, group_id)
        fut = self._txn_manager.add_offsets_to_txn(formatted_offsets, group_id)
        await asyncio.shield(fut, loop=self._loop) 

def _listener(self, subscription, interval, callback):
        while True:
            try:
                # FIXME: This should use an async requester as below
                packet = callback()
                self.__handle_packet(packet, subscription)
                await asyncio.sleep(interval)
            except asyncio.CancelledError:
                ## unsubscribe from all subscriptions
                confirmations = await asyncio.gather(
                    asyncio.shield(self._unsubscribe(subscription)) 
                    for subscription in self.subscriptions)
            except Exception as ex:
                logger.error(ex)
                logger.error(packet)
                raise 

def _listener(self):
        await self._connect()
        while True:
            try:
                packet = await self.websocket.recv()
                self.__handle_packet(packet)
            except websockets.exceptions.ConnectionClosed:
                await self._connect()
            except asyncio.CancelledError:
                ## unsubscribe from all subscriptions
                confirmations = await asyncio.gather(
                    asyncio.shield(self._unsubscribe(subscription)) 
                    for subscription in self.subscriptions)
            except Exception as ex:
                logger.error(ex)
                logger.error(packet)
                raise 

def disconnect(self, timeout: Optional[float] = None) -> None:
        """Close the connection gracefully."""
        await self._connection.close(code=1000)
        try:
            await asyncio.wait_for(
                asyncio.shield(self._message_processor), timeout or self.TIMEOUT
            )
            self._message_processor.result()
        except asyncio.TimeoutError:
            pass
        finally:
            if not self._message_processor.done():
                self._message_processor.cancel()
                try:
                    await self._message_processor
                except asyncio.CancelledError:
                    pass 

def init_workers(app: web.Application, conf: WorkersConfig) -> ThreadPoolExecutor:
    n = conf.max_workers
    executor = ThreadPoolExecutor(max_workers=n)

    loop = asyncio.get_event_loop()
    run = loop.run_in_executor
    fs = [run(executor, warm, conf.path_to_model_state) for _ in range(0, n)]
    await asyncio.gather(*fs)

    async def close_executor(app: web.Application) -> None:
        fs = [run(executor, clean) for _ in range(0, n)]
        await asyncio.shield(asyncio.gather(*fs))
        executor.shutdown(wait=True)

    app.on_cleanup.append(close_executor)
    app["executor"] = executor
    return executor 

def test_shield_effect(self):
        # Cancelling outer() does not affect inner().
        proof = 0
        waiter = asyncio.Future(loop=self.loop)

        @asyncio.coroutine
        def inner():
            nonlocal proof
            yield from waiter
            proof += 1

        @asyncio.coroutine
        def outer():
            nonlocal proof
            yield from asyncio.shield(inner(), loop=self.loop)
            proof += 100

        f = asyncio.ensure_future(outer(), loop=self.loop)
        test_utils.run_briefly(self.loop)
        f.cancel()
        with self.assertRaises(asyncio.CancelledError):
            self.loop.run_until_complete(f)
        waiter.set_result(None)
        test_utils.run_briefly(self.loop)
        self.assertEqual(proof, 1) 

def test_gather_shield(self):
        child1 = asyncio.Future(loop=self.loop)
        child2 = asyncio.Future(loop=self.loop)
        inner1 = asyncio.shield(child1, loop=self.loop)
        inner2 = asyncio.shield(child2, loop=self.loop)
        parent = asyncio.gather(inner1, inner2, loop=self.loop)
        test_utils.run_briefly(self.loop)
        parent.cancel()
        # This should cancel inner1 and inner2 but bot child1 and child2.
        test_utils.run_briefly(self.loop)
        self.assertIsInstance(parent.exception(), asyncio.CancelledError)
        self.assertTrue(inner1.cancelled())
        self.assertTrue(inner2.cancelled())
        child1.set_result(1)
        child2.set_result(2)
        test_utils.run_briefly(self.loop) 

def test_shield_effect(self):
        # Cancelling outer() does not affect inner().
        proof = 0
        waiter = asyncio.Future(loop=self.loop)

        @asyncio.coroutine
        def inner():
            nonlocal proof
            yield from waiter
            proof += 1

        @asyncio.coroutine
        def outer():
            nonlocal proof
            yield from asyncio.shield(inner(), loop=self.loop)
            proof += 100

        f = asyncio.ensure_future(outer(), loop=self.loop)
        test_utils.run_briefly(self.loop)
        f.cancel()
        with self.assertRaises(asyncio.CancelledError):
            self.loop.run_until_complete(f)
        waiter.set_result(None)
        test_utils.run_briefly(self.loop)
        self.assertEqual(proof, 1) 

def test_gather_shield(self):
        child1 = asyncio.Future(loop=self.loop)
        child2 = asyncio.Future(loop=self.loop)
        inner1 = asyncio.shield(child1, loop=self.loop)
        inner2 = asyncio.shield(child2, loop=self.loop)
        parent = asyncio.gather(inner1, inner2, loop=self.loop)
        test_utils.run_briefly(self.loop)
        parent.cancel()
        # This should cancel inner1 and inner2 but bot child1 and child2.
        test_utils.run_briefly(self.loop)
        self.assertIsInstance(parent.exception(), asyncio.CancelledError)
        self.assertTrue(inner1.cancelled())
        self.assertTrue(inner2.cancelled())
        child1.set_result(1)
        child2.set_result(2)
        test_utils.run_briefly(self.loop) 

def test_maxsize_release(self):
        pool = Pool(url="ws://localhost:8182/",
                    maxsize=2,
                    username="stephen",
                    password="password",
                    future_class=Future)

        async def go():
            c1 = await pool.acquire()
            c2 = await pool.acquire()
            c3 = pool.acquire()
            self.assertIsInstance(c3, Future)
            with self.assertRaises(asyncio.TimeoutError):
                shielded_fut = asyncio.shield(c3)
                await asyncio.wait_for(shielded_fut, 0.1)
            await pool.release(c2)
            c3 = await c3
            self.assertEqual(c2, c3)
            c1.conn.close()
            c2.conn.close()
            c3.conn.close()

        self.loop.run_until_complete(go()) 

def test_maxsize_release(self):
        pool = Pool(url="ws://localhost:8182/",
                    maxsize=2,
                    username="stephen",
                    password="password",
                    future_class=Future)

        @asyncio.coroutine
        def go():
            c1 = yield from pool.acquire()
            c2 = yield from pool.acquire()
            c3 = pool.acquire()
            self.assertIsInstance(c3, Future)
            with self.assertRaises(asyncio.TimeoutError):
                shielded_fut = asyncio.shield(c3)
                yield from asyncio.wait_for(shielded_fut, 0.1)
            yield from pool.release(c2)
            c3 = yield from c3
            self.assertEqual(c2, c3)
            c1.conn.close()
            c2.conn.close()
            c3.conn.close()

        self.loop.run_until_complete(go()) 

def get_tx_events(self):

        org = 'org1.example.com'
        peer = self.client.get_peer('peer0.' + org)

        org_admin = self.client.get_user(org, 'Admin')
        channel = self.client.get_channel(self.channel_name)
        channel_event_hub = channel.newChannelEventHub(peer, org_admin)
        stream = channel_event_hub.connect(start='oldest',
                                           stop='newest', filtered=False)

        self.txs = {}
        channel_event_hub.registerTxEvent('all', onEvent=self.onTxEvent)

        try:
            await shield(stream)
        except Exception:
            pass

        channel_event_hub.disconnect()

        self.assertEqual(len(self.txs['all']), 4) 

def send_offsets_to_transaction(self, offsets, group_id):
        self._ensure_transactional()

        if not self._txn_manager.is_in_transaction():
            raise IllegalOperation("Not in the middle of a transaction")

        if not group_id or not isinstance(group_id, str):
            raise ValueError(group_id)

        # validate `offsets` structure
        formatted_offsets = commit_structure_validate(offsets)

        log.debug(
            "Begin adding offsets %s for consumer group %s to transaction",
            formatted_offsets, group_id)
        fut = self._txn_manager.add_offsets_to_txn(formatted_offsets, group_id)
        await asyncio.shield(fut, loop=self._loop) 

def _fail(self, e: Exception, clean_close: bool = True) -> None:
        if self.close_task is None:
            self.close_task = asyncio.shield(
                asyncio.ensure_future(self._close_coro(e, clean_close=clean_close))
            ) 

def disconnected(self):
        """
        Future that resolves when the connection to Telegram
        ends, either by user action or in the background.

        Note that it may resolve in either a ``ConnectionError``
        or any other unexpected error that could not be handled.
        """
        return asyncio.shield(self._disconnected, loop=self._loop)

    # Private methods 

def _scheduled_task(self, data: TaskData) -> None:
        """Await the `data.callback` coroutine after waiting for `data.wait_time` seconds."""
        try:
            log.trace(f"Waiting {data.wait_time} seconds before awaiting callback.")
            await asyncio.sleep(data.wait_time)

            # Use asyncio.shield to prevent callback from cancelling itself.
            # The parent task (_scheduled_task) will still get cancelled.
            log.trace("Done waiting; now awaiting the callback.")
            await asyncio.shield(data.callback)
        finally:
            if inspect.iscoroutine(data.callback):
                log.trace("Explicitly closing coroutine.")
                data.callback.close() 

def _scheduled_task(self, infraction: utils.Infraction) -> None:
        """
        Marks an infraction expired after the delay from time of scheduling to time of expiration.

        At the time of expiration, the infraction is marked as inactive on the website and the
        expiration task is cancelled.
        """
        expiry = dateutil.parser.isoparse(infraction["expires_at"]).replace(tzinfo=None)
        await time.wait_until(expiry)

        # Because deactivate_infraction() explicitly cancels this scheduled task, it is shielded
        # to avoid prematurely cancelling itself.
        await asyncio.shield(self.deactivate_infraction(infraction)) 

def _scheduled_task(self, task: TaskData) -> None:
        """Calls `self.unsilence` on expired silenced channel to unsilence it."""
        await asyncio.sleep(task.delay)
        log.info("Unsilencing channel after set delay.")

        # Because `self.unsilence` explicitly cancels this scheduled task, it is shielded
        # to avoid prematurely cancelling itself
        await asyncio.shield(task.ctx.invoke(self.unsilence)) 

def _do_execute(self, conn, *, return_exceptions=False):
        self._waiters = waiters = []
        with conn._buffered():
            multi = conn.execute('MULTI')
            coros = list(self._send_pipeline(conn))
            exec_ = conn.execute('EXEC')
        gather = asyncio.gather(multi, *coros,
                                return_exceptions=True)
        last_error = None
        try:
            await asyncio.shield(gather)
        except asyncio.CancelledError:
            await gather
        except Exception as err:
            last_error = err
            raise
        finally:
            if conn.closed:
                if last_error is None:
                    last_error = ConnectionClosedError()
                for fut in waiters:
                    _set_exception(fut, last_error)
                    # fut.cancel()
                for fut in self._results:
                    if not fut.done():
                        fut.set_exception(last_error)
                        # fut.cancel()
            else:
                try:
                    results = await exec_
                except RedisError as err:
                    for fut in waiters:
                        fut.set_exception(err)
                else:
                    assert len(results) == len(waiters), (
                        "Results does not match waiters", results, waiters)
                    self._resolve_waiters(results, return_exceptions)
            return (await self._gather_result(return_exceptions)) 

def _keep_alive_async(self):
        start_time = self._counter.get_current_ms()
        try:
            while self._connection and not self._shutdown:
                current_time = self._counter.get_current_ms()
                elapsed_time = (current_time - start_time)/1000
                if elapsed_time >= self._keep_alive_interval:
                    _logger.info("Keeping %r connection alive. %r",
                                 self.__class__.__name__,
                                 self._connection.container_id)
                    await asyncio.shield(self._connection.work_async(), loop=self.loop)
                    start_time = current_time
                await asyncio.sleep(1, loop=self.loop)
        except Exception as e:  # pylint: disable=broad-except
            _logger.info("Connection keep-alive for %r failed: %r.", self.__class__.__name__, e) 

def _client_run_async(self):
        """Perform a single Connection iteration."""
        await asyncio.shield(self._connection.work_async(), loop=self.loop) 

def _build_session_async(self):
        """Build self._session based on current self.connection.
        """
        # pylint: disable=protected-access
        if not self._connection._cbs and isinstance(self._auth, authentication.CBSAsyncAuthMixin):
            self._connection._cbs = await asyncio.shield(
                self._auth.create_authenticator_async(
                    self._connection,
                    debug=self._debug_trace,
                    incoming_window=self._incoming_window,
                    outgoing_window=self._outgoing_window,
                    handle_max=self._handle_max,
                    on_attach=self._on_attach,
                    loop=self.loop),
                loop=self.loop)
            self._session = self._auth._session  # pylint: disable=protected-access
        elif self._connection._cbs:
            self._session = self._auth._session  # pylint: disable=protected-access
        else:
            self._session = self.session_type(
                self._connection,
                incoming_window=self._incoming_window,
                outgoing_window=self._outgoing_window,
                handle_max=self._handle_max,
                on_attach=self._on_attach,
                loop=self.loop) 

def close_async(self):
        """Close the client asynchronously. This includes closing the Session
        and CBS authentication layer as well as the Connection.
        If the client was opened using an external Connection,
        this will be left intact.
        """
        if self.message_handler:
            await self.message_handler.destroy_async()
            self.message_handler = None
        self._shutdown = True
        if self._keep_alive_thread:
            await self._keep_alive_thread
            self._keep_alive_thread = None
        if not self._session:
            return  # already closed.
        if not self._connection._cbs:  # pylint: disable=protected-access
            _logger.info("Closing non-CBS session.")
            await asyncio.shield(self._session.destroy_async(), loop=self.loop)
        else:
            _logger.info("CBS session pending %r.", self._connection.container_id)
        self._session = None
        if not self._ext_connection:
            _logger.info("Closing exclusive connection %r.", self._connection.container_id)
            await asyncio.shield(self._connection.destroy_async(), loop=self.loop)
        else:
            _logger.info("Shared connection remaining open.")
        self._connection = None 

def _client_ready_async(self):
        """Determine whether the client is ready to start sending messages.
        To be ready, the connection must be open and authentication complete,
        The Session, Link and MessageSender must be open and in non-errored
        states.

        :rtype: bool
        :raises: ~uamqp.errors.MessageHandlerError if the MessageSender
         goes into an error state.
        """
        # pylint: disable=protected-access
        if not self.message_handler:
            self.message_handler = self.sender_type(
                self._session, self._name, self._remote_address,
                name='sender-link-{}'.format(uuid.uuid4()),
                debug=self._debug_trace,
                send_settle_mode=self._send_settle_mode,
                receive_settle_mode=self._receive_settle_mode,
                max_message_size=self._max_message_size,
                properties=self._link_properties,
                error_policy=self._error_policy,
                encoding=self._encoding,
                loop=self.loop)
            await asyncio.shield(self.message_handler.open_async(), loop=self.loop)
            return False
        if self.message_handler.get_state() == constants.MessageSenderState.Error:
            raise errors.MessageHandlerError(
                "Message Sender Client is in an error state. "
                "Please confirm credentials and access permissions."
                "\nSee debug trace for more details.")
        if self.message_handler.get_state() != constants.MessageSenderState.Open:
            return False
        return True 

def _transfer_message_async(self, message, timeout):
        sent = await asyncio.shield(
            self.message_handler.send_async(message, self._on_message_sent, timeout=timeout),
            loop=self.loop)
        if not sent:
            _logger.info("Message not sent, raising RuntimeError.")
            raise RuntimeError("Message sender failed to add message data to outgoing queue.") 

def _client_run_async(self):
        """MessageSender Link is now open - perform message send
        on all pending messages.
        Will return True if operation successful and client can remain open for
        further work.

        :rtype: bool
        """
        # pylint: disable=protected-access
        await self.message_handler.work_async()
        self._waiting_messages = 0
        async with self._pending_messages_lock:
            self._pending_messages = await self._filter_pending_async()
        if self._backoff and not self._waiting_messages:
            _logger.info("Client told to backoff - sleeping for %r seconds", self._backoff)
            await self._connection.sleep_async(self._backoff)
            self._backoff = 0
        await asyncio.shield(self._connection.work_async(), loop=self.loop)
        return True 

def _client_ready_async(self):
        """Determine whether the client is ready to start receiving messages.
        To be ready, the connection must be open and authentication complete,
        The Session, Link and MessageReceiver must be open and in non-errored
        states.

        :rtype: bool
        :raises: ~uamqp.errors.MessageHandlerError if the MessageReceiver
         goes into an error state.
        """
        # pylint: disable=protected-access
        if not self.message_handler:
            self.message_handler = self.receiver_type(
                self._session, self._remote_address, self._name,
                on_message_received=self._message_received,
                name='receiver-link-{}'.format(uuid.uuid4()),
                debug=self._debug_trace,
                receive_settle_mode=self._receive_settle_mode,
                send_settle_mode=self._send_settle_mode,
                prefetch=self._prefetch,
                max_message_size=self._max_message_size,
                properties=self._link_properties,
                error_policy=self._error_policy,
                encoding=self._encoding,
                desired_capabilities=self._desired_capabilities,
                loop=self.loop)
            await asyncio.shield(self.message_handler.open_async(), loop=self.loop)
            return False
        if self.message_handler.get_state() == constants.MessageReceiverState.Error:
            raise errors.MessageHandlerError(
                "Message Receiver Client is in an error state. "
                "Please confirm credentials and access permissions."
                "\nSee debug trace for more details.")
        if self.message_handler.get_state() != constants.MessageReceiverState.Open:
            self._last_activity_timestamp = self._counter.get_current_ms()
            return False
        return True 

def response(self):
        shield = asyncio.shield(self.fut)
        shield._log_traceback = False

        async with async_timeout.timeout(timeout=self.timeout):
            payload = await shield

        return await self.packer.unmarshal(payload)