)Jf_A|dZddlZddlZddlZddlZddlZ ddlZn#e$rddlZYnwxYwddl Z ddl Z ddl Z ddl Z ddlmZddlmZddlmZmZddlmZmZejeZGddeZGd d eZGd d eZGd deZGddeZ GddeZ!dS)z Use pika with the Gevent IOLoop.N)BaseConnection)check_callback_arg)AbstractIOReferenceAbstractIOServices)AbstractSelectorIOLoopSelectorIOServicesAdaptercLeZdZdZ dfd Ze ddZxZS)GeventConnectionzwImplementation of pika's ``BaseConnection``. An async selector-based connection which integrates with Gevent. NTc6tjjrtd|pt t j}t|tr|}nt|}tt| ||||||dS)aCreate a new GeventConnection instance and connect to RabbitMQ on Gevent's event-loop. :param pika.connection.Parameters|None parameters: The connection parameters :param callable|None on_open_callback: The method to call when the connection is open :param callable|None on_open_error_callback: Called if the connection can't be established or connection establishment is interrupted by `Connection.close()`: on_open_error_callback(Connection, exception) :param callable|None on_close_callback: Called when a previously fully open connection is closed: `on_close_callback(Connection, exception)`, where `exception` is either an instance of `exceptions.ConnectionClosed` if closed by user or broker or exception of another type that describes the cause of connection failure :param gevent._interfaces.ILoop|nbio_interface.AbstractIOServices|None custom_ioloop: Use a custom Gevent ILoop. :param bool internal_connection_workflow: True for autonomous connection establishment which is default; False for externally-managed connection workflow via the `create_connection()` factory z-GeventConnection is not supported on Windows.)internal_connection_workflowN) pikacompat ON_WINDOWS RuntimeError_GeventSelectorIOLoopgeventget_hub isinstancer _GeventSelectorIOServicesAdaptersuperr __init__) self parameterson_open_callbackon_open_error_callbackon_close_callback custom_ioloopr nbio __class__s b/home/alex/cs2snipeproduction/venv/lib/python3.11/site-packages/pika/adapters/gevent_connection.pyrzGeventConnection.__init__'s< ; ! PNOO O&A.v~/?/?@@  m%7 8 8 C DD3MBBD %%..   "  )E / G G G G Gc|pttj}t|fd}||||S)z_Implement :py:classmethod::`pika.adapters.BaseConnection.create_connection()`. cB|td|dS)zConnection factory.NzIExpected pika.connection.Parameters instance, but got None in params arg.F)rrr ) ValueError)paramsclsrs r connection_factoryz>GeventConnection.create_connection..connection_factoryfsD~ "IJJJ3&%)49;;; ;r!)connection_configsr'rworkflowon_done)rrrr_start_connection_workflow)r&r(r*rr)r'rs` @r create_connectionz"GeventConnection.create_connectionXs'A.v~/?/?@@ 0 >> ; ; ; ; ; ;--11 . r!)NNNNNT)NN)__name__ __module__ __qualname____doc__r classmethodr, __classcell__)rs@r r r !s !"&(,#'#.2 /G/G/G/G/G/Gb)-#' [r!r c:eZdZdZdZedZdZdZdS)_TSafeCallbackQueueziDispatch callbacks from any thread to be executed in the main thread efficiently with IO events. ctj|_tj\|_|_tj|_ dS)zQ :param _GeventSelectorIOLoop loop: IO loop to add callbacks to. N) queueQueue_queueospipe_read_fd _write_fd threadingRLock _write_lockrs r rz_TSafeCallbackQueue.__init__|s= kmm (* % t~%?,,r!c|jS)z?The file-descriptor to register for READ events in the IO loop.)r;r@s r fdz_TSafeCallbackQueue.fds }r!c|j||j5tj|jdddddS#1swxYwYdS)zAdd an item to the queue from any thread. The configured handler will be invoked with the item in the main thread. :param item: Object to add to the queue. N)r8putr?r9writer<rcallbacks r add_callback_threadsafez+_TSafeCallbackQueue.add_callback_threadsafes !!!   . . HT^W - - - . . . . . . . . . . . . . . . . . .sA  AAc |j}tj|jd|dS#t j$rtdYdSwxYw)aInvoke the next callback from the queue. MUST run in the main thread. If no callback was added to the queue, this will block the IO loop. Performs a blocking READ on the pipe so must only be called when the pipe is ready for reading. zCallback queue was empty.N) r8 get_nowaitr9readr;r6EmptyLOGGERwarningrGs r run_next_callbackz%_TSafeCallbackQueue.run_next_callbacksz {--//H GDM1 % % % HJJJJJ { 8 8 8 NN6 7 7 7 7 7 7 8sA)A.-A.N) r-r.r/r0rpropertyrBrIrQr!r r4r4wsf - - -X . . .r!r4c\eZdZdZdZdZdZddZdZdZ d Z d Z d Z d Z d ZdZdZdS)rzImplementation of `AbstractSelectorIOLoop` using the Gevent event loop. Required by implementations of `SelectorIOServicesAdapter`. rKrNc |ptj_i_tj_t_fd} jj |j dS)z> :param gevent._interfaces.ILoop gevent_loop: c>~~jdS)z$Swallow the fd and events arguments.N)_callback_queuerQ)rBeventsrs r run_callback_in_main_threadzC_GeventSelectorIOLoop.__init__..run_callback_in_main_threads$  2 2 4 4 4 4 4r!N) rr_hub_io_watchers_by_fdhubWaiter_waiterr4rX add_handlerrBREAD)r gevent_hubrZs` r rz_GeventSelectorIOLoop.__init__s2&."2"2 "$z((**  344 5 5 5 5 5 -02M $ $ $ $ $r!cP|jjd|_dS)zRelease the loop's resources.N)r[loopdestroyr@s r closez_GeventSelectorIOLoop.closes#     r!ctd|jtd|jdS)zNRun the I/O loop. It will loop until requested to exit. See `stop()`. z"Passing control to Gevent's IOLoopz,Control was passed back from Gevent's IOLoopN)rOdebugr_getclearr@s r startz_GeventSelectorIOLoop.startsX  9:::  CDDD r!c:|jddS)a$Request exit from the ioloop. The loop is NOT guaranteed to stop before this method returns. To invoke `stop()` safely from a thread other than this IOLoop's thread, call it via `add_callback_threadsafe`; e.g., `ioloop.add_callback(ioloop.stop)` N)r_switchr@s r stopz_GeventSelectorIOLoop.stops  D!!!!!r!cdtj|jkr;td|jj|dStdtj|jjj|}|j |dS)aRequests a call to the given function as soon as possible in the context of this IOLoop's thread. NOTE: This is the only thread-safe method in IOLoop. All other manipulations of IOLoop must be performed from the IOLoop's thread. For example, a thread may request a call to the `stop` method of an ioloop that is running in a different thread via `ioloop.add_callback_threadsafe(ioloop.stop)` :param callable callback: The callback method z Adding callback from main threadz#Adding callback from another threadN) rrr[rOrhrd run_callback functoolspartialrXrIrGs r add_callbackz"_GeventSelectorIOLoop.add_callbacks >  ty ( ( LL; < < < IN ' ' 1 1 1 1 1 LL> ? ? ? ()DhOOH  8 8 B B B B Br!cn|jj|}|||S)aAdd the callback to the IOLoop timer to be called after delay seconds from the time of call on best-effort basis. Returns a handle to the timeout. :param float delay: The number of seconds to wait to call callback :param callable callback: The callback method :returns: handle to the created timeout that may be passed to `remove_timeout()` :rtype: object )r[rdtimerrk)rdelayrHrus r call_laterz _GeventSelectorIOLoop.call_laters2 $$U++ H r!c.|dS)zURemove a timeout :param timeout_handle: Handle of timeout to remove N)rf)rtimeout_handles r remove_timeoutz$_GeventSelectorIOLoop.remove_timeouts r!c|jj||}||j|<||||dS)aStart watching the given file descriptor for events :param int fd: The file descriptor :param callable handler: When requested event(s) occur, `handler(fd, events)` will be called. :param int events: The event mask (READ|WRITE) N)r[rdior\rk)rrBhandlerrY io_watchers r r`z!_GeventSelectorIOLoop.add_handlersHY^&&r622 &0#"f-----r!c|j|}|j}||j|=||||dS)zChange the events being watched for. :param int fd: The file descriptor :param int events: The new event mask (READ|WRITE) N)r\rHrfr`)rrBrYr~rHs r update_handlerz$_GeventSelectorIOLoop.update_handler!sV ,R0 &  #B ' Xv.....r!cX|j|}||j|=dS)zgStop watching the given file descriptor for events :param int fd: The file descriptor N)r\rf)rrBr~s r remove_handlerz$_GeventSelectorIOLoop.remove_handler/s4 ,R0   #B ' ' 'r!)N)r-r.r/r0raWRITEERRORrrfrkrnrsrwrzr`rrrSr!r rrs D E E$$$$*  " " "CCC2    . . . / / /(((((r!rc"eZdZdZ ddZdS)rzESelectorIOServicesAdapter implementation using Gevent's DNS resolver.rc t|j|||||||}|t|S)zOImplement :py:meth:`.nbio_interface.AbstractIOServices.getaddrinfo()`. ) native_loophostportfamilysocktypeprotoflagsr*)_GeventAddressResolver_looprk_GeventIOLoopIOHandle) rrrr*rrrrresolvers r getaddrinfoz,_GeventSelectorIOServicesAdapter.getaddrinfo<sQ*dj/3/3173;050529;;; $X...r!N)rrrr)r-r.r/r0rrSr!r rr9s<OO //////r!rceZdZdZdZdZdS)rzXImplement `AbstractIOReference`. Only used to wrap the _GeventAddressResolver. c|j|_dS)zY :param subject: subject of the reference containing a `cancel()` method N)cancel_cancel)rsubjects r rz_GeventIOLoopIOHandle.__init__Ys~ r!c*|S)zCancel pending operation :returns: False if was already done or cancelled; True otherwise :rtype: bool )rr@s r rz_GeventIOLoopIOHandle.cancel_s ||~~r!N)r-r.r/r0rrrSr!r rrSs< &&& r!rc@eZdZdZdZdZdZdZdZdZ dZ d Z d S) rzPerforms getaddrinfo asynchronously Gevent's configured resolver in a separate greenlet and invoking the provided callback with the result. See: http://www.gevent.org/dns.html ) r_on_done _greenlet_ga_host_ga_port _ga_family _ga_socktype _ga_proto _ga_flagsc t|d||_||_d|_||_||_||_||_||_||_ dS)aInitialize the `_GeventAddressResolver`. :param AbstractSelectorIOLoop native_loop: :param host: `see socket.getaddrinfo()` :param port: `see socket.getaddrinfo()` :param family: `see socket.getaddrinfo()` :param socktype: `see socket.getaddrinfo()` :param proto: `see socket.getaddrinfo()` :param flags: `see socket.getaddrinfo()` :param on_done: on_done(records|BaseException) callback for reporting result from the given I/O loop. The single arg will be either an exception object (check for `BaseException`) in case of failure or the result returned by `socket.getaddrinfo()`. r*N) rrrrrrrrrr) rrrrrrrrr*s r rz_GeventAddressResolver.__init__zsX 7I...      $r!c|j tj|j|_dStddS)z-Start an asynchronous getaddrinfo invocation.Nz&_GeventAddressResolver already started)rr spawn_raw_resolverOrPr@s r rkz_GeventAddressResolver.starts; > !#-dm< %G    ! ! ! r!cJ|d|_d|_dS)z,Stop the resolver and release any resources.N)rrrr@s r rz_GeventAddressResolver._cleanups&   r!cX|j"tj|jd|_dSdS)zbStop the greenlet performing getaddrinfo if running. Otherwise, this is a no-op. N)rrkillr@s r rz%_GeventAddressResolver._stop_greenlets1 > % K ' ' '!DNNN & %r!c^ tj|j|j|j|j|j|j}n4#t$r'}t d||}Yd}~nd}~wwxYwtj |j|}|j|dS)zoCall `getaddrinfo()` and return result via user's callback function on the configured IO loop. zAddress resolution failed: %rN)rsocketrrrrrrr ExceptionrOerrorrqrr_dispatch_callbackrrs)rresultexcrHs r rz_GeventAddressResolver._resolves ]..t}dm/3/3/@/3~t~OOFF    LL8# > > >FFFFFF $T%rsX&& LLLL 888888DDDDDD  8 $ $SSSSS~SSSl22222&222jJ(J(J(J(J(2J(J(J(Z/////'@///4/*mmmmmVmmmmms  ))