ԆEWz d dl mZ d dlZd dlmZmZmZmZmZm Z d dl
m Z d dl m
Z
d dlmZ erd dlmZ dZd Z G d
d Z G d d
ZdS ) ) annotationsN)
TYPE_CHECKING
AsyncIterator AwaitableCallableListOptional)uuid4)errors)Msg)JetStreamContexti i c e Zd ZdZddddddeefd'dZed(d Zed(d Z ed)d Z
ed*d Z ed*d Z ed*d Z
d+d,d Zd Zd! Zd-d"Zd.d/d$Zd-d%Zd-d&ZdS )0 Subscriptiona
A Subscription represents interest in a particular subject.
A Subscription should not be constructed directly, rather
`connection.subscribe()` should be used to get a subscription.
::
nc = await nats.connect()
# Async Subscription
async def cb(msg):
print('Received', msg)
await nc.subscribe('foo', cb=cb)
# Sync Subscription
sub = nc.subscribe('foo')
msg = await sub.next_msg()
print('Received', msg)
r Nidintsubjectstrqueuecb*Optional[Callable[[Msg], Awaitable[None]]]futureOptional[asyncio.Future]max_msgspending_msgs_limitpending_bytes_limitreturnNonec
. || _ || _ || _ || _ || _ d| _ || _ || _ d| _ || _ | | _
t j | | _
|i | _ nd | _ d| _ d | _ d | _ d | _ d S )Nr F)maxsize)_conn_id_subject_queue _max_msgs _received_cb_future_closed_pending_msgs_limit_pending_bytes_limitasyncioQueue_pending_queue_pending_next_msgs_calls
_pending_size_wait_for_msgs_task_message_iterator_jsi)
selfconnr r r r r r r r s
u/builddir/build/BUILD/imunify360-venv-2.6.2/opt/imunify360/venv/lib/python3.11/site-packages/nats/aio/subscription.py__init__zSubscription.__init__? s
! $6 $7!29-HZ2[2[2[ :,.D ) ),0D ) #' !% 6: c | j S )z<
Returns the subject of the `Subscription`.
)r# r4 s r6 r zSubscription.subjectf s
} r8 c | j S )zX
Returns the queue name of the `Subscription` if part of a queue group.
)r$ r: s r6 r zSubscription.queuem s
{r8 AsyncIterator[Msg]c F | j st j d | j S )a
Retrieves an async iterator for the messages from the subscription.
This is only available if a callback isn't provided when creating a
subscription.
::
nc = await nats.connect()
sub = await nc.subscribe('foo')
# Use `async for` which implicitly awaits messages
async for msg in sub.messages:
print('Received', msg)
zCcannot iterate over messages with a non iteration subscription type)r2 r Errorr: s r6 messageszSubscription.messagest s* % f, dee e%%r8 c 4 | j S )zw
Number of delivered messages by the NATS Server that are being buffered
in the pending queue.
)r. qsizer: s r6 pending_msgszSubscription.pending_msgs s "((***r8 c | j S )zk
Size of data sent by the NATS Server that is being buffered
in the pending queue.
)r0 r: s r6
pending_byteszSubscription.pending_bytes s
!!r8 c | j S )zK
Number of delivered messages to this subscription so far.
)r& r: s r6 deliveredzSubscription.delivered s
~ r8 ?timeoutOptional[float]r c K d fd } j j r t j j rt j d t
t } t j | }| j
|< | d{V } xj t |j
z c_ j | j
|d S # t j $ r% j j r t j t j t j $ r j j r t j w xY w# j
|d w xY w)a
:params timeout: Time in seconds to wait for next message before timing out.
:raises nats.errors.TimeoutError:
next_msg can be used to retrieve the next message from a stream of messages using
await syntax, this only works when not passing a callback on `subscribe`::
sub = await nc.subscribe('hello')
msg = await sub.next_msg(timeout=1)
r r c l K t j j d {V S N)r, wait_forr. get)r4 rH s r6 timed_getz(Subscription.next_msg..timed_get s; )$*=*A*A*C*CWMMMMMMMM Mr8 z4nats: next_msg cannot be used in async subscriptionsNr r )r! is_closedr ConnectionClosedErrorr' r> r r
r, create_taskr/ r0 lendatar. task_donepop TimeoutErrorCancelledError)r4 rH rO task_namer msgs `` r6 next_msgzSubscription.next_msg s N N N N N N N : /. . 8 W, UVV VLL ? (55F7=D )) 4 ,,,,,,C
#ch-- /
) ) + + + ) - -i > > > >! # & & &z#
3 22% %% z#
3 22
) - -i > > > >s .C! get_running_looprS _wait_for_msgsr1 r( _SubscriptionMessageIteratorr2 )r4 error_cbs r6 _startzSubscription._start s 8 H.tx88
Q&))
Q.5.I$(-.X.X
Q l#OPPP'.'?'A'A'M'MdNaNabjNkNk'l'lD $ $ $
\ H D%A$%G%GD " " "r8 c K | j j r t j | j j r t j | j r t j | d{V dS )zU
Removes interest in a subject, but will process remaining messages.
N) r! rQ r rR is_drainingConnectionDrainingErrorr) BadSubscriptionError_drainr: s r6 drainzSubscription.drain sm : /. . : ! 10 0 < .- -kkmmr8 c K | j | j d {V | j d {V | j r | j d {V | | j | j n# t j $ r w xY w d| _
d S # d| _
w xY wNT) r! _send_unsubscriber" flushr. join_stop_processing _remove_subr, rY r) r: s r6 rj zSubscription._drain s
*..tx88 8 8 8 8 8 8 8 *""$$ $ $ $ $ $ $ $"
1 )..000000000
! ! # # #
J " "48 , , , ,%
- DLLL4DL s B B" !C "B33C C limitc K | j j r t j | j j r t j | j r t j || _ |dk s$| j |k rS| j
r:d| _ | | j
| j | j j s)| j | j | d{V dS dS )aX
:param limit: Max number of messages to receive before unsubscribing.
Removes interest in a subject, remaining messages will be discarded.
If `limit` is greater than zero, interest is not immediately removed,
rather, interest will be automatically removed after `limit` messages
are received.
r T)rs N)r! rQ r rR rg rh r) ri r% r& r. emptyrq rr r" is_reconnectingrn )r4 rs s r6 unsubscribezSubscription.unsubscribe s : /. . : ! 10 0 < .- - A::$.E11d6I6O6O6Q6Q1 DL ! ! # # # J " "48 , , ,z) F*..txu.EE E E E E E E E E E F Fr8 c | j r2| j s| j | j r| j dS dS )zF
Stops the subscription from processing new messages.
N)r1 donecancelr2 _cancelr: s r6 rq z Subscription._stop_processing sj # .D,D,I,I,K,K . $ + + - - - ! - " * * , , , , , - -r8 c K | j s
J d | j d{V }| xj t |j z c_ | | d{V nT# t j $ r Y | j dS t $ r }|r || d{V Y d}~nd}~ww xY w| j n # | j w xY w| j
dk r0| j | j
k r | j j r|
n# t j $ r Y dS w xY w?)zz
A coroutine to read and process messages if a callback is provided.
Should be called as a task.
z-_wait_for_msgs can be called only from _startTNr )r' r. rN r0 rT rU r, rY rV Exceptionr% r& ru rq )r4 rd r[ es r6 rb zSubscription._wait_for_msgs' s xHHHHHx
/ 3 3 5 5555555""c#(mm3""4 ((3--'''''''' - '1133333 ! * * * * &hqkk))))))) * '113333D'113333 >A%%$.DN*J*JtObOh*J))+++)
1 s` AD< A4 3C" 4CC" D< C(C ;C" CC" D< "C==>D< <EE)r r r r r r r r r r r r r r r r r r )r r )r r< )r r )rG )rH rI r r r r )r )rs r )__name__
__module__ __qualname____doc__ DEFAULT_SUB_PENDING_MSGS_LIMIT DEFAULT_SUB_PENDING_BYTES_LIMITr7 propertyr r r? rB rD rF r\ re rk rj rw rq rb r8 r6 r r ( s 2 9=+/"@#B%: %: %: %: %:N X X & & & X&( + + + X+ " " " X" X +? +? +? +? +?ZH H H$
2F F F F F4- - - - r8 r c . e Zd Zd dZd dZd
dZdd Zd
S )rc subr r r c \ || _ |j | _ t j | _ d S rL )_subr. r$ r, Future_unsubscribed_future)r4 r s r6 r7 z%_SubscriptionMessageIterator.__init__K s) "% *-*< :A.:J:J!!!r8 c p | j s | j d d S d S rm )r ry
set_resultr: s r6 r{ z$_SubscriptionMessageIterator._cancelP s@ (--// 7 % 0 0 6 6 6 6 6 7 7r8 c | S rL r r: s r6 __aiter__z&_SubscriptionMessageIterator.__aiter__T s r8 r c l K t j | j }|| j g}t j |t j d {V \ }}| j }||v r| j |
}| j xj t |j
z c_ |j dk r$|j |j k r| |S | j r| t& )N) return_whenr )r, ra rS r$ rN r waitFIRST_COMPLETEDr rV resultr0 rT rU r% r& r{ ry rz StopAsyncIteration)r4 get_tasktasksfinished_r r[ s r6 __anext__z&_SubscriptionMessageIterator.__anext__W s +--99$+//:K:KLL'/1J&K #LGr s5 # " " " " " )((((((!+ "3 _ _ _ _ _ _ _ _D ! ! ! ! ! ! ! ! ! !r8