def test_float_maxsize(self):
        _q = janus.Queue(maxsize=1.3)
        q = _q.async_q
        q.put_nowait(1)
        q.put_nowait(2)
        self.assertTrue(q.full())
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 3)

        _q.close()
        await _q.wait_closed()

        _q = janus.Queue(maxsize=1.3)
        q = _q.async_q

        async def queue_put():
            await q.put(1)
            await q.put(2)
            self.assertTrue(q.full())

        await queue_put()

        self.assertFalse(_q._sync_mutex.locked())
        _q.close()
        await _q.wait_closed() 

def _handle_commands(
            self,
            msg_stream: AsyncIterator[Tuple[ProtocolAPI, CommandAPI[Any]]]) -> None:

        async for protocol, cmd in msg_stream:
            self._last_msg_time = time.monotonic()
            # track total number of messages received for each command type.
            self._msg_counts[type(cmd)] += 1

            queue = self._protocol_queues[type(protocol)]
            try:
                # We must use `put_nowait` here to ensure that in the event
                # that a single protocol queue is full that we don't block
                # other protocol messages getting through.
                queue.put_nowait(cmd)
            except asyncio.QueueFull:
                self.logger.error(
                    (
                        "Multiplexing queue for protocol '%s' full. "
                        "discarding message: %s"
                    ),
                    protocol,
                    cmd,
                ) 

def push(self, message: MessageClass, transfer: pyuavcan.transport.TransferFrom) -> None:
        try:
            self.queue.put_nowait((message, transfer))
            self.push_count += 1
        except asyncio.QueueFull:
            self.overrun_count += 1 

def _process_frame(self, source_node_id: int, frame: typing.Optional[UDPFrame]) -> None:
        """
        The source node-ID is always valid because anonymous transfers are not defined for the UDP transport.
        The frame argument may be None to indicate that the underlying transport has received a datagram
        which is valid but does not contain a UAVCAN UDP frame inside. This is needed for error stats tracking.

        This is a part of the transport-internal API. It's a public method despite the name because Python's
        visibility handling capabilities are limited. I guess we could define a private abstract base to
        handle this but it feels like too much work. Why can't we have protected visibility in Python?
        """
        assert isinstance(source_node_id, int) and source_node_id >= 0, 'Internal protocol violation'
        if frame is None:   # Malformed frame.
            self._statistics.errors += 1
            return
        self._statistics.frames += 1

        # TODO: implement data type hash validation. https://github.com/UAVCAN/specification/issues/60

        transfer = self._get_reassembler(source_node_id).process_frame(frame, self._transfer_id_timeout)
        if transfer is not None:
            self._statistics.transfers += 1
            self._statistics.payload_bytes += sum(map(len, transfer.fragmented_payload))
            _logger.debug('%s: Received transfer: %s; current stats: %s', self, transfer, self._statistics)
            try:
                self._queue.put_nowait(transfer)
            except asyncio.QueueFull:  # pragma: no cover
                # TODO: make the queue capacity configurable
                self._statistics.drops += len(transfer.fragmented_payload) 

def _push_frame(self, can_id: _identifier.CANID, frame: _frame.TimestampedUAVCANFrame) -> None:
        """
        This is a part of the transport-internal API. It's a public method despite the name because Python's
        visibility handling capabilities are limited. I guess we could define a private abstract base to
        handle this but it feels like too much work. Why can't we have protected visibility in Python?
        """
        try:
            self._queue.put_nowait((can_id, frame))
        except asyncio.QueueFull:
            self._statistics.drops += 1
            _logger.info('Input session %s: input queue overflow; frame %s (CAN ID fields: %s) is dropped',
                         self, frame, can_id) 

def _process_frame(self, frame: SerialFrame) -> None:
        """
        This is a part of the transport-internal API. It's a public method despite the name because Python's
        visibility handling capabilities are limited. I guess we could define a private abstract base to
        handle this but it feels like too much work. Why can't we have protected visibility in Python?
        """
        assert frame.data_specifier == self._specifier.data_specifier, 'Internal protocol violation'
        self._statistics.frames += 1

        # TODO: implement data type hash validation. https://github.com/UAVCAN/specification/issues/60

        transfer: typing.Optional[pyuavcan.transport.TransferFrom]
        if frame.source_node_id is None:
            transfer = TransferReassembler.construct_anonymous_transfer(frame)
            if transfer is None:
                self._statistics.errors += 1
                _logger.debug('%s: Invalid anonymous frame: %s', self, frame)
        else:
            transfer = self._get_reassembler(frame.source_node_id).process_frame(frame, self._transfer_id_timeout)

        if transfer is not None:
            self._statistics.transfers += 1
            self._statistics.payload_bytes += sum(map(len, transfer.fragmented_payload))
            _logger.debug('%s: Received transfer: %s; current stats: %s', self, transfer, self._statistics)
            try:
                self._queue.put_nowait(transfer)
            except asyncio.QueueFull:  # pragma: no cover
                # TODO: make the queue capacity configurable
                self._statistics.drops += len(transfer.fragmented_payload) 

