o 1li_A@sdZddlZddlZddlZddlZddlZzddlZWn ey)ddlZYnwddl Z ddl Z ddl Z ddl Z ddlmZddlmZddlmZmZddlmZmZeeZGdddeZGd d d eZGd d d eZGd ddeZGdddeZ GdddeZ!dS)z Use pika with the Gevent IOLoop.N)BaseConnection)check_callback_arg)AbstractIOReferenceAbstractIOServices)AbstractSelectorIOLoopSelectorIOServicesAdaptercs@eZdZdZ      dfdd Ze  d ddZZS) GeventConnectionzwImplementation of pika's ``BaseConnection``. An async selector-based connection which integrates with Gevent. NTcsXtjjrtd|ptt}t|tr|}nt |}t t |j ||||||ddS)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__o/var/www/html/tatsat2dev/dashboard-backend/venv/lib/python3.10/site-packages/pika/adapters/gevent_connection.pyr's     zGeventConnection.__init__cs:|ptt}t|fdd}j||||dS)z_Implement :py:classmethod::`pika.adapters.BaseConnection.create_connection()`. cs|durtd|ddS)zConnection factory.NzIExpected pika.connection.Parameters instance, but got None in params arg.F)rrr ) ValueError)paramsclsrrrconnection_factoryfs z>GeventConnection.create_connection..connection_factory)connection_configsr$rworkflowon_done)rrrr_start_connection_workflow)r#r%r'rr&r$rr"rcreate_connectionXs  z"GeventConnection.create_connection)NNNNNT)NN)__name__ __module__ __qualname____doc__r classmethodr) __classcell__rrrrr!s1rc@s4eZdZdZddZeddZddZdd Zd S) _TSafeCallbackQueueziDispatch callbacks from any thread to be executed in the main thread efficiently with IO events. cCs(t|_t\|_|_t|_ dS)zQ :param _GeventSelectorIOLoop loop: IO loop to add callbacks to. N) queueQueue_queueospipe_read_fd _write_fd threadingRLock _write_lockrrrrr|s z_TSafeCallbackQueue.__init__cCs|jS)z?The file-descriptor to register for READ events in the IO loop.)r6r;rrrfdsz_TSafeCallbackQueue.fdcCsF|j||jt|jdWddS1swYdS)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)r3putr:r4writer7rcallbackrrradd_callback_threadsafes "z+_TSafeCallbackQueue.add_callback_threadsafecCsHz|j}WntjytdYdSwt|jd|dS)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) r3 get_nowaitr1EmptyLOGGERwarningr4readr6r@rrrrun_next_callbacks  z%_TSafeCallbackQueue.run_next_callbackN) r*r+r,r-rpropertyr<rBrIrrrrr0ws  r0c@sneZdZdZdZdZdZdddZdd Zd d Z d d Z ddZ ddZ ddZ ddZddZddZdS)rzImplementation of `AbstractSelectorIOLoop` using the Gevent event loop. Required by implementations of `SelectorIOServicesAdapter`. rCrNcsL|pt_i_tj_t_fdd} jj |j dS)z> :param gevent._interfaces.ILoop gevent_loop: cs~~jdS)z$Swallow the fd and events arguments.N)_callback_queuerI)r<eventsr;rrrun_callback_in_main_threadszC_GeventSelectorIOLoop.__init__..run_callback_in_main_threadN) rr_hub_io_watchers_by_fdhubWaiter_waiterr0rL add_handlerr<READ)r gevent_hubrNrr;rrs   z_GeventSelectorIOLoop.__init__cCs|jjd|_dS)zRelease the loop's resources.N)rOloopdestroyr;rrrcloses  z_GeventSelectorIOLoop.closecCs,td|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)rFdebugrSgetclearr;rrrstarts   z_GeventSelectorIOLoop.startcCs|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)rSswitchr;rrrstops z_GeventSelectorIOLoop.stopcCsVt|jkrtd|jj|dStdt|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) rrrOrFrZrW run_callback functoolspartialrLrBr@rrr add_callbacks   z"_GeventSelectorIOLoop.add_callbackcCs|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 )rOrWtimerr])rdelayrArdrrr call_laters z _GeventSelectorIOLoop.call_latercCs |dS)zURemove a timeout :param timeout_handle: Handle of timeout to remove N)rY)rtimeout_handlerrrremove_timeouts z$_GeventSelectorIOLoop.remove_timeoutcCs,|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)rOrWiorPr])rr<handlerrM io_watcherrrrrTs z!_GeventSelectorIOLoop.add_handlercCs2|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)rPrArYrT)rr<rMrkrArrrupdate_handler!s z$_GeventSelectorIOLoop.update_handlercCs|j|}||j|=dS)zgStop watching the given file descriptor for events :param int fd: The file descriptor N)rPrY)rr<rkrrrremove_handler/s  z$_GeventSelectorIOLoop.remove_handler)N)r*r+r,r-rUWRITEERRORrrYr]r_rcrfrhrTrlrmrrrrrs    rc@s"eZdZdZ    dddZdS)rzESelectorIOServicesAdapter implementation using Gevent's DNS resolver.rc Cs*t|j|||||||d}|t|S)zOImplement :py:meth:`.nbio_interface.AbstractIOServices.getaddrinfo()`. ) native_loophostportfamilysocktypeprotoflagsr')_GeventAddressResolver_loopr]_GeventIOLoopIOHandle) rrqrrr'rsrtrurvresolverrrr getaddrinfo<s z,_GeventSelectorIOServicesAdapter.getaddrinfoN)rrrr)r*r+r,r-r{rrrrr9src@s eZdZdZddZddZdS)ryzXImplement `AbstractIOReference`. Only used to wrap the _GeventAddressResolver. cCs |j|_dS)zY :param subject: subject of the reference containing a `cancel()` method N)cancel_cancel)rsubjectrrrrYs z_GeventIOLoopIOHandle.__init__cCs|S)zCancel pending operation :returns: False if was already done or cancelled; True otherwise :rtype: bool )r}r;rrrr|_sz_GeventIOLoopIOHandle.cancelN)r*r+r,r-rr|rrrrrySs ryc@sLeZdZdZdZddZddZddZd d Zd d Z d dZ ddZ dS)rwzPerforms 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 ) rx_on_done _greenlet_ga_host_ga_port _ga_family _ga_socktype _ga_proto _ga_flagsc CsDt|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) rrxrrrrrrrr) rrprqrrrsrtrurvr'rrrrzs  z_GeventAddressResolver.__init__cCs*|jdurt|j|_dStddS)z-Start an asynchronous getaddrinfo invocation.Nz&_GeventAddressResolver already started)rr spawn_raw_resolverFrGr;rrrr]s z_GeventAddressResolver.startcCs&d}|jdur d}|||S)zCancel the pending resolver.FNT)r_stop_greenlet_cleanup)rchangedrrrr|s  z_GeventAddressResolver.cancelcCs|d|_d|_dS)z,Stop the resolver and release any resources.N)rrxrr;rrrrs z_GeventAddressResolver._cleanupcCs$|jdurt|jd|_dSdS)zbStop the greenlet performing getaddrinfo if running. Otherwise, this is a no-op. N)rrkillr;rrrrs   z%_GeventAddressResolver._stop_greenletc Cszztj|j|j|j|j|j|j}Wnt y-}zt d||}WYd}~nd}~wwt |j|}|j|dS)zoCall `getaddrinfo()` and return result via user's callback function on the configured IO loop. zAddress resolution failed: %rN)rsocketr{rrrrrr ExceptionrFerrorrarb_dispatch_callbackrxrc)rresultexcrArrrrs z_GeventAddressResolver._resolvecCs2ztd|j||W|dS|w)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)rFrZrrr)rrrrrrs z)_GeventAddressResolver._dispatch_callbackN) r*r+r,r- __slots__rr]r|rrrrrrrrrwhs   rw)"r-raloggingr4r8weakrefr1 ImportErrorr2r 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*rFrobjectr0rrryrwrrrrs6      V5