4115567202

Build Type

github-actions

Committed by GitHub

Commit Message

Merge 81a560da7 into 0af4f6336

Pull Request Pull Request #164: Use dataclasses instead of NamedTuple

Run Details

440 of 522 branches covered (84.29%)

Branch coverage included in aggregate %.

113 of 136 new or added lines in 3 files covered. (83.09%)

3 existing lines in 2 files now uncovered.

1176 of 1361 relevant lines covered (86.41%)

4.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.97

/aiormq/abc.py

import asyncio
import dataclasses
import io
import logging
from abc import ABC, abstractmethod, abstractproperty
from types import TracebackType
from typing import (
    Any, Awaitable, Callable, Coroutine, Dict, Iterable, Optional, Set, Tuple,
    Type, Union,
)

import pamqp
from pamqp import commands as spec
from pamqp.base import Frame
from pamqp.body import ContentBody
from pamqp.commands import Basic, Channel, Confirm, Exchange, Queue, Tx
from pamqp.common import FieldArray, FieldTable, FieldValue
from pamqp.constants import REPLY_SUCCESS
from pamqp.header import ContentHeader
from pamqp.heartbeat import Heartbeat
from yarl import URL


connection_logger = logging.getLogger("aiormq.connection")
ExceptionType = Union[BaseException, Type[BaseException]]


# noinspection PyShadowingNames
class TaskWrapper:
    __slots__ = "_exception", "task"

    _exception: Union[BaseException, Type[BaseException]]
    task: asyncio.Task

    def __init__(self, task: asyncio.Task):
        self.task = task
        self._exception = asyncio.CancelledError

    def throw(self, exception: ExceptionType) -> None:
        self._exception = exception
        self.task.cancel()

    async def __inner(self) -> Any:
        try:
            return await self.task
        except asyncio.CancelledError as e:
            raise self._exception from e

    def __await__(self, *args: Any, **kwargs: Any) -> Any:
        return self.__inner().__await__()

    def cancel(self) -> None:
        return self.throw(asyncio.CancelledError())

    def __getattr__(self, item: str) -> Any:
        return getattr(self.task, item)

    def __repr__(self) -> str:
        return "<%s: %s>" % (self.__class__.__name__, repr(self.task))


TaskType = Union[asyncio.Task, TaskWrapper]
CoroutineType = Coroutine[Any, None, Any]
GetResultType = Union[Basic.GetEmpty, Basic.GetOk]


@dataclasses.dataclass(frozen=True)
class DeliveredMessage:
    delivery: Union[spec.Basic.Deliver, spec.Basic.Return, GetResultType]
    header: ContentHeader
    body: bytes
    channel: "AbstractChannel"

    @property
    def routing_key(self) -> Optional[str]:
        if isinstance(
            self.delivery, (
                spec.Basic.Return,
                spec.Basic.GetOk,
                spec.Basic.Deliver,
            ),
        ):
            return self.delivery.routing_key
        return None

    @property
    def exchange(self) -> Optional[str]:
        if isinstance(
            self.delivery, (
                spec.Basic.Return,
                spec.Basic.GetOk,
                spec.Basic.Deliver,
            ),
        ):
            return self.delivery.exchange
        return None

    @property
    def delivery_tag(self) -> Optional[int]:
        if isinstance(
            self.delivery, (
                spec.Basic.GetOk,
                spec.Basic.Deliver,
            ),
        ):
            return self.delivery.delivery_tag
        return None

    @property
    def redelivered(self) -> Optional[bool]:
        if isinstance(
            self.delivery, (
                spec.Basic.GetOk,
                spec.Basic.Deliver,
            ),
        ):
            return self.delivery.redelivered
        return None

    @property
    def consumer_tag(self) -> Optional[str]:
        if isinstance(self.delivery, spec.Basic.Deliver):
            return self.delivery.consumer_tag
        return None

    @property
    def message_count(self) -> Optional[int]:
        if isinstance(self.delivery, spec.Basic.GetOk):
            return self.delivery.message_count
        return None


ChannelRType = Tuple[int, Channel.OpenOk]

CallbackCoro = Coroutine[Any, Any, Any]
ConsumerCallback = Callable[[DeliveredMessage], CallbackCoro]
ReturnCallback = Callable[[DeliveredMessage], Any]

ArgumentsType = FieldTable

ConfirmationFrameType = Union[
    Basic.Ack, Basic.Nack, Basic.Reject,
]


@dataclasses.dataclass(frozen=True)
class SSLCerts:
    cert: Optional[str]
    key: Optional[str]
    capath: Optional[str]
    cafile: Optional[str]
    cadata: Optional[bytes]
    verify: bool


@dataclasses.dataclass(frozen=True)
class FrameReceived:
    channel: int
    frame: str


