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