B 2di%@sJdZddlZddlZddlZddlZddlmZddlmZddlm Z ddlm Z ddlm Z ddlm Z dd lm Z dd lmZddlZddlZdd lmZdd lmZdd lmZddlmZddlmZddlmZddlmZddlmZe ee ee ej fdddZ!Gddde"Z#GdddeZ$GdddeZ%dS)zc Defines a KernelClient that provides thread-safe sockets with async callbacks on message replies. N)Event)Thread)Any) Awaitable)Dict)List)Optional)Union)IOLoop)Instance)Type)ZMQError) zmqstream)Session) KernelClient) HBChannel)msgreturncs |IdHS)N)rrrd/work/yifan.wang/ringdown/master-ringdown-env/lib/python3.7/site-packages/jupyter_client/threaded.pyget_msg"srcseZdZdZdZdZdZdZdZe e j e e e e ddfdd ZdZeddd Zddd d Zddd d ZddddZeeefddddZeeeefddddZeeefddddZddddZde ddddZ!ddddZ"Z#S) ThreadedZMQSocketChannelz.A ZMQ socket invoking a callback in the ioloopN)socketsessionlooprcsVt|_|_|_tfdd}jdk s>tj|dS)a'Create a channel. Parameters ---------- socket : :class:`zmq.Socket` The ZMQ socket to use. session : :class:`session.Session` The session to use. loop A pyzmq ioloop to connect the socket to using a ZMQStream cs:jdk sttjj_jjdS)N) rAssertionErrorrZ ZMQStreamioloopstreamZon_recv _handle_recvsetr)evtselfrr setup_streamGsz7ThreadedZMQSocketChannel.__init__..setup_streamN) super__init__rrrrr add_callbackwait)r"rrrr#) __class__)r!r"rr%/s  z!ThreadedZMQSocketChannel.__init__F)rcCs|jS)N) _is_alive)r"rrris_aliveSsz!ThreadedZMQSocketChannel.is_alivecCs d|_dS)NT)r))r"rrrstartVszThreadedZMQSocketChannel.startcCs d|_dS)NF)r))r"rrrstopYszThreadedZMQSocketChannel.stopcCs<|jdk r8y|jjddWntk r0YnXd|_dS)Nr)Zlinger)rclose Exception)r"rrrr-\s  zThreadedZMQSocketChannel.close)rrcs,fdd}jdk stj|dS)zQueue a message to be sent from the IOLoop's thread. Parameters ---------- msg : message to send This is threadsafe, as it uses IOLoop.add_callback to give the loop's thread control of the action. cs"jdk stjjdS)N)rrsendrr)rr"rr thread_sendosz2ThreadedZMQSocketChannel.send..thread_sendN)rrr&)r"rr0r)rr"rr/ds zThreadedZMQSocketChannel.sendcCs|t|r0|jdk st|jj}|t|}n|}|jdk sBt|j|\}}|j |}|j rn| || |dS)z[Callback for stream.on_recv. Unpacks message, and calls handlers with it. N) asyncioisfuturerr_asyncio_event_looprun_until_completerrZfeed_identitiesZ deserialize_inspect call_handlers)r"rrZmsg_listidentZsmsgZnew_msgrrrrvs   z%ThreadedZMQSocketChannel._handle_recvcCsdS)aiThis method is called in the ioloop thread when a message arrives. Subclasses should override this method to handle incoming messages. It is important to remember that this method is called in the thread so that some logic must be done to ensure that the application level handlers are called in the application thread. Nr)r"rrrrr6sz&ThreadedZMQSocketChannel.call_handlerscCsdS)zaSubclasses should override this with a method processing any pending GUI events. Nr)r"rrrprocess_eventssz'ThreadedZMQSocketChannel.process_events?)timeoutrcCsft|}|jdk stxFtdD]:}d|_|j|jx |js\t|kr\tdq>Wq$WdS)aImmediately processes all pending messages on this channel. This is only used for the IOPub channel. Callers should use this method to ensure that :meth:`call_handlers` has been called for all messages that have been received on the 0MQ SUB socket of this channel. This method is thread safe. Parameters ---------- timeout : float, optional The maximum amount of time to spend flushing, in seconds. The default is one second. NFg{Gz?)timerrrange_flushedr&_flushsleep)r"r: stop_time_rrrflushs zThreadedZMQSocketChannel.flushcCs"|jdk st|jd|_dS)z"Callback for :method:`self.flush`.NT)rrrCr>)r"rrrr?s zThreadedZMQSocketChannel._flush)r9)$__name__ __module__ __qualname____doc__rrrrr5rzmqZSocketrr r%r)boolr*r+r,r-rstrrr/r rbytesrrr6r8floatrCr? __classcell__rr)r(rr&s* rcseZdZdZdZdZfddZeej ddddZ ddd d Z ddd d Z ddd dZ ddZddddZZS) IOLoopThreadz;Run a pyzmq ioloop in a thread to send and receive messagesFNcstd|_dS)NT)r$r%daemon)r")r(rrr%s zIOLoopThread.__init__)rcCstdk rdt_dS)NT)rN_exitingrrrr _notice_exitszIOLoopThread._notice_exitcCs t|_t||jdS)z{Start the IOLoop thread Don't return until self.ioloop is defined, which is created in the thread N)r _start_eventrr+r')r"rrrr+s zIOLoopThread.startc Cst}t|t|t|_||j_|j xhy|j WnRt k rz}z|j t j krhw8nWdd}~XYq8tk r|jrPnYq8XPq8WdS)z0Run my loop, ignoring EINTR events in the pollerN)r1new_event_loopset_event_loop nest_asyncioapplyr rr3rRr r+r errnoZEINTRr.rP)r"rerrrruns$    zIOLoopThread.runcCs4|jdk r|j|jj||d|_dS)zStop the channel's event loop and join its thread. This calls :meth:`~threading.Thread.join` and returns when the thread terminates. :class:`RuntimeError` will be raised if :meth:`~threading.Thread.start` is called again. N)rr&r,joinr-)r"rrrr,s  zIOLoopThread.stopcCs |dS)N)r-)r"rrr__del__szIOLoopThread.__del__cCs6|jdk r2y|jjddWntk r0YnXdS)NT)Zall_fds)rr-r.)r"rrrr-s  zIOLoopThread.close)rDrErFrGrPrr% staticmethodatexitregisterrQr+rYr,r[r-rMrr)r(rrNs   rNcseZdZdZeddZeeddZde e e e e ddfdd Z e e e fdd d d Zdd fdd ZeeZeeZeeZeeZeeZe d ddZZS)ThreadedKernelClientzYA KernelClient that provides thread-safe sockets with async callbacks on message replies.cCs|jjS)N) ioloop_threadr)r"rrrr szThreadedKernelClient.ioloopT) allow_noneN)shelliopubstdinhbcontrolrcs8t|_|j|r |j|j_t|||||dS)N)rNr`r+_check_kernel_info_reply shell_channelr5r$start_channels)r"rbrcrdrerf)r(rrris   z#ThreadedKernelClient.start_channels)rrcCs"|ddkr||d|j_dS)zGThis is run in the ioloop thread when the kernel info reply is receivedZmsg_typeZkernel_info_replyN)Z_handle_kernel_info_replyrhr5)r"rrrrrg!s  z-ThreadedKernelClient._check_kernel_info_reply)rcs"t|jr|jdS)N)r$ stop_channelsr`r*r,)r")r(rrrj's  z"ThreadedKernelClient.stop_channelscCs|jdk r|jSdS)z$Is the kernel process still running?NT)Z _hb_channelZ is_beating)r"rrrr*2s  zThreadedKernelClient.is_alive)TTTTT)rDrErFrGpropertyrr rNr`rIrirrJrrgrjr rZiopub_channel_classZshell_channel_classZstdin_channel_classrZhb_channel_classZcontrol_channel_classr*rMrr)r(rr_s"    r_)&rGr1r]rWr< threadingrrtypingrrrrrr rUrHZtornado.ioloopr Z traitletsr r r Z zmq.eventlooprrrZjupyter_clientrZjupyter_client.channelsrrKMessagerobjectrrNr_rrrrs6                "M