URLorStr = Union[URL, str]
DrainResult = Awaitable[None]
TimeoutType = Optional[Union[float, int]]
FrameType = Union[Frame, ContentHeader, ContentBody]
RpcReturnType = Optional[
    Union[
        Basic.CancelOk,
        Basic.ConsumeOk,
        Basic.GetOk,
        Basic.QosOk,
        Basic.RecoverOk,
        Channel.CloseOk,
        Channel.FlowOk,
        Channel.OpenOk,
        Confirm.SelectOk,
        Exchange.BindOk,
        Exchange.DeclareOk,
        Exchange.DeleteOk,
        Exchange.UnbindOk,
        Queue.BindOk,
        Queue.DeleteOk,
        Queue.DeleteOk,
        Queue.PurgeOk,
        Queue.UnbindOk,
        Tx.CommitOk,
        Tx.RollbackOk,
        Tx.SelectOk,
    ]
]


@dataclasses.dataclass(frozen=True)
class ChannelFrame:
    payload: bytes
    should_close: bool
    drain_future: Optional[asyncio.Future] = None

    def drain(self) -> None:
        if not self.should_drain:
            return

        if self.drain_future is not None:
            self.drain_future.set_result(None)

    @property
    def should_drain(self) -> bool:
        return self.drain_future is not None and not self.drain_future.done()

    @classmethod
    def marshall(
        cls, channel_number: int,
        frames: Iterable[Union[FrameType, Heartbeat, ContentBody]],
        drain_future: Optional[asyncio.Future] = None,
    ) -> "ChannelFrame":
        should_close = False

        with io.BytesIO() as fp:
            for frame in frames:
                if should_close:
                    connection_logger.warning(
                        "It looks like you are going to send a frame %r after "
                        "the connection is closed, it's pointless, "
                        "the frame is dropped.", frame,
                    )
                    continue
                if isinstance(frame, spec.Connection.CloseOk):
                    should_close = True
                fp.write(pamqp.frame.marshal(frame, channel_number))

            return cls(
                payload=fp.getvalue(),
                drain_future=drain_future,
                should_close=should_close,
            )


class AbstractFutureStore:
    futures: Set[Union[asyncio.Future, TaskType]]
    loop: asyncio.AbstractEventLoop

    @abstractmethod
    def add(self, future: Union[asyncio.Future, TaskWrapper]) -> None:
        raise NotImplementedError

    @abstractmethod
    def reject_all(self, exception: Optional[ExceptionType]) -> Any:
        raise NotImplementedError

    @abstractmethod
    def create_task(self, coro: CoroutineType) -> TaskType:
        raise NotImplementedError

    @abstractmethod
    def create_future(self) -> asyncio.Future:
        raise NotImplementedError

    @abstractmethod
    def get_child(self) -> "AbstractFutureStore":
        raise NotImplementedError


class AbstractBase(ABC):
    loop: asyncio.AbstractEventLoop

    @abstractmethod
    def _future_store_child(self) -> AbstractFutureStore:
        raise NotImplementedError

    @abstractmethod
    def create_task(self, coro: CoroutineType) -> TaskType:
        raise NotImplementedError

    def create_future(self) -> asyncio.Future:
        raise NotImplementedError

    @abstractmethod
    async def _on_close(self, exc: Optional[Exception] = None) -> None:
        raise NotImplementedError

    @abstractmethod
    async def close(
        self, exc: Optional[ExceptionType] = asyncio.CancelledError(),
    ) -> None:
        raise NotImplementedError

    @abstractmethod
    def __str__(self) -> str:
        raise NotImplementedError

    @abstractproperty
    def is_closed(self) -> bool:
        raise NotImplementedError


