>i_AXdZddlZddlZddlZddlZddlZ ddlZddl Z ddl Z ddl Z ddl Z ddlmZddlmZddlmZmZddlmZmZej0eZGddeZGd d eZGd d eZGd deZGddeZ GddeZ!y#e$rddlZYwxYw)z Use pika with the Gevent IOLoop.N)BaseConnection)check_callback_arg)AbstractIOReferenceAbstractIOServices)AbstractSelectorIOLoopSelectorIOServicesAdaptercFeZdZdZ dfd Ze ddZxZS)GeventConnectionzwImplementation of pika's ``BaseConnection``. An async selector-based connection which integrates with Gevent. ctjjr td|xst t j }t|tr|}n t|}tt|3||||||y)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 Y/var/www/html/aiagents/fe/lib/python3.12/site-packages/pika/adapters/gevent_connection.pyrzGeventConnection.__init__'s< ;; ! !NO O&A.v~~/?@  m%7 8 D3MBD .   "  )E / Gc|xsttj}t|fd}j ||||S)z_Implement :py:classmethod::`pika.adapters.BaseConnection.create_connection()`. c4| td|dS)zConnection factory.zIExpected pika.connection.Parameters instance, but got None in params arg.F)rrr ) ValueError)paramsclsrs r connection_factoryz>GeventConnection.create_connection..connection_factoryfs2~ "IJJ&%)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 /Gb)-#' r!r c2eZdZdZdZedZdZdZy)_TSafeCallbackQueueziDispatch callbacks from any thread to be executed in the main thread efficiently with IO events. ctj|_tj\|_|_tj|_ y)zQ :param _GeventSelectorIOLoop loop: IO loop to add callbacks to. N) queueQueue_queueospipe_read_fd _write_fd threadingRLock _write_lockrs r rz_TSafeCallbackQueue.__init__|s6 kkm (* % 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|jj||j5tj|j ddddy#1swYyxYw)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_threadsafesD !    . HHT^^W - . . .s !AAc |jj}tj|jd|y#t j $rtjdYywxYw)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_callbacksT {{--/H GGDMM1 % J {{ 8 NN6 7 8sA(A/.A/N) r-r.r/r0rpropertyrBrIrQr!r r4r4ws* - .r!r4cZeZdZdZdZdZdZddZdZdZ d Z d Z d Z d Z d ZdZdZy)rzImplementation of `AbstractSelectorIOLoop` using the Gevent event loop. Required by implementations of `SelectorIOServicesAdapter`. rKrNc|xstj_i_tjj _t_fd}jjj|jy)z> :param gevent._interfaces.ILoop gevent_loop: c>~~jjy)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 4r!N) rr_hub_io_watchers_by_fdhubWaiter_waiterr4rX add_handlerrBREAD)r gevent_hubrZs` r rz_GeventSelectorIOLoop.__init__sn2&.."2 "$zz((*  34 5 --002M $r!cZ|jjjd|_y)zRelease the loop's resources.N)r[loopdestroyr@s r closez_GeventSelectorIOLoop.closes   r!ctjd|jjtjd|jj y)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.starts>  9:  CD r!c:|jjdy)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!ctj|jk(r;tj d|jj j |ytj dtj|jj j |}|jj|y)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} >> tyy ( LL; < IINN ' ' 1 LL> ? (()D)DhOH  8 8 Br!cr|jjj|}|j||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_laters- $$U+ H r!c$|jy)zURemove a timeout :param timeout_handle: Handle of timeout to remove N)rf)rtimeout_handles r remove_timeoutz$_GeventSelectorIOLoop.remove_timeouts r!c|jjj||}||j|<|j |||y)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_handlers@YY^^&&r62 &0#"f-r!c|j|}|j}|j|j|=|j|||y)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!sM ,,R0 &&  # #B ' Xv.r!c\|j|}|j|j|=y)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/s. ,,R0   # #B 'r!)N)r-r.r/r0raWRITEERRORrrfrkrnrsrwrzr`rrrSr!r rrsK D E E$*  "C2  . /(r!rc eZdZdZ ddZy)rzESelectorIOServicesAdapter implementation using Gevent's DNS resolver.c rt|j|||||||}|jt|S)zOImplement :py:meth:`.nbio_interface.AbstractIOServices.getaddrinfo()`. ) native_loophostportfamilysocktypeprotoflagsr*)_GeventAddressResolver_looprk_GeventIOLoopIOHandle) rrrr*rrrrresolvers r getaddrinfoz,_GeventSelectorIOServicesAdapter.getaddrinfo<s@*djj/3/3173;050529; $X..r!N)rrrr)r-r.r/r0rrSr!r rr9sO /r!rceZdZdZdZdZy)rzXImplement `AbstractIOReference`. Only used to wrap the _GeventAddressResolver. c&|j|_y)zY :param subject: subject of the reference containing a `cancel()` method N)cancel_cancel)rsubjects r rz_GeventIOLoopIOHandle.__init__Ys~~ r!c"|jS)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 y ) 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|_||_||_||_||_||_||_ y)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__zsN 7I.      $r!c|j%tj|j|_ytj dy)z-Start an asynchronous getaddrinfo invocation.Nz&_GeventAddressResolver already started)rr spawn_raw_resolverOrPr@s r rkz_GeventAddressResolver.starts/ >> !#--dmm> %G    ! r!c@|jd|_d|_y)z,Stop the resolver and release any resources.N)rrrr@s r rz_GeventAddressResolver._cleanups   r!cj|j'tj|jd|_yy)zbStop the greenlet performing getaddrinfo if running. Otherwise, this is a no-op. N)rrkillr@s r rz%_GeventAddressResolver._stop_greenlets* >> % KK '!DN &r!c tjj|j|j|j |j |j|j}tj|j|}|jj!|y#t$r"}tjd||}Yd}~bd}~wwxYw)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}}dmm/3/3/@/@/3~~t~~OF$$T%<%F sA B C 'CC c tjd|j|j||j y#|j wxYw)zInvoke the configured completion callback and any subsequent cleanup. :param result: result from getaddrinfo, or the exception if raised. z9Invoking async getaddrinfo() completion callback; host=%rN)rOrhrrr)rrs r rz)_GeventAddressResolver._dispatch_callbacks>   LLK   MM& ! MMODMMOs 1AAN) r-r.r/r0 __slots__rrkrrrrrrSr!r rrhs2 I<E  "*" r!r)"r0rqloggingr9r=weakrefr6 ImportErrorr7r gevent.hub gevent.socket pika.compatr pika.adapters.base_connectionr%pika.adapters.utils.io_services_utilsr"pika.adapters.utils.nbio_interfacerr+pika.adapters.utils.selector_ioloop_adapterrr getLoggerr-rOr objectr4rrrrrSr!r rs& 8D   8 $S~Sl2&2jJ(2J(Z/'@/4/*mVm{ sB B)(B)