def put_nowait(self, msg: Any) -> bool:
            try:
                self._queue.put_nowait(msg)
                return True
            except QueueFull:
                return False 

def _async_send(self, url, data, headers, success_cb, failure_cb):
        data = url, data, headers, success_cb, failure_cb

        try:
            self._queue.put_nowait(data)
        except asyncio.QueueFull as exc:
            skipped = self._queue.get_nowait()
            self._queue.task_done()

            *_, failure_cb = skipped

            failure_cb(RuntimeError(
                'QueuedAioHttpTransport internal queue is full'))

            self._queue.put_nowait(data) 

def _close(self):
        try:
            self._queue.put_nowait(...)
        except asyncio.QueueFull as exc:
            skipped = self._queue.get_nowait()
            self._queue.task_done()

            *_, failure_cb = skipped

            failure_cb(RuntimeError(
                'QueuedAioHttpTransport internal queue was full'))

            self._queue.put_nowait(...)

        yield from asyncio.gather(
            *self._workers,
            return_exceptions=True,
            loop=self._loop
        )

        assert len(self._workers) == 0
        assert self._queue.qsize() == 1
        try:
            assert self._queue.get_nowait() is ...
        finally:
            self._queue.task_done() 

def request(req, addr, timeout=3.0):
    '''
    Send raw data with a connection pool.
    '''
    qdata = req.pack()
    bsize = struct.pack('!H', len(qdata))
    key = addr.to_str(53)
    queue = _connections.setdefault(key, asyncio.Queue(maxsize=_DEFAULT_QUEUE_SIZE))
    for _retry in range(5):
        reader = writer = None
        try:
            reader, writer = queue.get_nowait()
        except asyncio.QueueEmpty:
            pass
        if reader is None:
            try:
                reader, writer = await asyncio.wait_for(asyncio.open_connection(addr.host, addr.port), timeout)
            except asyncio.TimeoutError:
                pass
        if reader is None:
            continue
        writer.write(bsize)
        writer.write(qdata)
        try:
            await writer.drain()
            size, = struct.unpack('!H', await reader.readexactly(2))
            data = await reader.readexactly(size)
            queue.put_nowait((reader, writer))
        except asyncio.QueueFull:
            pass
        return data
    else:
        raise DNSConnectionError 

def datagram_received(self, data: bytes, address: Tuple[bytes, str]) -> None:  # type: ignore
        try:
            self.protocol_queue.put_nowait(RawData(data=data, address=address))  # type: ignore
        except asyncio.QueueFull:
            pass  # Just throw the data away, is UDP 

def _process_broadcast(self, message):
        """Process the broadcast message.

        NOTE: Depending on the value of the `strict_ordering` setting
        this method is either awaited directly or offloaded to an async
        task by the `broadcast` method (message handler).
        """
        # Assert we run in a proper thread. In particular, we can access
        # the `_subscriptions` and `_sids_by_group` without any locks.
        self._assert_thread()

        group = message["group"]

        # Do nothing if group does not exist. It is quite possible for
        # a client and a backend to concurrently unsubscribe and send
        # notification. And these events do not need to be synchronized.
        if group not in self._sids_by_group:
            return

        payload = message["payload"]

        # Put the payload to the notification queues of subscriptions
        # belonging to the subscription group. Drop the oldest payloads
        # if the `notification_queue` is full.
        for sid in self._sids_by_group[group]:
            subinf = self._subscriptions[sid]
            while True:
                try:
                    subinf.notification_queue.put_nowait(payload)
                    break
                except asyncio.QueueFull:
                    # The queue is full - issue a warning and throw away
                    # the oldest item from the queue.
                    LOG.warning(
                        "Subscription notification dropped!"
                        " Subscription operation id: %s.",
                        sid,
                    )
                    subinf.notification_queue.get_nowait() 

def test_nonblocking_put_exception(self):
        q = asyncio.Queue(maxsize=1, loop=self.loop)
        q.put_nowait(1)
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 2) 

def test_float_maxsize(self):
        q = asyncio.Queue(maxsize=1.3, loop=self.loop)
        q.put_nowait(1)
        q.put_nowait(2)
        self.assertTrue(q.full())
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 3)

        q = asyncio.Queue(maxsize=1.3, loop=self.loop)
        @asyncio.coroutine
        def queue_put():
            yield from q.put(1)
            yield from q.put(2)
            self.assertTrue(q.full())
        self.loop.run_until_complete(queue_put()) 

