B \@sdZddlmZddlZddlZddlZddlmZmZddl Z ddl m Z ddl m Z mZddlmZmZddlmZdd lmZGd d d eZGd d d eZGdddeZdS)zc Defines a KernelClient that provides thread-safe sockets with async callbacks on message replies. )absolute_importN)ThreadEvent)ZMQError)ioloop zmqstream)TypeInstance) HBChannel) KernelClientcseZdZdZdZdZdZdZdZfddZ dZ ddZ dd Z d d Z d d ZddZddZddZddZdddZddZZS)ThreadedZMQSocketChannelz.A ZMQ socket invoking a callback in the ioloopNcsLtt|_|_|_tfdd}j|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,tjj_jjdS)N)rZ ZMQStreamsocketrstreamZon_recv _handle_recvset)evtselfr6lib/python3.7/site-packages/jupyter_client/threaded.py setup_stream/sz7ThreadedZMQSocketChannel.__init__..setup_streamN) superr __init__r sessionrr add_callbackwait)rr rZloopr) __class__)rrrrs  z!ThreadedZMQSocketChannel.__init__FcCs|jS)N) _is_alive)rrrris_alive8sz!ThreadedZMQSocketChannel.is_alivecCs d|_dS)NT)r)rrrrstart;szThreadedZMQSocketChannel.startcCs d|_dS)NF)r)rrrrstop>szThreadedZMQSocketChannel.stopcCs<|jdk r8y|jjddWntk r0YnXd|_dS)Nr)Zlinger)r close Exception)rrrrr As  zThreadedZMQSocketChannel.closecsfdd}j|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. csjjdS)N)rsendrr)msgrrr thread_sendSsz2ThreadedZMQSocketChannel.send..thread_sendN)rr)rr#r$r)r#rrr"Is zThreadedZMQSocketChannel.sendcCs:|j|\}}|j|}|jr,||||dS)z[Callback for stream.on_recv. Unpacks message, and calls handlers with it. N)rZfeed_identitiesZ deserialize_inspect call_handlers)rr#ZidentZsmsgrrrrWs   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)rr#rrrr&csz&ThreadedZMQSocketChannel.call_handlerscCsdS)zaSubclasses should override this with a method processing any pending GUI events. Nr)rrrrprocess_eventsmsz'ThreadedZMQSocketChannel.process_events?cCsXt|}xFtdD]:}d|_|j|jx |jsNt|krNtdq0WqWdS)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. Fg{Gz?N)timerange_flushedrr_flushZsleep)rZtimeoutZ stop_timeirrrflushts  zThreadedZMQSocketChannel.flushcCs|jd|_dS)z"Callback for :method:`self.flush`.TN)rr/r,)rrrrr-s zThreadedZMQSocketChannel._flush)r()__name__ __module__ __qualname____doc__rr rrr%rrrrrr r"rr&r'r/r- __classcell__rr)rrr s$    r csZeZdZdZdZdZfddZeej ddZ dd Z d d Z d d Z ddZZS) IOLoopThreadz@Run a pyzmq ioloop in a thread to send and receive messages FNcstt|d|_dS)NT)rr5rZdaemon)r)rrrrszIOLoopThread.__init__cCstdk rdt_dS)NT)r5_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_eventrrr)rrrrrs zIOLoopThread.startc Csdtjkr ddl}||t|_|jxhy|j WnRt k rx}z|j t j krfw6nWdd}~XYq6t k r|jrPnYq6XPq6WdS)z0Run my loop, ignoring EINTR events in the pollerasynciorN)sysmodulesr9Zset_event_loopZnew_event_looprZIOLoopr8rrrerrnoZEINTRr!r6)rr9errrruns"    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)rrrjoinr )rrrrrs  zIOLoopThread.stopcCs6|jdk r2y|jjddWntk r0YnXdS)NT)Zall_fds)rr r!)rrrrr s  zIOLoopThread.close)r0r1r2r3r6rr staticmethodatexitregisterr7rr>rr r4rr)rrr5s   r5csneZdZdZeddZeeddZd fdd Z dd Z fd d Z e e Ze e Ze e Ze eZZS) ThreadedKernelClientz_ A KernelClient that provides thread-safe sockets with async callbacks on message replies. cCs|jjS)N) ioloop_threadr)rrrrrszThreadedKernelClient.ioloopT)Z allow_nonecs:t|_|j|r |j|j_tt|||||dS)N) r5rDr_check_kernel_info_reply shell_channelr%rrCstart_channels)rshellZiopubstdinZhb)rrrrGs   z#ThreadedKernelClient.start_channelscCs"|ddkr||d|j_dS)zPThis is run in the ioloop thread when the kernel info reply is received Zmsg_typeZkernel_info_replyN)Z_handle_kernel_info_replyrFr%)rr#rrrrEs  z-ThreadedKernelClient._check_kernel_info_replycs&tt||jr"|jdS)N)rrC stop_channelsrDrr)r)rrrrJs z"ThreadedKernelClient.stop_channels)TTTT)r0r1r2r3propertyrr r5rDrGrErJrr Ziopub_channel_classZshell_channel_classZstdin_channel_classr Zhb_channel_classr4rr)rrrCs    rC)r3Z __future__rrAr<r:Z threadingrrr*ZzmqrZ zmq.eventlooprrZ traitletsrr Zjupyter_client.channelsr Zjupyter_clientr objectr r5rCrrrrs    K