o NDi( @sFdZddlZddlZddlZddlmZddlmZmZddl Z ddl m Z m Z ddl mZmZmZddlmZddlmZdd lmZdd lmZmZmZmZmZz dd lmZd ZWne yrZ!z d Ze!Z"WYdZ![!ndZ![!wwe#e$Z%ddZ&Gddde j'Z(Gddde)Z*eeGddde)Z+GdddeZ,dS)zV Module that implements an event loop based on twisted ( https://twistedmatrix.com ). N)partial)ThreadLock)reactorprotocol)connectProtocolTCP4ClientEndpointSSL4ClientEndpoint)IOpenSSLClientConnectionCreator)Failure) implementer) ConnectionConnectionShutdownTimer TimerManagerConnectionException)SSLTFcCs&z|WdStyYdSwN)_cleanupReferenceError)cleanup_weakrefrT/var/www/Datamplify/venv/lib/python3.10/site-packages/cassandra/io/twistedreactor.pyr*s  rc@s0eZdZdZddZddZddZdd Zd S) TwistedConnectionProtocolz[ Twisted Protocol class for handling data received and connection made events. cCs ||_dSr) connection)selfrrrr__init__7s z"TwistedConnectionProtocol.__init__cCs|jj||jdS)z Callback function that is called when data has been received on the connection. Reaches back to the Connection object and queues the data for processing. N)r_iobufwrite handle_readrdatarrr dataReceived:sz&TwistedConnectionProtocol.dataReceivedcCs|j|jdS)z Callback function that is called when a connection has succeeded. Reaches back to the Connection object and confirms that the connection is ready. N)rclient_connection_made transportrrrrconnectionMadeEsz(TwistedConnectionProtocol.connectionMadecCstd||j|jdS)NzConnect lost: %s)logdebugrdefunctvalue)rreasonrrrconnectionLostNs z(TwistedConnectionProtocol.connectionLostN)__name__ __module__ __qualname____doc__rr"r&r,rrrrr1s  rc@sTeZdZdZdZdZdZddZddZddZ dd Z d d Z d d Z ddZ dS) TwistedLoopNcCst|_t|_dSr)r_lockr_timersr%rrrr[s zTwistedLoop.__init__cCs|j3tjs.ttjdddid|_d|j_|jt t t t |WddSWddS1s9wYdS)N#cassandra_driver_twisted_event_loopinstallSignalHandlersF)targetnamekwargsT)r2rrunningrrun_threaddaemonstartatexitregisterrrweakrefrefr%rrr maybe_start_s "zTwistedLoop.maybe_startcCstjSr)r_stoppedr%rrr_reactor_stoppediszTwistedLoop._reactor_stoppedcCsF|jr!ttj|jjdd|jrtdtddSdS)Ng?timeoutzrEvent loop thread could not be joined, so shutdown may not be clean. Please call Cluster.shutdown() to avoid this.zEvent loop thread was joined) r;rcallFromThreadstopjoinis_aliver'warningr(r%rrrrls   zTwistedLoop._cleanupcCs |j|t|j|jdSr)r3 add_timerrrG_schedule_timeoutend)rtimerrrrrLvs zTwistedLoop.add_timercCsh|r2t|td}|jr%|jr%||jkr#|j|||_dSdSt||j|_||_dSdS)Nr) maxtime _timeout_taskactive_timeoutresetr callLater_on_loop_timer)r next_timeoutdelayrrrrM|s    zTwistedLoop._schedule_timeoutcCs|j||jjdSr)r3service_timeoutsrMrXr%rrrrWs zTwistedLoop._on_loop_timer)r-r.r/r2r;rRrTrrBrDrrLrMrWrrrrr1Ts   r1c@s,eZdZddZddZddZddZd S) _SSLCreatorcCs||_||_||_||_|r||_nBttj|_d|jvr'|j|jdd|jvr5|j |jdd|jvrC|j |jdd|jvrT|jj |jd|j d|j |jdS)Ncertfilekeyfileca_certs cert_reqs)callback)endpoint ssl_optionscheck_hostnamerFcontextrContext TLSv1_METHODuse_certificate_fileuse_privatekey_fileload_verify_locations set_verifyverify_callbackset_info_callback info_callback)rra ssl_contextrbrcrFrrrrs&    z_SSLCreator.__init__cCs|Srr)rrx509errnumerrdepthokrrrrksz_SSLCreator.verify_callbackcCsT|tj@r$|jr&|jj|jkr(|}| t t d|jdSdSdSdS)NzHostname verification failed) rSSL_CB_HANDSHAKE_DONErcraaddressget_peer_certificate get_subject commonName get_app_datafailVerificationr r)rrwhereretr$rrrrms z_SSLCreator.info_callbackcCsBt|jd}|||jrd|jvr||jdd|S)Nserver_hostnameascii)rr rd set_app_datarbset_tlsext_host_nameencode)r tlsProtocolrrrrclientConnectionForTLSs  z"_SSLCreator.clientConnectionForTLSN)r-r.r/rrkrmrrrrrr[s  r[c@sdeZdZdZdZeddZeddZddZd d Z d d Z d dZ ddZ ddZ ddZdS)TwistedConnectionz] An implementation of :class:`.Connection` that utilizes the Twisted event loop. NcCs|js t|_dSdSr)_loopr1)clsrrrinitialize_reactors z$TwistedConnection.initialize_reactorcCst||}|j||Sr)rrrL)rrFr`rOrrr create_timers  zTwistedConnection.create_timercOsDtj|g|Ri|d|_d|_d|_t|j|j dS)a Initialization method. Note that we can't call reactor methods directly here because it's not thread-safe, so we schedule the reactor/connection stuff to be run from the event loop thread when it gets the chance. TN) r r is_closed connectorr$rrGadd_connectionrrB)rargsr8rrrrs   zTwistedConnection.__init__cCs(|js|jrtstttddSdS)NzO, pyOpenSSL must be installed to enable SSL support with the Twisted event loop)rnrb_HAS_SSL ImportErrorstrimport_exceptionr%rrr_check_pyopenssls z"TwistedConnection._check_pyopensslcCs|j\}}|js |jr.|t|j|jr|jnd|j|j|j}tt ||||jd}n t t |||jd}t |t |dS)z\ Convenience function to connect and store the resulting connector. N)sslContextFactoryrFrE) raresolvernrbrr[_check_hostnameconnect_timeoutr rrrr)rhostportssl_connection_creatorrarrrrs0 z TwistedConnection.add_connectioncCs>|j d|_Wdn1swY||_|dS)z] Called by twisted protocol when a connection attempt has succeeded. FN)lockrr$_send_options_message)rr$rrrr#s  z(TwistedConnection.client_connection_madecCs|j|jr WddSd|_Wdn1swYtdt||jt|jj j td|j|j sO| t d|j|jdSdS)z8 Disconnect and error-out all requests. NTzClosing connection (%s) to %szClosed socket to %szConnection to %s was closed)rrr'r(idrarrGr$r disconnect is_defuncterror_all_requestsrconnected_eventsetr%rrrcloses zTwistedConnection.closecCs |dS)z3 Process the incoming data buffer. N)process_io_bufferr%rrrr#s zTwistedConnection.handle_readcCst|jj|dS)a This function is called when outgoing data should be queued for sending. Note that we can't call transport.write() directly because it is not thread-safe, so we schedule it to run from within the event loop when it gets the chance. N)rrGr$rr rrrpush)s zTwistedConnection.push)r-r.r/r0r classmethodrrrrrr#rrrrrrrrs  #  r)-r0r>loggingrQ functoolsr threadingrrr@twisted.internetrrtwisted.internet.endpointsrrr twisted.internet.interfacesr twisted.python.failurer zope.interfacer cassandra.connectionr rrrrOpenSSLrrrer getLoggerr-r'rProtocolrobjectr1r[rrrrrs8       #8)