def _del(self):
        try:
            self.queue.put_nowait(self.StopNow)
        except asyncio.QueueFull:
            log.warning("EndgameAdvisor.gamewidget_closed: Queue.Full")
        self.egtb_task.cancel() 

def test_nonblocking_put_exception(self):
        q = asyncio.Queue(maxsize=1, loop=self.loop)
        q.put_nowait(1)
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 2) 

def test_float_maxsize(self):
        q = asyncio.Queue(maxsize=1.3, loop=self.loop)
        q.put_nowait(1)
        q.put_nowait(2)
        self.assertTrue(q.full())
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 3)

        q = asyncio.Queue(maxsize=1.3, loop=self.loop)
        @asyncio.coroutine
        def queue_put():
            yield from q.put(1)
            yield from q.put(2)
            self.assertTrue(q.full())
        self.loop.run_until_complete(queue_put()) 

def try_return_connection(self, connection):
        if self.disposed:
            return

        try:
            self._idle_connections.put_nowait(connection)
        except QueueFull:
            pass 

def test_nonblocking_put_exception(self):
        q = asyncio.Queue(maxsize=1, loop=self.loop)
        q.put_nowait(1)
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 2) 

def test_float_maxsize(self):
        q = asyncio.Queue(maxsize=1.3, loop=self.loop)
        q.put_nowait(1)
        q.put_nowait(2)
        self.assertTrue(q.full())
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 3)

        q = asyncio.Queue(maxsize=1.3, loop=self.loop)
        @asyncio.coroutine
        def queue_put():
            yield from q.put(1)
            yield from q.put(2)
            self.assertTrue(q.full())
        self.loop.run_until_complete(queue_put()) 

def send(self, message):
        """Sends a message to the websocket connection"""

        try:
            self._sender_queue.put_nowait(message)
        except asyncio.QueueFull:
            _LOGGER.error('Client sender queue size exceeded {}'.format(
                MAX_PENDING_MESSAGES))
            self.close() 

def test_nonblocking_put_exception(self):
        q = asyncio.Queue(maxsize=1, loop=self.loop)
        q.put_nowait(1)
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 2) 

def test_float_maxsize(self):
        q = asyncio.Queue(maxsize=1.3, loop=self.loop)
        q.put_nowait(1)
        q.put_nowait(2)
        self.assertTrue(q.full())
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 3)

        q = asyncio.Queue(maxsize=1.3, loop=self.loop)
        @asyncio.coroutine
        def queue_put():
            yield from q.put(1)
            yield from q.put(2)
            self.assertTrue(q.full())
        self.loop.run_until_complete(queue_put()) 

def test_nonblocking_put_exception(self):
        _q = janus.Queue(maxsize=1)
        q = _q.async_q
        q.put_nowait(1)
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 2)

        self.assertFalse(_q._sync_mutex.locked())
        _q.close()
        await _q.wait_closed() 

def test_get_nowait_queuefull(get_size):
    q = TaskQueue()
    with pytest.raises(asyncio.QueueFull):
        q.get_nowait(get_size) 

def test_internal_queue_put_nowait_full():
    queue = InternalQueue(maxsize=1)
    queue.put_nowait(True)
    with pytest.raises(QueueFull):
        queue.put_nowait(True) 

def put_nowait(self, item: T):
        """Put an item into the queue without blocking.

        If no free slot is immediately available, raise QueueFull.
        """
        with self._mutex:
            if self.full():
                raise asyncio.QueueFull
            self._put(item)
            self._unfinished_tasks += 1
            self._finished.clear()
            self._wakeup_next(self._getters) 

def test_nonblocking_put_exception(self):
        q = asyncio.Queue(maxsize=1, loop=self.loop)
        q.put_nowait(1)
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 2) 

def test_float_maxsize(self):
        q = asyncio.Queue(maxsize=1.3, loop=self.loop)
        q.put_nowait(1)
        q.put_nowait(2)
        self.assertTrue(q.full())
        self.assertRaises(asyncio.QueueFull, q.put_nowait, 3)

        q = asyncio.Queue(maxsize=1.3, loop=self.loop)

        async def queue_put():
            await q.put(1)
            await q.put(2)
            self.assertTrue(q.full())
        self.loop.run_until_complete(queue_put()) 

def send_nowait(self, item: Any) -> None:
        """
        尝试立即发送数据到channel中

        Args:
            item (any): 待发送的对象

        Raises:
            asyncio.QueueFull: 如果channel已满则会抛出 asyncio.QueueFull
        """
        if not self._closed:
            if self._last_only:
                while not self.empty():
                    asyncio.Queue.get_nowait(self)
            asyncio.Queue.put_nowait(self, item)