class AbstractChannel(AbstractBase):
    frames: asyncio.Queue
    connection: "AbstractConnection"
    number: int
    on_return_callbacks: Set[ReturnCallback]
    closing: asyncio.Future

    @abstractmethod
    async def open(self) -> spec.Channel.OpenOk:
        pass

    @abstractmethod
    async def basic_get(
        self, queue: str = "", no_ack: bool = False,
        timeout: TimeoutType = None,
    ) -> DeliveredMessage:
        raise NotImplementedError

    @abstractmethod
    async def basic_cancel(
        self, consumer_tag: str, *, nowait: bool = False,
        timeout: TimeoutType = None,
    ) -> spec.Basic.CancelOk:
        raise NotImplementedError

    @abstractmethod
    async def basic_consume(
        self,
        queue: str,
        consumer_callback: ConsumerCallback,
        *,
        no_ack: bool = False,
        exclusive: bool = False,
        arguments: Optional[ArgumentsType] = None,
        consumer_tag: Optional[str] = None,
        timeout: TimeoutType = None,
    ) -> spec.Basic.ConsumeOk:
        raise NotImplementedError

    @abstractmethod
    def basic_ack(
        self, delivery_tag: int, multiple: bool = False, wait: bool = True,
    ) -> DrainResult:
        raise NotImplementedError

    @abstractmethod
    def basic_nack(
        self,
        delivery_tag: int,
        multiple: bool = False,
        requeue: bool = True,
        wait: bool = True,
    ) -> DrainResult:
        raise NotImplementedError

    @abstractmethod
    def basic_reject(
        self, delivery_tag: int, *, requeue: bool = True, wait: bool = True,
    ) -> DrainResult:
        raise NotImplementedError

    @abstractmethod
    async def basic_publish(
        self,
        body: bytes,
        *,
        exchange: str = "",
        routing_key: str = "",
        properties: Optional[spec.Basic.Properties] = None,
        mandatory: bool = False,
        immediate: bool = False,
        timeout: TimeoutType = None,
    ) -> Optional[ConfirmationFrameType]:
        raise NotImplementedError

    @abstractmethod
    async def basic_qos(
        self,
        *,
        prefetch_size: Optional[int] = None,
        prefetch_count: Optional[int] = None,
        global_: bool = False,
        timeout: TimeoutType = None,
    ) -> spec.Basic.QosOk:
        raise NotImplementedError

    @abstractmethod
    async def basic_recover(
        self, *, nowait: bool = False, requeue: bool = False,
        timeout: TimeoutType = None,
    ) -> spec.Basic.RecoverOk:
        raise NotImplementedError

    @abstractmethod
    async def exchange_declare(
        self,
        exchange: str = "",
        *,
        exchange_type: str = "direct",
        passive: bool = False,
        durable: bool = False,
        auto_delete: bool = False,
        internal: bool = False,
        nowait: bool = False,
        arguments: Optional[ArgumentsType] = None,
        timeout: TimeoutType = None,
    ) -> spec.Exchange.DeclareOk:
        raise NotImplementedError

    @abstractmethod
    async def exchange_delete(
        self,
        exchange: str = "",
        *,
        if_unused: bool = False,
        nowait: bool = False,
        timeout: TimeoutType = None,
    ) -> spec.Exchange.DeleteOk:
        raise NotImplementedError

    @abstractmethod
    async def exchange_bind(
        self,
        destination: str = "",
        source: str = "",
        routing_key: str = "",
        *,
        nowait: bool = False,
        arguments: Optional[ArgumentsType] = None,
        timeout: TimeoutType = None,
    ) -> spec.Exchange.BindOk:
        raise NotImplementedError

    @abstractmethod
    async def exchange_unbind(
        self,
        destination: str = "",
        source: str = "",
        routing_key: str = "",
        *,
        nowait: bool = False,
        arguments: Optional[ArgumentsType] = None,
        timeout: TimeoutType = None,
    ) -> spec.Exchange.UnbindOk:
        raise NotImplementedError

    @abstractmethod
    async def flow(
        self, active: bool,
        timeout: TimeoutType = None,
    ) -> spec.Channel.FlowOk:
        raise NotImplementedError

    @abstractmethod
    async def queue_bind(
        self,
        queue: str,
        exchange: str,
        routing_key: str = "",
        nowait: bool = False,
        arguments: Optional[ArgumentsType] = None,
        timeout: TimeoutType = None,
    ) -> spec.Queue.BindOk:
        raise NotImplementedError

    @abstractmethod
    async def queue_declare(
        self,
        queue: str = "",
        *,
        passive: bool = False,
        durable: bool = False,
        exclusive: bool = False,
        auto_delete: bool = False,
        nowait: bool = False,
        arguments: Optional[ArgumentsType] = None,
        timeout: TimeoutType = None,
    ) -> spec.Queue.DeclareOk:
        raise NotImplementedError

    @abstractmethod
    async def queue_delete(
        self,
        queue: str = "",
        if_unused: bool = False,
        if_empty: bool = False,
        nowait: bool = False,
        timeout: TimeoutType = None,
    ) -> spec.Queue.DeleteOk:
        raise NotImplementedError

    @abstractmethod
    async def queue_purge(
        self, queue: str = "", nowait: bool = False,
        timeout: TimeoutType = None,
    ) -> spec.Queue.PurgeOk:
        raise NotImplementedError

    @abstractmethod
    async def queue_unbind(
        self,
        queue: str = "",
        exchange: str = "",
        routing_key: str = "",
        arguments: Optional[ArgumentsType] = None,
        timeout: TimeoutType = None,
    ) -> spec.Queue.UnbindOk:
        raise NotImplementedError

    @abstractmethod
    async def tx_commit(
        self, timeout: TimeoutType = None,
    ) -> spec.Tx.CommitOk:
        raise NotImplementedError

    @abstractmethod
    async def tx_rollback(
        self, timeout: TimeoutType = None,
    ) -> spec.Tx.RollbackOk:
        raise NotImplementedError

    @abstractmethod
    async def tx_select(self, timeout: TimeoutType = None) -> spec.Tx.SelectOk:
        raise NotImplementedError

    @abstractmethod
    async def confirm_delivery(
        self, nowait: bool = False,
        timeout: TimeoutType = None,
    ) -> spec.Confirm.SelectOk:
        raise NotImplementedError


