L # d dl mZ d dlZd dlZd dlZd dlmZ d dlmZ d dl m
Z
m Z m Z m
Z
mZmZmZ d dlZd dlZd dlmZ d dlmZ d dlmZ d d lmZmZm Z m Z m Z d d
l m Z d d l!m"Z" d d l#m$Z$m%Z%m&Z&m'Z'm(Z( e
rd d
lm)Z) dZ* e+d Z, e-e, Z.dZ/ e-e/ Z0dZ1dZ2e
dge d f Z3dZ4dZ5dZ6 G d de" Z7dS ) ) annotationsN) BytesParser) token_hex)
TYPE_CHECKINGAny AwaitableCallableDictListOptional)Msg) Subscription)api)BadBucketErrorBucketNotFoundErrorFetchTimeoutErrorInvalidBucketNameError
NotFoundError)KeyValue)JetStreamManager)OBJ_ALL_CHUNKS_PRE_TEMPLATEOBJ_ALL_META_PRE_TEMPLATEOBJ_STREAM_TEMPLATEVALID_BUCKET_RE ObjectStore)NATS503s NATS/1.0s
z KV_{bucket}z
$KV.{bucket}.r
i i @ c e Zd ZdZej dddfdbdZedcd ZdddZ dedZ
dfdgd"Z dfdhd&Z did'Z
ddd(Zdddddd)d)dd)eedddfdjd=Zdd)d)eefdkd@ZedldC ZdddeedfdmdGZdddeeddfdndIZedodK ZedpdM ZedqdN ZedqdO ZedrdQ Z G dR dS Z G dT d>> ''--- " 6qqq 9""4(((h !4!;!;!=!=$BZ [[[[[[[[[[[r= msgr
c K |j t | j j dz dz d }| j | }|sd S | rd S |j r]|j t j j
t k r+| t j j j d S t# j |j }d|v rFt j j j |d }| | d S t j | }| | d S # t2 j t2 j f$ r Y d S w xY w)N error) subjectlenr/ rE r3 getdoneheadersr HeaderSTATUSNO_RESPONDERS_STATUS
set_exceptionnatsjserrorsNoStreamResponseErrorjsonloadsdataAPIError
from_errorPubAck
from_response
set_resultr4 CancelledErrorInvalidStateError)r: rN tokenfutureresperracks r; rK z$JetStreamContext._handle_async_reply sc C 6 7 7" .handle_done
sj ' + +ELLNND A A A4.//144377999 1 9 9 ; ; ; ; ;r= )replyrW ) r2 rM r rX rw r$ r* rx r4 wait_forr9 acquire TimeoutErrorrh r\ r] r^ TooManyStalledMsgsErrorr/ rG rH rF r encodeFutureadd_done_callbackr3 rJ r6 is_setclearr| ) r: rS ro r} rr rW rt r{ inboxrk r rj s ` @r;
publish_asyncz JetStreamContext.publish_async s( ( ' +((** * * * * * * *'''' )C.4C
* + )C&)#g,,&7&7C
" # 9"4#H#P#P#R#R\fggg g g g g g g g g$g&<= 9 9 9'.8 8 9
##%%
Yq\\((**+++(+
U!( !1!1 < < < < < < ---6<#ELLNN3 . 5 5 7 7 8 / 5 5 7 7 7h wu||~~s SSSSSSSSS
s 3B4 42C&c * t | j S )z@
returns the number of pending async publishes.
)rT r3 r? s r; publish_async_pendingz&JetStreamContext.publish_async_pending s 4.///r= c H K | j d{V dS )zH
waits for all pending async publishes to be completed.
N)r6 waitr? s r; publish_async_completedz(JetStreamContext.publish_async_completed% s5 16688888888888r= FqueuerD Optional[Callback]durableconfig Optional[api.ConsumerConfig]
manual_ackboolordered_consumeridle_heartbeat flow_controlpending_msgs_limitpending_bytes_limitdeliver_policyOptional[api.DeliverPolicy] headers_onlyOptional[bool]inactive_thresholdPushSubscriptionc
K | | j | d{V }d}d}|r5|r1||k r+t j j d| d| d |}d}| }|rF | j || d{V }|}n!# t j j j $ r d}Y nw xY w||j }|j j }|sS|r$t j j d |j
r$t j j d ni|s't j j d| ||k r*t j j d | d
| n|r |t j }|j
s||_
|j s||_ |j s||_ |
r|
|_ |r||_ |j | j }||_ |j s||_ |
|_ | r| |_ n |j pd } |r@d|_ t j j |_ d |_ d
|_ | |_ d |_ d|_ | j || d{V }|j }|tC d |tC d | " ||||||| | d{V S )a Create consumer if needed and push-subscribe to it.
1. Check if consumer exists.
2. Creates consumer if needed.
3. Calls `subscribe_bind`.
:param subject: Subject from a stream from JetStream.
:param queue: Deliver group name from a set a of queue subscribers.
:param durable: Name of the durable consumer to which the the subscription should be bound.
:param stream: Name of the stream to which the subscription should be bound. If not set,
then the client will automatically look it up based on the subject.
:param manual_ack: Disables auto acking for async subscriptions.
:param ordered_consumer: Enable ordered consumer mode.
:param idle_heartbeat: Enable Heartbeats for a consumer to detect failures.
:param flow_control: Enable Flow Control for a consumer.
::
import asyncio
import nats
async def main():
nc = await nats.connect()
js = nc.jetstream()
await js.add_stream(name='hello', subjects=['hello'])
await js.publish('hello', b'Hello JS!')
async def cb(msg):
print('Received:', msg)
# Ephemeral Async Subscribe
await js.subscribe('hello', cb=cb)
# Durable Async Subscribe
# NOTE: Only one subscription can be bound to a durable name. It also auto acks by default.
await js.subscribe('hello', cb=cb, durable='foo')
# Durable Sync Subscribe
# NOTE: Sync subscribers do not auto ack.
await js.subscribe('hello', durable='bar')
# Queue Async Subscribe
# NOTE: Here 'workers' becomes deliver_group, durable name and queue name.
await js.subscribe('hello', 'workers', cb=cb)
if __name__ == '__main__':
asyncio.run(main())
Nz"cannot create queue subscription 'z' to consumer ''TzIcannot create a queue subscription for a consumer without a deliver groupz+consumer is already bound to a subscriptionzAcannot create a subscription for a consumer with a deliver group z#cannot create a queue subscription z% for a consumer with a deliver group r! i`5 r zcannot detect consumerz0config is required for existing durable consumer)rD rr r r r consumerr r )#r@ find_stream_name_by_subjectr\ r] r^ Error
consumer_infor r
deliver_group
push_boundr ConsumerConfig durable_namer r r deliver_subjectr/ new_inboxfilter_subjectsfilter_subjectr r AckPolicyNONE
ack_policy max_deliverack_wait num_replicas mem_storage add_consumername TypeErrorsubscribe_bind)r: rS r rD r rr r r r r r r r r r r deliverr r
should_creater s r; rI zJetStreamContext.subscribe+ s H > 9@@IIIIIIIIF
7e++gn**+pPU+p+pfm+p+p+pqqq
#
%
%&*i&=&=fg&N&N N N N N N N
"7>/
%
%
% $
% $")F * 0 >M
^ '...c #- ^ '.../\]]]^
'...k\ikk m++ '...@e @ @0=@ @ ,
, *~ +--&
.&-#'
-',$&
3&2#
7(6%!
?,>) %- (,,..)0& )
0(/% #/F
<(6%%!'!6!;!
*&*#$'M$6!%&""+(6%&'#%)""&)"8"8"8"O"O O O O O O OM$)H 455 5 > NOO O((! - 1 3 )
s (#B B*)B*api.ConsumerConfigr c p K |r/|s-|j t j j ur| | }|j t
d | j |j |j pd||| d{V } t | | || }
t | | j |||
| | | _
|j r5t j | j
| j
_ |r5t j | j
| j
_ |
S )z'Push-subscribe to an existing consumer.Nz"config.deliver_subject is required )rS r rD r r )r] r" rr orderedpsubsubccreq)r r r r _auto_ack_callbackr r r/ rI r r r _JSI_jsir r4 create_taskactivity_check_hbtaskcheck_flow_control_response_fctask) r: rr r r rD r r r r r r s r; r z JetStreamContext.subscribe_bind sZ " -z -(9 AS(S(S((,,B ! ) @AA AH&&*&," 1 3 '
00sFHMM $(($ )
N & 2383J3J3L3L M MCH [ & 2383W3W3Y3Y Z ZCH r= callbackCallbackc d fd }|S )NrN r
r+ r, c K | d {V | d {V d S # t j j $ r Y d S w xY wN)rn r\ r^ MsgAlreadyAckdError)rN r s r; new_callbackz9JetStreamContext._auto_ack_callback..new_callback sp (3--
ggii;2
s 2 A
A
rN r
r+ r, )r r s ` r; r z#JetStreamContext._auto_ack_callback s) r= inbox_prefixOptional[bytes]!JetStreamContext.PullSubscriptionc > K | | j | d{V }d} |r#| j || d{V d}n # t j j j $ r Y nw xY w|} |r|t j }|j s||_
|r||_ ||_ n7| j
j } | |_ | j || d{V | |||||| d{V S )a* Create consumer and pull subscription.
1. Find stream name by subject if `stream` is not passed.
2. Create consumer with the given `config` if not created.
3. Call `pull_subscribe_bind`.
::
import asyncio
import nats
async def main():
nc = await nats.connect()
js = nc.jetstream()
await js.add_stream(name='mystream', subjects=['foo'])
await js.publish('foo', b'Hello World!')
sub = await js.pull_subscribe('foo', stream='mystream')
msgs = await sub.fetch()
msg = msgs[0]
await msg.ack()
await nc.close()
if __name__ == '__main__':
asyncio.run(main())
NTFr )r rr r r r r )r@ r r r\ r] r^ r r r r r r r r/ rG rH rJ r pull_subscribe_bind)
r: rS r rr r r r r r
consumer_names
r; pull_subscribez JetStreamContext.pull_subscribe s P > 9@@IIIIIIIIF
&i--fg>>>>>>>>> %
w~+ D
@~ +-- )
0(/%
, % &-## $ 3 3 5 5 < < > >
+ )(((?? ? ? ? ? ? ? ?--% 3 1
.
s %A A*)A*r c z K |st d |$t | j j dd dz }|| j j z }| j | || d{V } d}
|r|}
n|r|}
n|}
t | | ||
| S )a
pull_subscribe returns a `PullSubscription` that can be delivered messages
from a JetStream pull based consumer by calling `sub.fetch`.
::
import asyncio
import nats
async def main():
nc = await nats.connect()
js = nc.jetstream()
await js.add_stream(name='mystream', subjects=['foo'])
await js.publish('foo', b'Hello World!')
msgs = await sub.fetch()
msg = msgs[0]
await msg.ack()
await nc.close()
if __name__ == '__main__':
asyncio.run(main())
z nats: stream name is requiredNrB )r r )r] r rr r r )
ValueErrorrp r/ rE rG rH rI rJ r PullSubscription) r: r rr r r r r r r r r s r; r z$JetStreamContext.pull_subscribe_bindk s H > <== = !7!:;;dBL !4!4!6!66H&& NN 1 3 '
% #MM
% MM $M 00" 1
r=
Optional[Msg]c j ||j d S |j t j j S r )rW rU r rX rY )clsrN s r;
is_status_msgz JetStreamContext.is_status_msg s, ;#+-4{sz 0111r= statusc |sdS t | rdS t j j j | NTF)r _is_temporary_errorr\ r] r^ rc from_msg)r r rN s r; _is_processable_msgz$JetStreamContext._is_processable_msg sE 4 / / 7 7 5gn%..s333r= c |t j j k s*|t j j k s|t j j k rdS dS r )r
StatusCode NO_MESSAGESCONFLICTREQUEST_TIMEOUTr r s r; r z$JetStreamContext._is_temporary_error s>
cn0 0 0 000 77745r= c 4 |t j j k rdS dS r )r r CONTROL_MESSAGEr s r;
_is_heartbeatz JetStreamContext._is_heartbeat s S^3 3 345r=
start_timec < |d S |t j |z
z
S r )time monotonic)r r' r s r; _time_untilz JetStreamContext._time_until s$ ?4$.**Z788r= c P e Zd Zd!dZd"dZd"dZd Zd Zd Zd#dZ d$d Z
d%d Z d S )&JetStreamContext._JSIr] r r" r rr r$ r r r !JetStreamContext.PushSubscriptionr r r r r+ r, c || _ || _ || _ || _ || _ || _ || _ d | _ d | _ |r|j r |j | _ d| _
d| _ d | _ d | _
d| _ d| _ d| _ d | _ d S Nr r T)_conn_js_stream_ordered_psub_sub_ccreqr _hbir _dseq_sseq_cmeta_fcr_fcd_fciseq_activer )r: r] r" rr r r r r s r; r< z JetStreamContext._JSI.__init__ s DJDH!DL #DM DJDI DK DL DI
1-
1 ! 0 DJDJ)-DK'+DIDI DL+/DL DLLLr= r c 4 | xj dz
c_ || _ d S )Nr )r r r: r s r; track_sequencesz%JetStreamContext._JSI.track_sequences s LLA LL DKKKr= c : d| _ || _ | j | _ d S )NT)r r r r r s r; schedule_flow_control_responsez4JetStreamContext._JSI.schedule_flow_control_response s DL DI DIIIr= c ~ | j j r | j j S | j | j j z
S r )r
_cb deliveredr _pending_queueqsizer? s r; get_js_deliveredz&JetStreamContext._JSI.get_js_delivered s7 y}
+y**<$)":"@"@"B"BB Br= c K d} | j j rd S t j | j |z d {V | j }d| _ |s*| j r#| | j dz d {V n# t j $ r Y d S w xY w)NrQ TFr )
r is_closedr4 sleepr r r reset_ordered_consumerr rh )r:
hbc_thresholdactives r; r z$JetStreamContext._JSI.activity_check s M
z+
"- M(ABBBBBBBBB !\F#(DL! N = N"&"="=dj1n"M"M M M M M M M M - EE
s A2 A A2 2BBc z K | j j rd S | j | j j z
| j k rI| j } |r | j | d {V n# t $ r Y nw xY wd | _ d| _ t j d d {V n# t j $ r Y d S w xY w)NTr g ?)
r r r r r r r r r| Exceptionr4 r! rh )r: fc_replys r; r z1JetStreamContext._JSI.check_flow_control_response s
z+ tz'@'F'F'H'HHTYVV#'9! ' C&*j&8&8&B&B B B B B B B B ( ! ! ! D!$( $% !----------- - EE
s: B& 6B&
"A- ,B& -
A:7B& 9A::+B& &B98B9rN r
c , K d| _ | j sd S | | j }t |d }d }|j r:|j t j j }|rt | }d }||k rt |d }| j r$|
| j dz d {V }nGt j
j ||| }| j | d {V |S )NT r! r )stream_resume_sequenceconsumer_sequencelast_consumer_sequence)r r _get_metadata_fieldsr* rW rU r rX
LAST_CONSUMERr r" r r\ r] r^ ConsumerSequenceMismatchErrorr _error_cb) r: rN tokensdseqldseq ldseq_str did_resetsseqecss r; check_for_sequence_mismatchz1JetStreamContext._JSI.check_for_sequence_mismatch, s& DL;
t--dk::Fvay>>DE{
+ KOOCJ,D E E + NNE I}}6!9~~= 4&*&A&A$*q.&Q&Q Q Q Q Q Q QII '.FF/3*./4 G C
*..s333333333 r= r6
Optional[int]r c K | j j }| j | | j }| j xj dz
c_ | j j }| j | j j |< || j _ || j _ | j | d {V || j _ | j
| j d {V t j d d {V d | _
d| _ | j }||_ t" j j |_ ||_ || _ t j | dS r )r
_idr _remove_subr _sid_subsr _send_unsubscribe_subject_send_subscriber4 r! r r
r r r
DeliverPolicyBY_START_SEQUENCEr
opt_start_seqr recreate_consumer)r: r6 osid new_delivernsidr s r; r" z,JetStreamContext._JSI.reset_ordered_consumerH s` 9=D J " "4 ( ( ( *..00K
JOOq OO:?D%)YDJ T " DIM !DJN *..t44 4 4 4 4 4 4 4 "-DI *,,TY77 7 7 7 7 7 7 7 -"" " " " " " " " DKDJ [F%0F "$'$5$GF !#'F DK
6 6 8 8 9 9 94r= c K | j j | j | j | j j d {V }|j | j _ d S # t $ r+}| j
| d {V Y d }~d S d }~ww xY w)N)r r' ) r r@ r r r r0 r r _consumerr& r r0 )r: cinform s r; rE z'JetStreamContext._JSI.recreate_consumers s
0 "hm 8 8 dkcgckct 8 u uuuuuuu',z
$$$
0
0
0j**3///////////////
0s A
A
B BBN)r] r r" r rr r$ r r r r r r r r r+ r, )r r$ r+ r, )rN r
r+ r )r6 r9 r+ r r+ r, ) __name__
__module__ __qualname__r< r r r r r r8 r" rE r r= r; r r s D % % % %
C C C
( & 8) ) ) ) V 0 0 0 0 0 0r= r c e Zd ZdZdd Zd d
Zed d Zej d Zed Z e j d Z d d dZ
d d! fd
Z xZ S )"r zP
PushSubscription is a subscription that is delivered messages.
r] r r r rr r$ r r+ r, c || _ || _ || _ || _ |j | _ |j | _ |j | _ |j | _ |j | _ |j | _ |j
| _
|j | _ |j | _ |j
| _
|j | _ |j | _ |j | _ |j | _ |j | _ |j | _ d S r )r r rJ r
r r; r@ _queue _max_msgs _receivedr _future_closed_pending_msgs_limit_pending_bytes_limitr
_pending_size_wait_for_msgs_task_message_iterator_pending_next_msgs_calls)r: r] r rr r s r; r< z*JetStreamContext.PushSubscription.__init__ s DH!DL %DNDI DJwDH LDM *DK ]DN ]DNwDH ;DL ;DL (+'>D $(+(@D %"%"4D !$!2D '*'>D $%(%:D ",/,HD ) ) )r= api.ConsumerInfoc j K | j j | j | j d{V }|S ze
consumer_info gets the current info of the consumer from this subscription.
Nr r@ r r rJ r: infos r; r z/JetStreamContext.PushSubscription.consumer_info sP 44 D Kr= r* c | j j S zS
Number of delivered messages to this subscription so far.
r
rT r? s r; r z+JetStreamContext.PushSubscription.delivered
9& &r= c || j _ d S r re r: values r; r z+JetStreamContext.PushSubscription.delivered s "'DI r= c | j j S r r
rY r? s r; rY z/JetStreamContext.PushSubscription._pending_size s
9* *r= c || j _ d S r rk rh s r; rY z/JetStreamContext.PushSubscription._pending_size s &+DI # # #r= ?r' rq r
c | K | j | d{V }| j r| j j rd| j j _ | j j | j j j k rD| j j j }|r1| j | d{V d| j j _ |S )aJ
: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`::
NT) r
next_msgr r r r r r r| )r: r' rN r' s r; ro z*JetStreamContext.PushSubscription.next_msg s **733333333C y
3TY^
3)- &9>2244 8NNN #y~ 2H 3 "j 0 0 : ::::::::.2 +Jr= r limitc . K t | d{V | j j j r#| j j j | j j j r%| j j j dS dS )z
Unsubscribes from a subscription, canceling any heartbeat and flow control tasks,
and optionally limits the number of messages to process before unsubscribing.
N)super unsubscriber
r r cancelr )r: rp __class__s r; rs z-JetStreamContext.PushSubscription.unsubscribe s
''%%e,, , , , , , , ,y~%
0 &--///y~%
0 &--/////
0
0r= )
r] r r r rr r$ r r$ r+ r, r+ r] r+ r* )rm )r' rq r+ r
)r )rp r* )
rM rN rO __doc__r< r propertyr setterrY ro rs
__classcell__)ru s @r; r z!JetStreamContext.PushSubscriptionz s
I I I I>
' ' '
'
( (
(
+ +
+
, ,
, ( 0 0 0 0 0 0 0 0 0 0 0r= c e Zd ZdZd#d
Zed$d Zed$d Zed$d Zd%dZ d&dZ
d'd(d Z d)d*d!Z d)d+d"Z
dS ),r zM
PullSubscription is a subscription that can fetch messages.
r] r r r rr r$ r r rp r+ r, c || _ |j | _ || _ || _ || _ | j j }| d| d| | _ | | _ d S )Nz.CONSUMER.MSG.NEXT..) r r/ r
r rJ r. _nmsrJ _deliver)r: r] r rr r r r# s r; r< z*JetStreamContext.PullSubscription.__init__ sg DHvDH DI!DL %DNX%F!IIfIIxIIDI #NN , ,DMMMr= r* c > | j j S )z
Number of delivered messages by the NATS Server that are being buffered
in the pending queue.
)r
r r r? s r; pending_msgsz.JetStreamContext.PullSubscription.pending_msgs s 9+1133 3r= c | j j S )zw
Size of data sent by the NATS Server that is being buffered
in the pending queue.
rk r? s r;
pending_bytesz/JetStreamContext.PullSubscription.pending_bytes s 9* *r= c | j j S rd re r? s r; r z+JetStreamContext.PullSubscription.delivered rf r= c t K | j t d | j d{V dS )z
unsubscribe destroys the inboxes of the pull subscription making it
unable to continue to receive messages.
Nnats: invalid subscription)r
r rs r? s r; rs z-JetStreamContext.PullSubscription.unsubscribe sK
y !=>>>)'')) ) ) ) ) ) ) ) ) )r= r] c j K | j j | j | j d{V }|S r_ r` ra s r; r z/JetStreamContext.PullSubscription.consumer_info s< 44T\4>RRRRRRRRDKr= r r! Nbatchr' rq heartbeat List[Msg]c H K | j t d |dk rt d ||dk rt d |rt |dz dz
nd}|dk r | ||| d{V }|gS | |||| d{V }|S ) a
fetch makes a request to JetStream to be delivered a set of messages.
:param batch: Number of messages to fetch from server.
:param timeout: Max duration of the fetch request before it expires.
:param heartbeat: Idle Heartbeat interval in seconds for the fetch request.
::
import asyncio
import nats
async def main():
nc = await nats.connect()
js = nc.jetstream()
await js.add_stream(name='mystream', subjects=['foo'])
await js.publish('foo', b'Hello World!')
msgs = await sub.fetch(5)
for msg in msgs:
await msg.ack()
await nc.close()
if __name__ == '__main__':
asyncio.run(main())
Nr r znats: invalid batch sizer znats: invalid fetch timeout ʚ;i )r
r r*
_fetch_one_fetch_n)r: r r' r expiresrN msgss r; fetchz'JetStreamContext.PullSubscription.fetch s D y !=>>> qyy !;<<<"w!|| !>???@GQc'M122W< Y nw xY w)
Nr r r r r FTr r ) r
r empty
get_nowaitrY rT rb r r r& r* r/ r| r r` dumpsr r r r r ro r r r\ r^ r r] rc r r4 r ) r: r r' r r rN r next_reqr got_any_responsedeadlines r; r z,JetStreamContext.PullSubscription._fetch_oneO s I,E kkmm
**,,CI++s38}}<++ - ; ;C @ @F ! ! J D kkmm
H !HW
3&)'ll#
L-0]1J-K-K)*(""
8$$++--
))J $
/ ; ;GZ P PH $ 2 28 2 D DDDDDDDC . ; ;C @ @F #+99&AA %/3 , $ ,??GG H"&+": : #''."9"B"B3"G"G G "
+ / ; ;GZ P PH +1
, 4"3 3 +
s2 AA> <A> >
B
B ;A4H 0AH t | d k r)tD j# j$ j% & |
Z t7 | D ]}t || }|
|d k r|c S | j |
d {V }
t
|
} t | rd}t | |
r| dz } | |
n# t* j $ r Y nw xY wt | d k r |rtB |S ) NFr r r r r Tno_waitr r )'r
r r r r r rY rT rb r r appendr& r* r/ r| r r` r r r r4 r! ro r r r ranger r r r r r r\ r] r^ rc r )r: r r r' r r r r r neededrN r r ir _s r; r z*JetStreamContext.PullSubscription._fetch_n s/ DI,E ))J $ F
kkmm
**,,CI++s38}}<++ - ; ;C @ @F ! !aKFKK$$$$ D kkmm
H &HW
.&-#
L-0]1J-K-K)*"&HY (""
8$$++--
-"" " " " " " " "
I . .w 7 7777777'
KKK
% %33C88F --f55
$( !55fcBB
C ! "1f - - - -#3#?#?#T#T$(I$6$6x$6$H$H H H H H H H!1!?!?!D!D!S^%???6S^MkCkCk "E - ; ;F C C -/3 , $ - A A&# N N - "aKF KK , , , + D 4yy1}} H &HW
.&-#
L-0]1J-K-K)*(""
8$$++--
-"" " " " " " " " C&
DQ;; K+77LLt99>> $(I$6$6x$6$H$H H H H H H H"/ + 4"3 3
$(I$6$6x$6$H$H H H H H H H"/ D - ; ;C @ @F'55f== !+/( !
D ! C((( 3>#==== Ta "gn 5 > >s C CCM&
DR
v ) )A / ; ;GZ P PH +1 # $ 2 28 2 D DDDDDDDC - ; ;C @ @F'55f== !+/( ';;FCHH ) ! C((( ) '
4yyA~~"2~''Kst AB0 B0 0
B=<B=) F
F! F!C'K= =LL0!P P,0!Q Q$#Q$7W BW W-,W-) r] r r r rr r$ r r$ r rp r+ r, rw rL rv )r r! N)r r* r' rq r rq r+ r r )r r9 r' rq r rq r+ r
)
r r* r r9 r' rq r rq r+ r )rM rN rO rx r< ry r r r rs r r r r r r= r; r z!JetStreamContext.PullSubscription s( - - - -(
4 4 4
4
+ + +
+
' ' '
' * * * * '()- 0 0 0 0 0 l *. E E E E E X *. o o o o o o o r= r bucketr c K t j | t t | } | | d {V }n# t $ r t w xY w|j j dk rt t ||t | | t |j j S )Nr r r rr prer] direct)r matchr KV_STREAM_TEMPLATEformat stream_infor r r max_msgs_per_subjectr r KV_PRE_TEMPLATEr allow_direct)r: r rr sis r; key_valuezJetStreamContext.key_valueM s ( ( 0( (#**&*99 &''////////BB & & &% % &
9 )A - - &&f&55 .//
s A A( Optional[api.KeyValueConfig]c R K |t j |d } |j d i |}t j |j t d}|j r|j |k r|j }|j dk rt j
j j t j
d i dt |j d|j dd |j d
gd |j d d
dd
dd
dt j j d|d|j d|j ddd|j ddd|j d|j d|j d|j }| | d{V }|j J t7 |j |j t8 |j | t; |j j S ) z]
create_key_value takes an api.KeyValueConfig and creates a KV in JetStream.
Nr r x r r descriptionsubjectsz$KV.z.>r allow_rollup_hdrsT
allow_msg_ttl deny_deletediscardduplicate_windowmax_age max_bytes
max_consumers max_msg_sizemax_msgsr r storage republishr r ) r KeyValueConfigevolver r r r ttlhistoryr\ r] r^ KeyHistoryTooLargeError StreamConfigr r r r
DiscardPolicyNEWr max_value_sizereplicasr r
add_streamr r r r r r )r: r paramsr rr r s r; create_key_valuez!JetStreamContext.create_key_valuea sC >'vh/?@@@F (((( / / 7( ("( : *&*'777 %z >B '.8 8!
#**&-*@@@
**
/V]...//
#d
$
%))
. -
JJ
&&
"
..
R
"(
!
" NN#
$ &&%
( ??6**
*
*
*
*
*
*{&&& ; &&fm&<< .//
r= c K t j | t t | }| | d{V S )zr
delete_key_value deletes a JetStream KeyValue store by destroying
the associated stream.
Nr )r r r r r
delete_streamr: r rr s r; delete_key_valuez!JetStreamContext.delete_key_value s[
( ( 0( (#**&*99''/////////r= r c K t j | t t j | } | | d {V n# t $ r t w xY wt ||| S )Nr r rr r] ) r r r r r r r r r r s r; object_storez JetStreamContext.object_store s ( ( 0( ($+6::: &""6** * * * * * * * * & & &% % &
s A A" Optional[api.ObjectStoreConfig]c b K |t j | }n||_ |j di |}t j |j t |j }t j | }t j | }|j
}|dk rd}t j t j |j |j
||g|j |d|j |j |j t j j dd }| | d{V |j J t- |j |j | S ) zd
create_object_store takes an api.ObjectStoreConfig and creates a OBJ in JetStream.
Nr r r T) r r r r r r r r placementr r r r r )r ObjectStoreConfigr r r r r r r r r r r r r r r r r r r r r ) r: r r r r chunksmetar rr s r; create_object_storez$JetStreamContext.create_object_store sU >*&999FF "FM (((( / / 7( (},34@@@(/t<<<$ >>I!$+6=AAA * d^J N &%) "
oof%%%%%%%%%{&&& ;
r= c K t j | t t j | }| | d{V S )z]
delete_object_store will delete the underlying stream for the named object.
Nr )r r r r r r r s r; delete_object_storez$JetStreamContext.delete_object_store sX ( ( 0( ($+6:::''/////////r= ) r" r r# r$ r% r& r' r( r) r* r+ r, )r+ r rL r )r= NNNN)rS r$ ro rp r' rq rr r& rW rs rt rq r+ ru )rS r$ ro rp r} rq rr r& rW r~ rt rq r+ r rw ) rS r$ r r& rD r r r& rr r& r r r r r r r rq r r r r* r r* r r r r r rq r+ r )rr r$ r r r r$ rD r r r r r r r* r r* r+ r )r r r+ r )rS r$ r r& rr r& r r r r* r r* r r r+ r )r r& rr r& r r r r* r r* r r& r r& r+ r )rN r r+ r& )r r& rN r
r+ r )r r& r+ r )r' rq r r( r+ rq )r r$ r+ r r )r r r+ r )r r$ r+ r )r r$ r+ r )NN)r r$ r r r+ r )'rM rN rO rx r DEFAULT_PREFIXr< ry r@ rM rK r| r r r !DEFAULT_JS_SUB_PENDING_MSGS_LIMIT"DEFAULT_JS_SUB_PENDING_BYTES_LIMITrI r staticmethodr r r classmethodr r r r r r r r r r r r r r r r r= r; r r K s @ ( $)-
] ] ] ] ].
X
\ \ \ \ > #' $,0#'+. +. +. +. +.` &* $"&#'? ? ? ? ?B0 0 0 0 9 9 9 9 $!%!% $/3 !&*. ""C#E6:'+.2!s
s
s
s
s
t "& !&"C#E/ / / / /b \ "& $/3"C#E(,M
M
M
M
M
b #' $(,"C#E "!%@
@
@
@
@
D 2 2 2 [2
4 4 4 [4 [ [ 9 9 9 [9
d0 d0 d0 d0 d0 d0 d0 d0L`0 `0 `0 `0 `0< `0 `0 `0Di i i i i i i ib
, 044
4
4
4
4
l 0 0 0 0"
$ 26/
/
/
/
/
b0 0 0 0 0 0r= r )8
__future__r r4 r` r email.parserr secretsr typingr r r r r
r r nats.errorsr\ nats.js.errors nats.aio.msgr
nats.aio.subscriptionr nats.jsr r r r r r
nats.js.kvr nats.js.managerr nats.js.object_storer r r r r r rZ bytearray
NATS_HDR_LINErT NATS_HDR_LINE_SIZE_CRLF_
_CRLF_LEN_r r r r r KV_MAX_HISTORYr r r= r; r s{ # " " " " " $ $ $ $ $ $ . . . . . . , , , , , , +&&
S ''
S[[
" ! UGYt_, - %/ !%6 " f0 f0 f0 f0 f0' f0 f0 f0 f0 f0r=