o RhT@s`ddlZddlZddlZddlmZddlmZddlmZddl m Z m Z ddl m Zddl mZddlmZmZmZmZmZmZmZmZdd lmZd Zd Zed ZeeZ Gd ddeeZ!Gddde!eeeZ"Gddde!eeeZ#GdddeeZ$Gddde"eZ%Gddde#eZ&Gddde$eZ'Gddde$eZ(dS)N) QueueEmpty) QueueFull)deque)heappopheappush)Empty)Full)AnyCallableDequeGenericListOptionalSetTypeVar)Protocolz1.0.0)Queue PriorityQueue LifoQueue SyncQueue AsyncQueue BaseQueueTc@seZdZedefddZedefddZdddZdefd d Z edefd d Z defd dZ defddZ de ddfddZde fddZdS)rreturncCdSNselfrrP/home/air/sanwanet/backup_V2/venv/lib/python3.10/site-packages/janus/__init__.pymaxsizezBaseQueue.maxsizecCrrrrrrrclosed"r!zBaseQueue.closedNcCrrrrrrr task_done&zBaseQueue.task_donecCrrrrrrrqsize)r$zBaseQueue.qsizecCrrrrrrrunfinished_tasks,r!zBaseQueue.unfinished_taskscCrrrrrrrempty0r$zBaseQueue.emptycCrrrrrrrfull3r$zBaseQueue.fullitemcCrrrrr)rrr put_nowait6r$zBaseQueue.put_nowaitcCrrrrrrr get_nowait9r$zBaseQueue.get_nowaitrN)__name__ __module__ __qualname__propertyintr boolr"r#r%r&r'r(rr+r,rrrrrs rc @seZdZedefddZedefddZdddZdefd d Z edefd d Z defd dZ defddZ de ddfddZde fddZd de dededdfddZd dedede fddZdddZdS)!rrcCrrrrrrrr >r!zSyncQueue.maxsizecCrrrrrrrr"Br!zSyncQueue.closedNcCrrrrrrrr#Fr$zSyncQueue.task_donecCrrrrrrrr%Ir$zSyncQueue.qsizecCrrrrrrrr&Lr!zSyncQueue.unfinished_taskscCrrrrrrrr'Pr$zSyncQueue.emptycCrrrrrrrr(Sr$zSyncQueue.fullr)cCrrrr*rrrr+Vr$zSyncQueue.put_nowaitcCrrrrrrrr,Yr$zSyncQueue.get_nowaitTblocktimeoutcCrrr)rr)r4r5rrrput\r$z SyncQueue.putcCrrr)rr4r5rrrget_r$z SyncQueue.getcCrrrrrrrjoinbr$zSyncQueue.joinr-TN)r.r/r0r1r2r r3r"r#r%r&r'r(rr+r,OptFloatr6r7r8rrrrr=s  rc@s6eZdZdeddfddZdefddZd dd ZdS) rr)rNcdSrrr*rrrr6gzAsyncQueue.putcr;rrrrrrr7jr<zAsyncQueue.getcr;rrrrrrr8mr<zAsyncQueue.joinr-)r.r/r0rr6r7r8rrrrrfsrc@seZdZd+deddfddZd,ddZd,d d Zedefd d Z edefd dZ ed-ddZ ed.ddZ deddfddZ defddZdeddfddZdefddZdeddfddZd,d d!Zd,d"d#Zd$eddfd%d&Zd$eddfd'd(Zd,d)d*ZdS)/rrr rNcs"t_|_|d_t_t j_ t j_ t j_ t_ tjdddkrBtj dddt j _t j _t_jd_t_dtd d td dffd d }|_dtd d td dffdd }|_t_t_dS)Nr)r= r _get_loopcSrrrrrrrsz Queue.__init__..Fcallback).Nargsrcs0z jj|g|RWdStyYdSwr)_loopcall_soon_threadsafe RuntimeErrorrArBrrrchecked_call_soon_threadsafes  z4Queue.__init__..checked_call_soon_threadsafecs&jsjj|g|RdSdSr)rC is_closed call_soonrFrrrchecked_call_soons z)Queue.__init__..checked_call_soon) asyncioget_running_looprC_maxsize_init_unfinished_tasks threadingLock _sync_mutex Condition_sync_not_empty_sync_not_full_all_tasks_done _async_mutexsys version_infogetattr_async_not_empty_async_not_fullEvent _finishedset_closing_pendingr r _call_soon_threadsafe _call_soon_SyncQueueProxy _sync_queue_AsyncQueueProxy _async_queue)rr rGrJrrr__init__rs:        zQueue.__init__cCsZ|j d|_|jD]}|q |j|jWddS1s&wYdS)NT)rRr`racancelr^r_rV notify_all)rfutrrrcloses    "z Queue.closecs@|jstdtdIdH|jsdSt|jIdHdS)NzWaiting for non-closed queuer)r`rErKsleeprawaitrrrr wait_closedszQueue.wait_closedcCs|jo|j Sr)r`rarrrrr"sz Queue.closedcC|jSr)rMrrrrr z Queue.maxsize_SyncQueueProxy[T]cCrpr)rerrrrsync_qrqz Queue.sync_q_AsyncQueueProxy[T]cCrpr)rgrrrrasync_qrqz Queue.async_qcCs t|_dSr)r_queuerr rrrrNs z Queue._initcC t|jSrlenrvrrrr_qsize z Queue._qsizer)cC|j|dSrrvappendr*rrr_putz Queue._putcC |jSr)rvpopleftrrrr_getr|z Queue._getcCs&|||jd7_|jdS)N)rrOr^clearr*rrr _put_internals zQueue._put_internalcs dfdd }jd|dS)Nrc6jjWddS1swYdSr)rRrTnotifyrrrrf "z'Queue._notify_sync_not_empty..fr-)rCrun_in_executor)rrrrr_notify_sync_not_emptyszQueue._notify_sync_not_emptycs@dfdd }tjd|}|jjj|dS)Nrcrr)rRrUrrrrrrrz&Queue._notify_sync_not_full..fr-)rK ensure_futurerCradd_done_callbackradiscardadd)rrrkrrr_notify_sync_not_fullszQueue._notify_sync_not_full threadsafec>dfdd dfdd }|r|dS|dS)Nrc Lj4IdHjWdIdHdS1IdHswYdSr)rWr[rrrrrr .z(Queue._notify_async_not_empty..fc,j}|jjj|dSrrC create_taskrrarrtaskrrrr task_makerz1Queue._notify_async_not_empty..task_makerr-rbrcrrrrrr_notify_async_not_empty zQueue._notify_async_not_emptycr)Nrc rr)rWr\rrrrrrrz'Queue._notify_async_not_full..fcrrrrrrrrrz0Queue._notify_async_not_full..task_makerr-rrrrr_notify_async_not_fullrzQueue._notify_async_not_fullcCs|jrtddS)Nz*Operation on the closed queue is forbidden)r`rErrrr_check_closing szQueue._check_closing)rr-)rrr)rrt)r.r/r0r2rhrlror1r3r"r rsrurNr{rrrrrrrrrrrrrrqs, -      rc @seZdZdZdeefddZedefddZ ede fdd Z d#d d Z d#d dZ defddZedefddZde fddZde fddZd$dede dedd fddZd$de dedefddZdedd fdd Zdefd!d"Zd S)%rdjCreate a queue object with a given maximum size. If maxsize is <= 0, the queue size is infinite. parentcC ||_dSr_parentrrrrrrhr|z_SyncQueueProxy.__init__rcC|jjSrrrMrrrrr z_SyncQueueProxy.maxsizecCrrrr"rrrrr"rz_SyncQueueProxy.closedNcCs|j|jj/|jjd}|dkr,|dkrtd|jj|jj|jjj ||j_WddS1s;wYdS)a.Indicate that a formerly enqueued task is complete. Used by Queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete. If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue). Raises a ValueError if called more times than there were items placed in the queue. rr!task_done() called too many timesN) rrrVrO ValueErrorrjrCrDr^r_)r unfinishedrrrr#!s     "z_SyncQueueProxy.task_donecCsn|j|jj$|jjr%|jj|j|jjsWddSWddS1s0wYdS)aBlocks until all items in the Queue have been gotten and processed. The count of unfinished tasks goes up whenever an item is added to the queue. The count goes down whenever a consumer thread calls task_done() to indicate the item was retrieved and all work on it is complete. When the count of unfinished tasks drops to zero, join() unblocks. N)rrrVrOrnrrrrr89s    "z_SyncQueueProxy.joincCr)z9Return the approximate size of the queue (not reliable!).rr{rrrrr%H z_SyncQueueProxy.qsizecCrz&Return the number of unfinished tasks.rrOrrrrr&Lz _SyncQueueProxy.unfinished_taskscCs |j S)aReturn True if the queue is empty, False otherwise (not reliable!). This method is likely to be removed at some point. Use qsize() == 0 as a direct substitute, but be aware that either approach risks a race condition where a queue can grow before the result of empty() or qsize() can be used. To create code that needs to wait for all queued tasks to be completed, the preferred technique is to use the join() method. rrrrrr'Qs z_SyncQueueProxy.emptycCs"d|jjko|jkSS)aOReturn True if the queue is full, False otherwise (not reliable!). This method is likely to be removed at some point. Use qsize() >= n as a direct substitute, but be aware that either approach risks a race condition where a queue can shrink before the result of full() or qsize() can be used. r)rrMr{rrrrr(^s"z_SyncQueueProxy.fullTr)r4r5cCs,|j|jj|jjdkrq|s|j|jjkrtnS|dur;|j|jjkr:|jj|j|jjks+n6|dkrCtd|jjj }||}|j|jjkrq||}|dkrat|jj||j|jjksV|j ||jj |jj ddWddS1swYdS)aPut an item into the queue. If optional args 'block' is true and 'timeout' is None (the default), block if necessary until a free slot is available. If 'timeout' is a non-negative number, it blocks at most 'timeout' seconds and raises the Full exception if no free slot was available within that time. Otherwise ('block' is false), put an item on the queue if a free slot is immediately available, else raise the Full exception ('timeout' is ignored in that case). rN''timeout' must be a non-negative numberTr)rrrUrMr{ SyncQueueFullrnrrCtimerrTrr)rr)r4r5rendtime remainingrrrr6hs4        "z_SyncQueueProxy.putcCs|j|jji|s|jstnC|dur)|js(|jj|jrn.|dkr1td|jjj}||}|jsW||}|dkrKt|jj||jr@|j }|jj |jj dd|WdS1suwYdS)aRemove and return an item from the queue. If optional args 'block' is true and 'timeout' is None (the default), block if necessary until an item is available. If 'timeout' is a non-negative number, it blocks at most 'timeout' seconds and raises the Empty exception if no item was available within that time. Otherwise ('block' is false), return an item if one is immediately available, else raise the Empty exception ('timeout' is ignored in that case). NrrrTr) rrrTr{SyncQueueEmptyrnrrCrrrUrr)rr4r5rrrr)rrrr7s4           $z_SyncQueueProxy.getcCs|j|ddS)zPut an item into the queue without blocking. Only enqueue the item if a free slot is immediately available. Otherwise raise the Full exception. Fr4)r6r*rrrr+sz_SyncQueueProxy.put_nowaitcCs |jddS)zRemove and return an item from the queue without blocking. Only get an item if one is immediately available. Otherwise raise the Empty exception. Fr)r7rrrrr,s z_SyncQueueProxy.get_nowaitr-r9)r.r/r0__doc__rrrhr1r2r r3r"r#r8r%r&r'r(r:r6r7r+r,rrrrrds$    ""rdc@seZdZdZdeefddZedefddZ de fdd Z ede fd d Z ede fd d Z defddZdefddZdeddfddZdeddfddZdefddZdefddZd ddZd ddZdS)!rfrrcCrrrrrrrrhr|z_AsyncQueueProxy.__init__rcCrrrrrrrr"rz_AsyncQueueProxy.closedcCr)zNumber of items in the queue.rrrrrr%rz_AsyncQueueProxy.qsizecCrrrrrrrr&rz!_AsyncQueueProxy.unfinished_taskscCr)z%Number of items allowed in the queue.rrrrrr rz_AsyncQueueProxy.maxsizecCs |dkS)z3Return True if the queue is empty, False otherwise.r)r%rrrrr's z_AsyncQueueProxy.emptycCs |jjdkrdS||jjkS)zReturn True if there are maxsize items in the queue. Note: if the Queue was initialized with maxsize=0 (the default), then full() is never True. rF)rrMr%rrrrr(s z_AsyncQueueProxy.fullr)Nc s |j|jj4IdHj|jjd}zK|jjdkrHd}|rH|j|jjk}|rFd}|jj|jjIdH|jjd}|s"|j ||jj |j W|rb|jjn |rl|jjwwWdIdHdS1IdHs~wYdS)zPut an item into the queue. Put an item into the queue. If the queue is full, wait until a free slot is available before adding item. This method is a coroutine. NTrF) rrr\rRacquirerMr{releasernrr[rr)rr)lockeddo_waitrrrr6s8        .z_AsyncQueueProxy.putcCs~|j|jj,|jjdkr|j|jjkrt|j||jjdd|jWddS1s8wYdS)zyPut an item into the queue without blocking. If no free slot is immediately available, raise QueueFull. rFrN) rrrRrMr{AsyncQueueFullrrrr*rrrr+s     "z_AsyncQueueProxy.put_nowaitc s|j|jj4IdHa|jjd}zMd}|r@|jdk}|r>d}|jj|jjIdH|jjd}|s|j}|jj |j |W|rZ|jjWdIdHS|rn|jjww1IdHsuwYdS)zRemove and return an item from the queue. If queue is empty, wait until an item is available. This method is a coroutine. NTrF) rrr[rRrr{rrnrr\rr)rrrr)rrrr7s6       z_AsyncQueueProxy.getcCsn|j|jj$|jdkrt|j}|jjdd|j|WdS1s0wYdS)zRemove and return an item from the queue. Return an item if one is immediately available, else raise QueueEmpty. rFrN)rrrRr{AsyncQueueEmptyrrrr*rrrr,4s    $z_AsyncQueueProxy.get_nowaitcCs|j|jj5|jjdkrtd|jjd8_|jjdkr6|jj|jjWddSWddS1sAwYdS)a$Indicate that a formerly enqueued task is complete. Used by queue consumers. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete. If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue). Raises ValueError if called more times than there were items placed in the queue. rrrN)rrrVrOrr^r_rjrrrrr#Cs     "z_AsyncQueueProxy.task_donecsh |jj|j|jjdkr WddSWdn1s%wY|jjIdHq)aBlock until all items in the queue have been gotten and processed. The count of unfinished tasks goes up whenever an item is added to the queue. The count goes down whenever a consumer calls task_done() to indicate that the item was retrieved and all work on it is complete. When the count of unfinished tasks drops to zero, join() unblocks. TrN)rrRrrOr^rnrrrrr8Zs   z_AsyncQueueProxy.joinr-)r.r/r0rrrrhr1r3r"r2r%r&r r'r(r6r+r7r,r#r8rrrrrfs$  rfc@sPeZdZdZdeddfddZdefddZd eddfd d Zdefd d Z dS)rzVariant of Queue that retrieves open entries in priority order (lowest first). Entries are typically tuples of the form: (priority number, data). r rNcCs g|_dSr) _heap_queuerwrrrrNrr|zPriorityQueue._initcCrxr)rzrrrrrr{ur|zPriorityQueue._qsizer)cCst|j|dSr)rrr*rrrrxrzPriorityQueue._putcCrxr)rrrrrrr{r|zPriorityQueue._get) r.r/r0rr2rNr{rrrrrrrrjs rc@s>eZdZdZdefddZdeddfddZdefd d ZdS) rzBVariant of Queue that retrieves most recently added entries first.rcCrxrryrrrrr{r|zLifoQueue._qsizer)NcCr}rr~r*rrrrrzLifoQueue._putcCrr)rvpoprrrrrr|zLifoQueue._get) r.r/r0rr2r{rrrrrrrrs r))rKrXrPrrrr collectionsrheapqrrqueuerrrrtypingr r r r r rrrtyping_extensionsr __version____all__rfloatr:rrrrrdrfrrrrrrs4     (   )  ..