class AbstractConnection(AbstractBase):
    FRAME_BUFFER_SIZE: int = 10
    # Interval between sending heartbeats based on the heartbeat(timeout)
    HEARTBEAT_INTERVAL_MULTIPLIER: TimeoutType

    # Allow three missed heartbeats (based on heartbeat(timeout)
    HEARTBEAT_GRACE_MULTIPLIER: int

    server_properties: ArgumentsType
    connection_tune: spec.Connection.Tune
    channels: Dict[int, Optional[AbstractChannel]]
    write_queue: asyncio.Queue
    url: URL
    closing: asyncio.Future

    @abstractmethod
    def set_close_reason(
        self, reply_code: int = REPLY_SUCCESS,
        reply_text: str = "normally closed",
        class_id: int = 0, method_id: int = 0,
    ) -> None:
        raise NotImplementedError

    @abstractproperty
    def is_opened(self) -> bool:
        raise NotImplementedError

    @abstractmethod
    def __str__(self) -> str:
        raise NotImplementedError

    @abstractmethod
    async def connect(
        self, client_properties: Optional[FieldTable] = None,
    ) -> bool:
        raise NotImplementedError

    @abstractproperty
    def server_capabilities(self) -> ArgumentsType:
        raise NotImplementedError

    @abstractproperty
    def basic_nack(self) -> bool:
        raise NotImplementedError

    @abstractproperty
    def consumer_cancel_notify(self) -> bool:
        raise NotImplementedError

    @abstractproperty
    def exchange_exchange_bindings(self) -> bool:
        raise NotImplementedError

    @abstractproperty
    def publisher_confirms(self) -> Optional[bool]:
        raise NotImplementedError

    async def channel(
        self,
        channel_number: Optional[int] = None,
        publisher_confirms: bool = True,
        frame_buffer_size: int = FRAME_BUFFER_SIZE,
        timeout: TimeoutType = None,
        **kwargs: Any,
    ) -> AbstractChannel:
        raise NotImplementedError

    @abstractmethod
    async def __aenter__(self) -> "AbstractConnection":
        raise NotImplementedError

    @abstractmethod
    async def __aexit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> Optional[bool]:
        raise NotImplementedError

    @abstractmethod
    async def ready(self) -> None:
        raise NotImplementedError

    @abstractmethod
    async def update_secret(
        self, new_secret: str, *,
        reason: str = "", timeout: TimeoutType = None,
    ) -> spec.Connection.UpdateSecretOk:
        raise NotImplementedError


__all__ = (
    "AbstractBase", "AbstractChannel", "AbstractConnection",
    "AbstractFutureStore", "ArgumentsType", "CallbackCoro", "ChannelFrame",
    "ChannelRType", "ConfirmationFrameType", "ConsumerCallback",
    "CoroutineType", "DeliveredMessage", "DrainResult", "ExceptionType",
    "FieldArray", "FieldTable", "FieldValue", "FrameReceived", "FrameType",
    "GetResultType", "ReturnCallback", "RpcReturnType", "SSLCerts",
    "TaskType", "TaskWrapper", "TimeoutType", "URLorStr",
)

