/usr/lib/python3/dist-packages/ipykernel/inprocess/channels.py is in python3-ipykernel 4.8.2-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """A kernel client for in-process kernels."""
# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.
from jupyter_client.channelsabc import HBChannelABC
from .socket import DummySocket
#-----------------------------------------------------------------------------
# Channel classes
#-----------------------------------------------------------------------------
class InProcessChannel(object):
"""Base class for in-process channels."""
proxy_methods = []
def __init__(self, client=None):
super(InProcessChannel, self).__init__()
self.client = client
self._is_alive = False
def is_alive(self):
return self._is_alive
def start(self):
self._is_alive = True
def stop(self):
self._is_alive = False
def call_handlers(self, msg):
""" This method is called in the main thread when a message arrives.
Subclasses should override this method to handle incoming messages.
"""
raise NotImplementedError('call_handlers must be defined in a subclass.')
def flush(self, timeout=1.0):
pass
def call_handlers_later(self, *args, **kwds):
""" Call the message handlers later.
The default implementation just calls the handlers immediately, but this
method exists so that GUI toolkits can defer calling the handlers until
after the event loop has run, as expected by GUI frontends.
"""
self.call_handlers(*args, **kwds)
def process_events(self):
""" Process any pending GUI events.
This method will be never be called from a frontend without an event
loop (e.g., a terminal frontend).
"""
raise NotImplementedError
class InProcessHBChannel(object):
"""A dummy heartbeat channel interface for in-process kernels.
Normally we use the heartbeat to check that the kernel process is alive.
When the kernel is in-process, that doesn't make sense, but clients still
expect this interface.
"""
time_to_dead = 3.0
def __init__(self, client=None):
super(InProcessHBChannel, self).__init__()
self.client = client
self._is_alive = False
self._pause = True
def is_alive(self):
return self._is_alive
def start(self):
self._is_alive = True
def stop(self):
self._is_alive = False
def pause(self):
self._pause = True
def unpause(self):
self._pause = False
def is_beating(self):
return not self._pause
HBChannelABC.register(InProcessHBChannel)
|