o tBh_@sddlZddlZddlZddlmZddlmZddlmZddl m Z m Z m Z m Z mZmZmZmZddlZddlmZmZddlmZddlmZmZdd lmZdd lmZmZd d l m!Z!d d l"m#Z#m$Z$m%Z%m&Z&m'Z'e(dZ)ee*eefZ+GdddZ,Gddde!Z-dS)N)abstractmethod)suppress) SSLContext)AnyAsyncGeneratorDictListOptionalTupleUnioncast) DocumentNodeExecutionResult)WebSocketClientProtocol)Headers HeadersLike)ConnectionClosed)Data Subprotocol)AsyncTransport)TransportAlreadyConnectedTransportClosedTransportProtocolErrorTransportQueryErrorTransportServerErrorzgql.transport.websocketsc@sXeZdZdZdededdfddZdefdd Zd eddfd d Z d e ddfddZ dS) ListenerQueuea&Special queue used for each query waiting for server answers If the server is stopped while the listener is still waiting, Then we send an exception to the queue and this exception will be raised to the consumer once all the previous messages have been consumed from the queue query_id send_stopreturnNcCs ||_||_t|_d|_dS)NF)rrasyncioQueue_queue_closed)selfrrr%t/var/www/html/riverr-enterprise-integrations-main/venv/lib/python3.10/site-packages/gql/transport/websockets_base.py__init__&s  zListenerQueue.__init__cstz|j}Wntjy|jIdH}Ynw|jt|tr*d|_||\}}|dkr8d|_ d|_|S)NTcompleteF) r" get_nowaitr QueueEmptyget task_done isinstance Exceptionr#r)r$item answer_typeexecution_resultr%r%r&r+,s  zListenerQueue.getr/cs"|js|j|IdHdSdSN)r#r"put)r$r/r%r%r&r3CszListenerQueue.put exceptioncs$|j|IdHd|_d|_dS)NFT)r"r3rr#)r$r4r%r%r& set_exceptionHs zListenerQueue.set_exception) __name__ __module__ __qualname____doc__intboolr' ParsedAnswerr+r3r.r5r%r%r%r&rs rc@sdeZdZdZddiddddifdedeedeee fde ee fd eee e fd eee e fd eee e fd eee e fd e ee fddfddZddZde fddZddZddZddZddZdeddfdd Zdefd!d"Ze  dJd#ed$ee ee fd%eede fd&d'Zed(edeeee eeffd)d*ZdKd+d,ZdKd-d.Zd/ed0ee d1eeddfd2d3Z   4dLd#ed$ee ee fd%eed5ee de edff d6d7Z!  dJd#ed$ee ee fd%eedefd8d9Z"dKd:d;Z#dKde%ddfd?d@Z&dMd>e%dAe ddfdBdCZ'dMd>e%dAe ddfdDdEZ(dKdFdGZ)dKdHdIZ*dS)NWebsocketsTransportBasezabstract :ref:`Async Transport ` used to implement different websockets protocols. This transport uses asyncio and the websockets library in order to send requests on a websocket connection. NF urlheadersssl init_payloadconnect_timeout close_timeout ack_timeoutkeep_alive_timeout connect_argsrc Cs2||_||_||_||_||_||_||_||_| |_d|_ d|_ i|_ d|_ d|_ d|_z"ttjdddt|_Wdn1sIwYWntyct|_t|jYnwt|_|jt|_|j|jdurt|_|ji|_ d|_d|_g|_ d|_!dS)aInitialize the transport with the given parameters. :param url: The GraphQL server URL. Example: 'wss://server.com:PORT/graphql'. :param headers: Dict of HTTP Headers. :param ssl: ssl_context of the connection. Use ssl=False to disable encryption :param init_payload: Dict of the payload sent in the connection_init message. :param connect_timeout: Timeout in seconds for the establishment of the websocket connection. If None is provided this will wait forever. :param close_timeout: Timeout in seconds for the close. If None is provided this will wait forever. :param ack_timeout: Timeout in seconds to wait for the connection_ack message from the server. If None is provided this will wait forever. :param keep_alive_timeout: Optional Timeout in seconds to receive a sign of liveness from the server. :param connect_args: Other parameters forwarded to websockets.connect NrignorezThere is no current event loop)messageF)"r?r@rArBrCrDrErFrG websocket next_query_id listenersreceive_data_taskcheck_keep_alive_task close_taskwarningscatch_warningsfilterwarningsr get_event_loop_loop RuntimeErrornew_event_loopset_event_loopEvent _wait_closedset_no_more_listeners_next_keep_alive_messagepayloads _connectingclose_exceptionsupported_subprotocolsresponse_headers) r$r?r@rArBrCrDrErFrGr%r%r&r'ZsP            z WebsocketsTransportBase.__init__cdS)zxHook to send the initialization messages after the connection and potentially wait for the backend ack. Nr%r$r%r%r& _initializez#WebsocketsTransportBase._initializercrb)ziHook to stop to listen to a specific query. Will send a stop message in some subclasses. Nr%)r$rr%r%r&_stop_listenerrez&WebsocketsTransportBase._stop_listenercrb)zbHook to add custom code for subclasses after the connection has been established. Nr%rcr%r%r&_after_connectrez&WebsocketsTransportBase._after_connectcrbz_Hook to add custom code for subclasses after the initialization has been done. Nr%rcr%r%r&_after_initializerez)WebsocketsTransportBase._after_initializecrb)z?Hook to add custom code for subclasses for the connection closeNr%rcr%r%r& _close_hooksz#WebsocketsTransportBase._close_hookcrbrhr%rcr%r%r&_connection_terminaterez-WebsocketsTransportBase._connection_terminaterIc sj|js td|jz|j|IdHtd|WdSty4}z |j|ddIdH|d}~ww)zISend the provided message to the websocket connection and log the messagezTransport is not connectedNz>>> %sF clean_close)rJrr_sendloginfor_fail)r$rIer%r%r&_sendszWebsocketsTransportBase._sendcsJ|jdur td|jIdH}t|tstd|}td||S)zFWait the next message from the websocket connection and log the answerNzTransport is already closedz%Binary data received in the websocketz<<< %s)rJrrecvr-strrrorp)r$dataanswerr%r%r&_receives   z WebsocketsTransportBase._receivedocumentvariable_valuesoperation_namecstr2NotImplementedError)r$ryrzr{r%r%r& _send_querysz#WebsocketsTransportBase._send_queryrwcCstr2r|)r$rwr%r%r& _parse_answersz%WebsocketsTransportBase._parse_answercs~z t|j|jIdH|jqtjy4|jdur1|jt dddIdHYdSYdStj y>YdSw)zsCoroutine which will periodically check the liveness of the connection through keep-alive messages TNzeNo keep-alive message has been received within the expected interval ('keep_alive_timeout' parameter)Frl) r wait_forr\waitrFclear TimeoutErrorrOrqrCancelledErrorrcr%r%r&_check_ws_livenesss(    z*WebsocketsTransportBase._check_ws_livenessc spz z |IdH}Wn&ttfy*}z|j|ddIdHWYd}~ntd}~wty2Ynpwz ||\}}}WnQtyq}z(t|jt sOJdz|j |j |IdHWn t yfYnwWYd}~qd}~wt tfy}z|j|ddIdHWYd}~n d}~ww||||IdHqWtddSWtddSWtddStdw)zMain asyncio task which will listen to the incoming messages and will call the parse_answer and handle_answer methods of the subclass.TNFrlz7TransportQueryError should have a query_id defined herezExiting _receive_data_loop())rxrrrqrrrr-rr:rLr5KeyErrorr_handle_answerrodebug)r$rwrrr0 answer_idr1r%r%r&_receive_data_loop%sT     #!z*WebsocketsTransportBase._receive_data_loopr0rr1csFz|dur|j|||fIdHWdSWdSty"YdSwr2)rLr3r)r$r0rr1r%r%r&rUs  z&WebsocketsTransportBase._handle_answerTrc Cs6||||IdH}t||dud}||j|<|jzoz# |IdH\}}|dur1|Vn|dkr?td|dnq Wn+tj t fyl} ztd| |j rb| |IdHd|_ WYd} ~ nd} ~ wwWtd || |dSWtd || |dStd || |w) zSend a query and receive the results using a python async generator. The query can be a graphql query, mutation or subscription. The results are sent as an ExecutionResult object. NTrr(zComplete received for query z --> exit without errorzException in subscribe: Fz"In subscribe finally for query_id )r~rrLr[rr+rorr r GeneratorExitrrf_remove_listener) r$ryrzr{rrlistenerr0r1rrr%r%r& subscribedsD      z!WebsocketsTransportBase.subscribecsTd}|j|||dd}|2z3dHW}|}|IdHn6|dur(td|S)aExecute the provided document AST against the configured remote server using the current session. Send a query but close the async generator as soon as we have the first answer. The result is sent as an ExecutionResult object. NFrz;Query completed without any answer received from the server)racloser)r$ryrzr{ first_result generatorresultr%r%r&executes  zWebsocketsTransportBase.executec std|jdur|jsd|_|jr|j}n |jdr dnd}||j|jd}| |j zt t jj|jfi||jIdH|_Wd|_nd|_wtt|j|_|jj|_|IdHd|_d|_|jz |IdHWn(ty}z|d}~wtt jfy}z |j|ddIdH|d}~ww|IdH|jdurt |!|_"t |#|_$nt%d td dS) aOCoroutine which will: - connect to the websocket address - send the init message - wait for the connection acknowledge from the server - create an asyncio task which will be used to receive and parse the websocket answers Should be cleaned with a call to the close coroutine zconnect: startingNTwss)rA extra_headers subprotocolsFrrlzTransport is already connectedz connect: done)&rorrJr^rAr? startswithr@r`updaterGr r websocketsclientconnectrCr rrargrKr_rYrrdrrrrqrirF ensure_futurerrNrrMr)r$rArGrrr%r%r&rsT     zWebsocketsTransportBase.connectcCsN||jvr |j|=t|j}td|d|d|dkr%|jdSdS)zAfter exiting from a subscription, remove the listener and signal an event if this was the last listener for the client. z listener z deleted, z remainingrN)rLlenrorr[rZ)r$r remainingr%r%r&rs  z(WebsocketsTransportBase._remove_listenerrrcs|jD]\}}|jr||IdHd|_qzt|j|jIdHWntj y7t dYnw| IdHdS)zCoroutine which will: - send stop messages for each active subscription to the server - send the connection terminate message NFzTimer close_timeout fired) rLitemsrrfr rr[rrDrrorrk)r$rrrrr%r%r& _clean_close&sz$WebsocketsTransportBase._clean_closermc stdzz|jdusJ|jdur4|jttj|jIdHWdn1s/wY|IdH||_ |rktdz | |IdHWnt yj}zt dt |WYd}~nd}~wwtd|jD] \}}||IdHqutd|jIdHtdWnt y}zt dt |WYd}~nd}~wwWtd d|_d|_d|_|jntd d|_d|_d|_|jwtd dS) a%Coroutine which will: - do a clean_close if possible: - send stop messages for each active query to the server - send the connection terminate message - close the websocket connection - send the exception to all the remaining listeners z_close_coro: startingNz!_close_coro: starting clean_closez$Ignoring exception in _clean_close: z+_close_coro: sending exception to listenersz'_close_coro: close websocket connectionz(_close_coro: websocket connection closedz"Exception catched in _close_coro: z_close_coro: start cleanupz_close_coro: exiting)rorrJrNcancelrr rrjr_rr.warningreprrLrr5closerOrYrZ)r$rrrmexcrrr%r%r& _close_coro=sR           z#WebsocketsTransportBase._close_corocsztdt||jdur+|jdurtddStt|j||d|_dStdt|j dt|dS)Nz _fail: starting with exception: z;_fail started with self.websocket == None -> already closedrlz8close_task is not None in _fail. Previous exception is: z New exception is: ) rorrrOrJr shieldrrr_)r$rrrmr%r%r&rq}s"   zWebsocketsTransportBase._failcs<td|tdIdH|IdHtddS)Nzclose: startingz*Websocket GraphQL transport closed by userz close: done)rorrqr wait_closedrcr%r%r&rs  zWebsocketsTransportBase.closecs*td|jIdHtddS)Nzwait_close: startingzwait_close: done)rorrYrrcr%r%r&rs z#WebsocketsTransportBase.wait_closed)NN)rN)NNT)T)+r6r7r8r9rur rr rr;rrr:floatr'rdrfrgrirjrkrsrxrr r~r rrrrrrrrrrr.rrrqrrr%r%r%r&r=Rs      T   "0   < # Y @ r=).r loggingrPabcr contextlibrrArtypingrrrrr r r r rgraphqlr rwebsockets.clientrwebsockets.datastructuresrrwebsockets.exceptionsrwebsockets.typingrrasync_transportr exceptionsrrrrr getLoggerrorur<rr=r%r%r%r&s&   (    4