.Y'^ d dl mZ d dlZd dlZd dlmZ d dlmZmZm Z m
Z
m Z m Z d dl
mZ d dlmZ d dlmZmZmZ erd dlmZ ed Z ee Zd
Z ee Z G d d Z dS )
) annotationsN) BytesParser)
TYPE_CHECKINGAnyDictIterableListOptional)NoRespondersError)api)APIError
NotFoundErrorServiceUnavailableError)NATSs NATS/1.0s
c e Zd ZdZej dfdEd ZdFd
ZdGdZdHdIdZ dHdJdZ
dHdJdZ dKdZ dLdMd Z
dHdNd#ZdOdPd&ZdOdQd(Z dRdSd+ZdTd,Z dHdUd/Z dHdVd0ZdHdWd3Z dXdYd:ZedZd; Zd[d=Z d\d]d>Z d^d_dDZdS )`JetStreamManagerzA
JetStreamManager exposes management APIs for JetStream.
connr prefixstrtimeoutfloatreturnNonec V || _ || _ || _ t | _ d S N)_prefix_nc_timeoutr _hdr_parser)selfr r r s o/builddir/build/BUILD/imunify360-venv-2.6.2/opt/imunify360/venv/lib/python3.11/site-packages/nats/js/manager.py__init__zJetStreamManager.__init__( s+
&== api.AccountInfoc K | | j dd| j d {V }t j | S )Nz.INFOr$ r ) _api_requestr r r AccountInfo
from_response)r! resps r" account_infoz JetStreamManager.account_info3 sU &&$,'='='=sDM&ZZZZZZZZ,,T222r$ subjectc K | j d}t j d|i }| || | j d{V }|d st |d d S )zK
Find the stream to which a subject belongs in an account.
z
.STREAM.NAMESr- r' Nstreamsr )r jsondumpsr( encoder r )r! r- req_subreq_datainfos r" find_stream_name_by_subjectz,JetStreamManager.find_stream_name_by_subject7 s
\000:y' 233&&w0A0A4=&YYYYYYYYI Iq!!r$ Nnamesubjects_filter
Optional[str]api.StreamInfoc K d}|rt j d|i }| | j d| | | j d{V }t j | S )z;
Get the latest StreamInfo by stream name.
r8 z
.STREAM.INFO.r' N) r0 r1 r( r r2 r r
StreamInfor* )r! r7 r8 r4 r+ s r" stream_infoz JetStreamManager.stream_infoC s Hz#4o"FGGH&&| 0 0$ 0 0 OO M '
~++D111r$ configOptional[api.StreamConfig]c f K |t j } |j d
i |}|j t d t d }t
fd|D }t
d D } }|s|s|rt d d t j |
}| | j d |
| j d{V }t j | S ) z.
add_stream creates a stream.
N nats: stream name is requiredz.*>/\c 3 K | ]}|v V d S r ).0char stream_names r" z.JetStreamManager.add_stream.._ s( N N 3 N N N N N Nr$ c 3 > K | ]}| V d S r )isspace)rE rF s r" rH z.JetStreamManager.add_stream..` s* D DT\\^^ D D D D D Dr$ znats: stream name (z) is invalid. Names cannot contain whitespace, '.', '*', '>', path separators (forward or backward slash), or non-printable characters.z.STREAM.CREATE.r' rD )r StreamConfigevolver7
ValueErrorsetany isprintabler0 r1 as_dictr( r r2 r r= r* )
r! r? params
invalid_charshas_invalid_charshas_whitespaceis_not_printabledatar+ rG s
@r"
add_streamzJetStreamManager.add_streamQ s >%''F (((( k <== = H
N N N N
N N N N N D D D D DDD * 6 6 8 88 2B \k \ \ \
z&..**++&&| 9 9K 9 9 KKMMM '
~++D111r$ c K |t j } |j di |}|j t d t j | }| | j d|j |
| j d{V }t j
| S )z1
update_stream updates a stream.
NrB z.STREAM.UPDATE.r' rD )r rK rL r7 rM r0 r1 rQ r( r r2 r r= r* )r! r? rR rW r+ s r"
update_streamz JetStreamManager.update_streamq s >%''F (((( ; <== =z&..**++&&| 9 9FK 9 9 KKMMM '
~++D111r$ boolc n K | | j d| | j d{V }|d S )z*
Delete a stream by name.
z.STREAM.DELETE.r' Nsuccessr( r r )r! r7 r+ s r"
delete_streamz JetStreamManager.delete_stream sP &&$,'M'Mt'M'MW[Wd&eeeeeeeeI r$ seq
Optional[int]keepc K i }|r||d< |r||d< |r||d< t j | }| | j d| | | j d{V }|d S )z)
Purge a stream by name.
r` filterrb z.STREAM.PURGE.r' Nr] )r0 r1 r( r r2 r )r! r7 r` r- rb
stream_reqreqr+ s r" purge_streamz JetStreamManager.purge_stream s &(
$ #Ju +#*Jx &!%Jv j$$&&$,'L'Ld'L'Lcjjlldhdq&rrrrrrrrI r$ streamconsumerOptional[float]c K || j }| | j d| d| d| d {V }t j | S )Nz.CONSUMER.INFO..r$ r' )r r( r r ConsumerInfor* )r! rh ri r r+ s r"
consumer_infoz JetStreamManager.consumer_info sp ?mG&&$,'Z'Zv'Z'ZPX'Z'Z\_ip&qqqqqqqq --d333r$ r List[api.StreamInfo]c . K | | j dt j d|i | j d{V }g }|d D ]6}t j | }| | 7|S )zS
streams_info retrieves a list of streams with an optional offset.
.STREAM.LISToffsetr' Nr/ )
r( r r0 r1 r2 r r r= r* append)r! rr r+ r/ rh r> s r" streams_infoz JetStreamManager.streams_info s &&| ) ) ) J&) * * 1 1 3 3M '
9o ( (F .66v>>K NN; ' ' ' 'r$ Iterable[api.StreamInfo]c K | | j dt j d|i | j d{V }t
j |d |d |d S )zD
streams_info retrieves a list of streams Iterator.
rq rr r' Ntotalr/ )r( r r0 r1 r2 r r StreamsListIterator)r! rr r+ s r" streams_info_iteratorz&JetStreamManager.streams_info_iterator s &&| ) ) ) J&) * * 1 1 3 3M '
&tH~tG}d9oVVVr$ Optional[api.ConsumerConfig]api.ConsumerInfoc K |s| j }|t j } |j d
i |}|j }|| d}t
j | }d }d} | j j
}
|
j dk o
|
j dk } | rK|j
rD|j r(|j dk r | j d| d|j
d|j } n3| j d| d|j
} n |r| j d| d| } n | j d| } | | || d {V }t j | S ) N)rG r? r< >z.CONSUMER.CREATE.rl z.CONSUMER.DURABLE.CREATE.r' rD )r r ConsumerConfigrL durable_namerQ r0 r1 r2 r connected_server_versionmajorminorr7 filter_subjectr r( rm r* ) r! rh r? r rR r rf r4 r+ r- versionconsumer_name_supporteds r" add_consumerz JetStreamManager.add_consumer s $mG >'))F (((( * $0@0@AA:c??))++(3")-1"4"K !9K " Av{ A$
S)>#)E)E !\jjFjjV[jjSYShjj !\RRFRRV[RR
A WWWW WWGG @@@@G&&w'&JJJJJJJJ --d333r$ c v K | | j d| d| d| j d {V }|d S )Nz.CONSUMER.DELETE.rl r$ r' r] r^ )r! rh ri r+ s r" delete_consumerz JetStreamManager.delete_consumer sm &&| A Af A Ax A A M '
I r$ pause_untilapi.ConsumerPausec K || j }d|i}t j | }| | j d| d| || d{V }t j | S )a
Pause a consumer until the specified time.
Args:
stream: The stream name
consumer: The consumer name
pause_until: RFC 3339 timestamp string (e.g., "2025-10-22T12:00:00Z")
until which the consumer should be paused
timeout: Request timeout in seconds
Returns:
ConsumerPause with paused status
Note:
Requires nats-server 2.11.0 or later
Nr z.CONSUMER.PAUSE.rl r' ) r r0 r1 r2 r( r r
ConsumerPauser* )r! rh ri r r rf r4 r+ s r" pause_consumerz JetStreamManager.pause_consumer s . ?mG k*:c??))++&&| @ @V @ @h @ @ '
..t444r$ c B K | ||d| d{V S )a
Resume a paused consumer immediately.
This is equivalent to calling pause_consumer with a timestamp in the past.
Args:
stream: The stream name
consumer: The consumer name
timeout: Request timeout in seconds
Returns:
ConsumerPause with paused=False
Note:
Requires nats-server 2.11.0 or later
z1970-01-01T00:00:00ZN)r )r! rh ri r s r" resume_consumerz JetStreamManager.resume_consumer s6 . ((;QSZ[[[[[[[[[r$ rr List[api.ConsumerInfo]c : K | | j d| |dn't j d|i | j d{V }g }|d D ]6}t j | }| | 7|S )z
consumers_info retrieves a list of consumers. Consumers list limit is 256 for more
consider to use offset
:param stream: stream to get consumers
:param offset: consumers list offset
z.CONSUMER.LIST.Nr$ rr r' consumers)
r( r r0 r1 r2 r r rm r* rs )r! rh rr r+ r ri rn s r" consumers_infoz JetStreamManager.consumers_info+ s &&| 4 4F 4 4>CCtz8V2D'E'E'L'L'N'NM '
[) , ,H , : :8 D DM ] + + + +r$ FrG directOptional[bool]nextapi.RawStreamMsgc , K d}i }|r||d< |r d|d< | dd ||d< |r%||d< d|d< | dd ||d< t j | }|rx|r|d}| j d| d| }n | j d| }| j || | j d{V } t | }
|
S | j d | }|
|| | j d{V } t j
| d
}
|
j rt j |
j } | t" t$ z d }
| j |
}d}t+ | d k r!i }| D ]
\ }}|||< ||
_ d}|
j rt j |
j }||
_ |
S ) z<
get_msg retrieves a message from a stream.
Nr` last_by_subj next_by_subjr< z .DIRECT.GET.rl r' z.STREAM.MSG.GET.messager )popr0 r1 r r requestr2 r r _lift_msg_to_raw_msgr( r RawStreamMsgr* hdrsbase64 b64decodeNATS_HDR_LINE_SIZE
_CRLF_LEN_r
parsebyteslenitemsheadersrW )r! rG r` r- r r req_subjectrf rW r+ raw_msg resp_datar raw_headersparsed_headersr kvmsg_datas r" get_msgzJetStreamManager.get_msg= so CJ * CJ GGE4 ")C * CJ"&C GGND ) ) )")C z#
ICK!% R R; R R R R !% H H; H H ))+t{{}}dm)\\\\\\\\D&;;DAAGN DD{DD ++K PTP]+^^^^^^^^ "0091EFF < &#GL11D 1J > @ @AK ! - 8 8 E ENG>''))**Q.. * 0 0 2 2 # #DAq!"GAJJ %GO$( < 6 ' 55H r$ c z |j sDd |_ |j d }|r!|dk rt t j | t
j }|j d }||_ |j d }|rt | |_
|j |_ |j |_ |S )NStatus404z Nats-Subjectz
Nats-Sequence) rW r getr r
from_msgr r r- intr` )r! msgstatusr r- r` s r" r z%JetStreamManager._lift_msg_to_raw_msg{ s x 1CH[__X..F
1U??''"+C000"$$+n-!kooo.. # c((GKx +r$ r c K | j d| }d|i}t j | }| || d{V }|d S )zX
delete_msg retrieves a message from a stream based on the sequence ID.
z.STREAM.MSG.DELETE.r` Nr] )r r0 r1 r( r2 )r! rG r` r rf rW r+ s r"
delete_msgzJetStreamManager.delete_msg sl GG+GG clz#&&{DKKMMBBBBBBBBI r$ c B K | ||| d{V S )zH
get_last_msg retrieves the last message from a stream.
)r- r N)r )r! rG r- r s r" get_last_msgz JetStreamManager.get_last_msg s2 \\+wv\NNNNNNNNNr$ r$ r rf bytesDict[str, Any]c K | j ||| d {V }t j |j }n# t
$ r t w xY wd|v rt j |d |S )Nr' error) r r r0 loadsrW r r r
from_error)r! r rf r r r+ s r" r( z JetStreamManager._api_request s * ((c7(KKKKKKKKC:ch''DD * * *) ) * d??%d7m44 4 s rX rZ r_ rg rn rt ry r r r r r r classmethodr r r r( rD r$ r" r r # so ( ) ) ) ) )3 3 3 3
"
"
"
" 2 2 2 2 2 2 2 2 2 2@2 2 2 2 2$ "!% " ,4 4 4 4 4
W
W
W
W
W 04#' "4 "4 "4 "4 "4H $( "5 "5 "5 "5 "5P $( \ \ \ \ \2 * "!%!& $
< < < < <| [, "' O O O O O r$ r )
__future__r r r0 email.parserr typingr r r r r r
nats.errorsr nats.jsr nats.js.errorsr
r r natsr bytearray
NATS_HDR_LINEr r _CRLF_r r rD r$ r" r s6 # " " " " "
$ $ $ $ $ $ E E E E E E E E E E E E E E E E ) ) ) ) ) ) K K K K K K K K K K +&&
S ''
S[[
T T T T T T T T T Tr$