1	import asyncio	5✔
2	import dataclasses	5✔
3	import io	5✔
4	import logging	5✔
5	from abc import ABC, abstractmethod, abstractproperty	5✔
6	from types import TracebackType	5✔
7	from typing import (	5✔
8	Any, Awaitable, Callable, Coroutine, Dict, Iterable, Optional, Set, Tuple,
9	Type, Union,
10	)
11
12	import pamqp	5✔
13	from pamqp import commands as spec	5✔
14	from pamqp.base import Frame	5✔
15	from pamqp.body import ContentBody	5✔
16	from pamqp.commands import Basic, Channel, Confirm, Exchange, Queue, Tx	5✔
17	from pamqp.common import FieldArray, FieldTable, FieldValue	5✔
18	from pamqp.constants import REPLY_SUCCESS	5✔
19	from pamqp.header import ContentHeader	5✔
20	from pamqp.heartbeat import Heartbeat	5✔
21	from yarl import URL	5✔
22
23
24	connection_logger = logging.getLogger("aiormq.connection")	5✔
25	ExceptionType = Union[BaseException, Type[BaseException]]	5✔
26
27
28	# noinspection PyShadowingNames
29	class TaskWrapper:	5✔
30	__slots__ = "_exception", "task"	5✔
31
32	_exception: Union[BaseException, Type[BaseException]]	5✔
33	task: asyncio.Task	5✔
34
35	def __init__(self, task: asyncio.Task):	5✔
36	self.task = task	5✔
37	self._exception = asyncio.CancelledError	5✔
38
39	def throw(self, exception: ExceptionType) -> None:	5✔
40	self._exception = exception	5✔
41	self.task.cancel()	5✔
42
43	async def __inner(self) -> Any:	5✔
44	try:	5✔
45	return await self.task	5✔
46	except asyncio.CancelledError as e:	5✔
47	raise self._exception from e	5✔
48
49	def __await__(self, args: Any, *kwargs: Any) -> Any:	5✔
50	return self.__inner().__await__()	5✔
51
52	def cancel(self) -> None:	5✔
53	return self.throw(asyncio.CancelledError())	5✔
54
55	def __getattr__(self, item: str) -> Any:	5✔
56	return getattr(self.task, item)	5✔
57
58	def __repr__(self) -> str:	5✔
59	return "<%s: %s>" % (self.__class__.__name__, repr(self.task))	×
60
61
62	TaskType = Union[asyncio.Task, TaskWrapper]	5✔
63	CoroutineType = Coroutine[Any, None, Any]	5✔
64	GetResultType = Union[Basic.GetEmpty, Basic.GetOk]	5✔
65
66
67	@dataclasses.dataclass(frozen=True)	5✔
68	class DeliveredMessage:	4✔
69	delivery: Union[spec.Basic.Deliver, spec.Basic.Return, GetResultType]	5✔
70	header: ContentHeader	5✔
71	body: bytes	5✔
72	channel: "AbstractChannel"	5✔
73
74	@property	5✔
75	def routing_key(self) -> Optional[str]:	5✔
76	if isinstance(	×
77	self.delivery, (
78	spec.Basic.Return,
79	spec.Basic.GetOk,
80	spec.Basic.Deliver,
81	),
82	):
83	return self.delivery.routing_key	×
84	return None	×
85
86	@property	5✔
87	def exchange(self) -> Optional[str]:	5✔
88	if isinstance(	5!
89	self.delivery, (
90	spec.Basic.Return,
91	spec.Basic.GetOk,
92	spec.Basic.Deliver,
93	),
94	):
95	return self.delivery.exchange	5✔
96	return None	×
97
98	@property	5✔
99	def delivery_tag(self) -> Optional[int]:	5✔
100	if isinstance(	5!
101	self.delivery, (
102	spec.Basic.GetOk,
103	spec.Basic.Deliver,
104	),
105	):
106	return self.delivery.delivery_tag	5✔
107	return None	×
108
109	@property	5✔
110	def redelivered(self) -> Optional[bool]:	5✔
111	if isinstance(	5!
112	self.delivery, (
113	spec.Basic.GetOk,
114	spec.Basic.Deliver,
115	),
116	):
117	return self.delivery.redelivered	5✔
118	return None	×
119
120	@property	5✔
121	def consumer_tag(self) -> Optional[str]:	5✔
122	if isinstance(self.delivery, spec.Basic.Deliver):	×
123	return self.delivery.consumer_tag	×
124	return None	×
125
126	@property	5✔
127	def message_count(self) -> Optional[int]:	5✔
128	if isinstance(self.delivery, spec.Basic.GetOk):	×
129	return self.delivery.message_count	×
130	return None	×
131
132
133	ChannelRType = Tuple[int, Channel.OpenOk]	5✔
134
135	CallbackCoro = Coroutine[Any, Any, Any]	5✔
136	ConsumerCallback = Callable[[DeliveredMessage], CallbackCoro]	5✔
137	ReturnCallback = Callable[[DeliveredMessage], Any]	5✔
138
139	ArgumentsType = FieldTable	5✔
140
141	ConfirmationFrameType = Union[	5✔
142	Basic.Ack, Basic.Nack, Basic.Reject,
143	]
144
145
146	@dataclasses.dataclass(frozen=True)	5✔
147	class SSLCerts:	4✔
148	cert: Optional[str]	5✔
149	key: Optional[str]	5✔
150	capath: Optional[str]	5✔
151	cafile: Optional[str]	5✔
152	cadata: Optional[bytes]	5✔
153	verify: bool	5✔
154
155
156	@dataclasses.dataclass(frozen=True)	5✔
157	class FrameReceived:	4✔
158	channel: int	5✔
159	frame: str	5✔
160
161
162	URLorStr = Union[URL, str]	5✔
163	DrainResult = Awaitable[None]	5✔
164	TimeoutType = Optional[Union[float, int]]	5✔
165	FrameType = Union[Frame, ContentHeader, ContentBody]	5✔
166	RpcReturnType = Optional[	5✔
167	Union[
168	Basic.CancelOk,
169	Basic.ConsumeOk,
170	Basic.GetOk,
171	Basic.QosOk,
172	Basic.RecoverOk,
173	Channel.CloseOk,
174	Channel.FlowOk,
175	Channel.OpenOk,
176	Confirm.SelectOk,
177	Exchange.BindOk,
178	Exchange.DeclareOk,
179	Exchange.DeleteOk,
180	Exchange.UnbindOk,
181	Queue.BindOk,
182	Queue.DeleteOk,
183	Queue.DeleteOk,
184	Queue.PurgeOk,
185	Queue.UnbindOk,
186	Tx.CommitOk,
187	Tx.RollbackOk,
188	Tx.SelectOk,
189	]
190	]
191
192
193	@dataclasses.dataclass(frozen=True)	5✔
194	class ChannelFrame:	4✔
195	payload: bytes	5✔
196	should_close: bool	5✔
197	drain_future: Optional[asyncio.Future] = None	5✔
198
199	def drain(self) -> None:	5✔
200	if not self.should_drain:	5✔
201	return	5✔
202
203	if self.drain_future is not None:	5!
204	self.drain_future.set_result(None)	5✔
205
206	@property	5✔
207	def should_drain(self) -> bool:	5✔
208	return self.drain_future is not None and not self.drain_future.done()	5✔
209
210	@classmethod	5✔
211	def marshall(	5✔
212	cls, channel_number: int,
213	frames: Iterable[Union[FrameType, Heartbeat, ContentBody]],
214	drain_future: Optional[asyncio.Future] = None,
215	) -> "ChannelFrame":
216	should_close = False	5✔
217
218	with io.BytesIO() as fp:	5✔
219	for frame in frames:	5✔
220	if should_close:	5!
NEW 221	connection_logger.warning(	×
222	"It looks like you are going to send a frame %r after "
223	"the connection is closed, it's pointless, "
224	"the frame is dropped.", frame,
225	)
NEW 226	continue	×
227	if isinstance(frame, spec.Connection.CloseOk):	5✔
228	should_close = True	5✔
229	fp.write(pamqp.frame.marshal(frame, channel_number))	5✔
230
231	return cls(	5✔
232	payload=fp.getvalue(),
233	drain_future=drain_future,
234	should_close=should_close,
235	)
236
237
238	class AbstractFutureStore:	5✔
239	futures: Set[Union[asyncio.Future, TaskType]]	5✔
240	loop: asyncio.AbstractEventLoop	5✔
241
242	@abstractmethod	5✔
243	def add(self, future: Union[asyncio.Future, TaskWrapper]) -> None:	5✔
244	raise NotImplementedError	×
245
246	@abstractmethod	5✔
247	def reject_all(self, exception: Optional[ExceptionType]) -> Any:	5✔
248	raise NotImplementedError	×
249
250	@abstractmethod	5✔
251	def create_task(self, coro: CoroutineType) -> TaskType:	5✔
252	raise NotImplementedError	×
253
254	@abstractmethod	5✔
255	def create_future(self) -> asyncio.Future:	5✔
256	raise NotImplementedError	×
257
258	@abstractmethod	5✔
259	def get_child(self) -> "AbstractFutureStore":	5✔
260	raise NotImplementedError	×
261
262
263	class AbstractBase(ABC):	5✔
264	loop: asyncio.AbstractEventLoop	5✔
265
266	@abstractmethod	5✔
267	def _future_store_child(self) -> AbstractFutureStore:	5✔
268	raise NotImplementedError	×
269
270	@abstractmethod	5✔
271	def create_task(self, coro: CoroutineType) -> TaskType:	5✔
272	raise NotImplementedError	×
273
274	def create_future(self) -> asyncio.Future:	5✔
275	raise NotImplementedError	×
276
277	@abstractmethod	5✔
278	async def _on_close(self, exc: Optional[Exception] = None) -> None:	5✔
279	raise NotImplementedError	×
280
281	@abstractmethod	5✔
282	async def close(	5✔
283	self, exc: Optional[ExceptionType] = asyncio.CancelledError(),
284	) -> None:
285	raise NotImplementedError	×
286
287	@abstractmethod	5✔
288	def __str__(self) -> str:	5✔
289	raise NotImplementedError	×
290
291	@abstractproperty	5✔
292	def is_closed(self) -> bool:	5✔
293	raise NotImplementedError	×
294
295
296	class AbstractChannel(AbstractBase):	5✔
297	frames: asyncio.Queue	5✔
298	connection: "AbstractConnection"	5✔
299	number: int	5✔
300	on_return_callbacks: Set[ReturnCallback]	5✔
301	closing: asyncio.Future	5✔
302
303	@abstractmethod	5✔
304	async def open(self) -> spec.Channel.OpenOk:	5✔
305	pass	×
306
307	@abstractmethod	5✔
308	async def basic_get(	5✔
309	self, queue: str = "", no_ack: bool = False,
310	timeout: TimeoutType = None,
311	) -> DeliveredMessage:
312	raise NotImplementedError	×
313
314	@abstractmethod	5✔
315	async def basic_cancel(	5✔
316	self, consumer_tag: str, *, nowait: bool = False,
317	timeout: TimeoutType = None,
318	) -> spec.Basic.CancelOk:
319	raise NotImplementedError	×
320
321	@abstractmethod	5✔
322	async def basic_consume(	5✔
323	self,
324	queue: str,
325	consumer_callback: ConsumerCallback,
326	*,
327	no_ack: bool = False,
328	exclusive: bool = False,
329	arguments: Optional[ArgumentsType] = None,
330	consumer_tag: Optional[str] = None,
331	timeout: TimeoutType = None,
332	) -> spec.Basic.ConsumeOk:
333	raise NotImplementedError	×
334
335	@abstractmethod	5✔
336	def basic_ack(	5✔
337	self, delivery_tag: int, multiple: bool = False, wait: bool = True,
338	) -> DrainResult:
339	raise NotImplementedError	×
340
341	@abstractmethod	5✔
342	def basic_nack(	5✔
343	self,
344	delivery_tag: int,
345	multiple: bool = False,
346	requeue: bool = True,
347	wait: bool = True,
348	) -> DrainResult:
349	raise NotImplementedError	×
350
351	@abstractmethod	5✔
352	def basic_reject(	5✔
353	self, delivery_tag: int, *, requeue: bool = True, wait: bool = True,
354	) -> DrainResult:
355	raise NotImplementedError	×
356
357	@abstractmethod	5✔
358	async def basic_publish(	5✔
359	self,
360	body: bytes,
361	*,
362	exchange: str = "",
363	routing_key: str = "",
364	properties: Optional[spec.Basic.Properties] = None,
365	mandatory: bool = False,
366	immediate: bool = False,
367	timeout: TimeoutType = None,
368	) -> Optional[ConfirmationFrameType]:
369	raise NotImplementedError	×
370
371	@abstractmethod	5✔
372	async def basic_qos(	5✔
373	self,
374	*,
375	prefetch_size: Optional[int] = None,
376	prefetch_count: Optional[int] = None,
377	global_: bool = False,
378	timeout: TimeoutType = None,
379	) -> spec.Basic.QosOk:
380	raise NotImplementedError	×
381
382	@abstractmethod	5✔
383	async def basic_recover(	5✔
384	self, *, nowait: bool = False, requeue: bool = False,
385	timeout: TimeoutType = None,
386	) -> spec.Basic.RecoverOk:
387	raise NotImplementedError	×
388
389	@abstractmethod	5✔
390	async def exchange_declare(	5✔
391	self,
392	exchange: str = "",
393	*,
394	exchange_type: str = "direct",
395	passive: bool = False,
396	durable: bool = False,
397	auto_delete: bool = False,
398	internal: bool = False,
399	nowait: bool = False,
400	arguments: Optional[ArgumentsType] = None,
401	timeout: TimeoutType = None,
402	) -> spec.Exchange.DeclareOk:
403	raise NotImplementedError	×
404
405	@abstractmethod	5✔
406	async def exchange_delete(	5✔
407	self,
408	exchange: str = "",
409	*,
410	if_unused: bool = False,
411	nowait: bool = False,
412	timeout: TimeoutType = None,
413	) -> spec.Exchange.DeleteOk:
414	raise NotImplementedError	×
415
416	@abstractmethod	5✔
417	async def exchange_bind(	5✔
418	self,
419	destination: str = "",
420	source: str = "",
421	routing_key: str = "",
422	*,
423	nowait: bool = False,
424	arguments: Optional[ArgumentsType] = None,
425	timeout: TimeoutType = None,
426	) -> spec.Exchange.BindOk:
427	raise NotImplementedError	×
428
429	@abstractmethod	5✔
430	async def exchange_unbind(	5✔
431	self,
432	destination: str = "",
433	source: str = "",
434	routing_key: str = "",
435	*,
436	nowait: bool = False,
437	arguments: Optional[ArgumentsType] = None,
438	timeout: TimeoutType = None,
439	) -> spec.Exchange.UnbindOk:
440	raise NotImplementedError	×
441
442	@abstractmethod	5✔
443	async def flow(	5✔
444	self, active: bool,
445	timeout: TimeoutType = None,
446	) -> spec.Channel.FlowOk:
447	raise NotImplementedError	×
448
449	@abstractmethod	5✔
450	async def queue_bind(	5✔
451	self,
452	queue: str,
453	exchange: str,
454	routing_key: str = "",
455	nowait: bool = False,
456	arguments: Optional[ArgumentsType] = None,
457	timeout: TimeoutType = None,
458	) -> spec.Queue.BindOk:
459	raise NotImplementedError	×
460
461	@abstractmethod	5✔
462	async def queue_declare(	5✔
463	self,
464	queue: str = "",
465	*,
466	passive: bool = False,
467	durable: bool = False,
468	exclusive: bool = False,
469	auto_delete: bool = False,
470	nowait: bool = False,
471	arguments: Optional[ArgumentsType] = None,
472	timeout: TimeoutType = None,
473	) -> spec.Queue.DeclareOk:
474	raise NotImplementedError	×
475
476	@abstractmethod	5✔
477	async def queue_delete(	5✔
478	self,
479	queue: str = "",
480	if_unused: bool = False,
481	if_empty: bool = False,
482	nowait: bool = False,
483	timeout: TimeoutType = None,
484	) -> spec.Queue.DeleteOk:
485	raise NotImplementedError	×
486
487	@abstractmethod	5✔
488	async def queue_purge(	5✔
489	self, queue: str = "", nowait: bool = False,
490	timeout: TimeoutType = None,
491	) -> spec.Queue.PurgeOk:
492	raise NotImplementedError	×
493
494	@abstractmethod	5✔
495	async def queue_unbind(	5✔
496	self,
497	queue: str = "",
498	exchange: str = "",
499	routing_key: str = "",
500	arguments: Optional[ArgumentsType] = None,
501	timeout: TimeoutType = None,
502	) -> spec.Queue.UnbindOk:
503	raise NotImplementedError	×
504
505	@abstractmethod	5✔
506	async def tx_commit(	5✔
507	self, timeout: TimeoutType = None,
508	) -> spec.Tx.CommitOk:
509	raise NotImplementedError	×
510
511	@abstractmethod	5✔
512	async def tx_rollback(	5✔
513	self, timeout: TimeoutType = None,
514	) -> spec.Tx.RollbackOk:
515	raise NotImplementedError	×
516
517	@abstractmethod	5✔
518	async def tx_select(self, timeout: TimeoutType = None) -> spec.Tx.SelectOk:	5✔
519	raise NotImplementedError	×
520
521	@abstractmethod	5✔
522	async def confirm_delivery(	5✔
523	self, nowait: bool = False,
524	timeout: TimeoutType = None,
525	) -> spec.Confirm.SelectOk:
526	raise NotImplementedError	×
527
528
529	class AbstractConnection(AbstractBase):	5✔
530	FRAME_BUFFER_SIZE: int = 10	5✔
531	# Interval between sending heartbeats based on the heartbeat(timeout)
532	HEARTBEAT_INTERVAL_MULTIPLIER: TimeoutType	5✔
533
534	# Allow three missed heartbeats (based on heartbeat(timeout)
535	HEARTBEAT_GRACE_MULTIPLIER: int	5✔
536
537	server_properties: ArgumentsType	5✔
538	connection_tune: spec.Connection.Tune	5✔
539	channels: Dict[int, Optional[AbstractChannel]]	5✔
540	write_queue: asyncio.Queue	5✔
541	url: URL	5✔
542	closing: asyncio.Future	5✔
543
544	@abstractmethod	5✔
545	def set_close_reason(	5✔
546	self, reply_code: int = REPLY_SUCCESS,
547	reply_text: str = "normally closed",
548	class_id: int = 0, method_id: int = 0,
549	) -> None:
550	raise NotImplementedError	×
551
552	@abstractproperty	5✔
553	def is_opened(self) -> bool:	5✔
554	raise NotImplementedError	×
555
556	@abstractmethod	5✔
557	def __str__(self) -> str:	5✔
558	raise NotImplementedError	×
559
560	@abstractmethod	5✔
561	async def connect(	5✔
562	self, client_properties: Optional[FieldTable] = None,
563	) -> bool:
564	raise NotImplementedError	×
565
566	@abstractproperty	5✔
567	def server_capabilities(self) -> ArgumentsType:	5✔
568	raise NotImplementedError	×
569
570	@abstractproperty	5✔
571	def basic_nack(self) -> bool:	5✔
572	raise NotImplementedError	×
573
574	@abstractproperty	5✔
575	def consumer_cancel_notify(self) -> bool:	5✔
576	raise NotImplementedError	×
577
578	@abstractproperty	5✔
579	def exchange_exchange_bindings(self) -> bool:	5✔
580	raise NotImplementedError	×
581
582	@abstractproperty	5✔
583	def publisher_confirms(self) -> Optional[bool]:	5✔
584	raise NotImplementedError	×
585
586	async def channel(	5✔
587	self,
588	channel_number: Optional[int] = None,
589	publisher_confirms: bool = True,
590	frame_buffer_size: int = FRAME_BUFFER_SIZE,
591	timeout: TimeoutType = None,
592	**kwargs: Any,
593	) -> AbstractChannel:
594	raise NotImplementedError	×
595
596	@abstractmethod	5✔
597	async def __aenter__(self) -> "AbstractConnection":	5✔
598	raise NotImplementedError	×
599
600	@abstractmethod	5✔
601	async def __aexit__(	5✔
602	self,
603	exc_type: Optional[Type[BaseException]],
604	exc_val: Optional[BaseException],
605	exc_tb: Optional[TracebackType],
606	) -> Optional[bool]:
607	raise NotImplementedError	×
608
609	@abstractmethod	5✔
610	async def ready(self) -> None:	5✔
611	raise NotImplementedError	×
612
613	@abstractmethod	5✔
614	async def update_secret(	5✔
615	self, new_secret: str, *,
616	reason: str = "", timeout: TimeoutType = None,
617	) -> spec.Connection.UpdateSecretOk:
618	raise NotImplementedError	×
619
620
621	__all__ = (	5✔
622	"AbstractBase", "AbstractChannel", "AbstractConnection",
623	"AbstractFutureStore", "ArgumentsType", "CallbackCoro", "ChannelFrame",
624	"ChannelRType", "ConfirmationFrameType", "ConsumerCallback",
625	"CoroutineType", "DeliveredMessage", "DrainResult", "ExceptionType",
626	"FieldArray", "FieldTable", "FieldValue", "FrameReceived", "FrameType",
627	"GetResultType", "ReturnCallback", "RpcReturnType", "SSLCerts",
628	"TaskType", "TaskWrapper", "TimeoutType", "URLorStr",
629	)

mosquito / aiormq / 4115567202

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous