1  """Base class to manage comms"""
   2  
   3  # Copyright (c) IPython Development Team.
   4  # Distributed under the terms of the Modified BSD License.
   5  
   6  import logging
   7  
   8  import comm.base_comm
   9  import traitlets
  10  import traitlets.config
  11  
  12  from .comm import Comm
  13  
  14  logger = logging.getLogger("ipykernel.comm")
  15  
  16  
  17  class CommManager(comm.base_comm.CommManager, traitlets.config.LoggingConfigurable):  # type:ignore[misc]
  18      """A comm manager."""
  19  
  20      kernel = traitlets.Instance("ipykernel.kernelbase.Kernel")
  21      comms = traitlets.Dict()
  22      targets = traitlets.Dict()
  23  
  24      def __init__(self, **kwargs):
  25          """Initialize the manager."""
  26          # CommManager doesn't take arguments, so we explicitly forward arguments
  27          comm.base_comm.CommManager.__init__(self)
  28          traitlets.config.LoggingConfigurable.__init__(self, **kwargs)
  29  
  30      def comm_open(self, stream, ident, msg):
  31          """Handler for comm_open messages"""
  32          # This is for backward compatibility, the comm_open creates a a new ipykernel.comm.Comm
  33          # but we should let the base class create the comm with comm.create_comm in a major release
  34          content = msg["content"]
  35          comm_id = content["comm_id"]
  36          target_name = content["target_name"]
  37          f = self.targets.get(target_name, None)
  38          comm = Comm(
  39              comm_id=comm_id,
  40              primary=False,
  41              target_name=target_name,
  42              show_warning=False,
  43          )
  44          self.register_comm(comm)
  45          if f is None:
  46              logger.error("No such comm target registered: %s", target_name)
  47          else:
  48              try:
  49                  f(comm, msg)
  50                  return
  51              except Exception:
  52                  logger.error("Exception opening comm with target: %s", target_name, exc_info=True)  # noqa: G201
  53  
  54          # Failure.
  55          try:
  56              comm.close()
  57          except Exception:
  58              logger.error(  # noqa: G201
  59                  """Could not close comm during `comm_open` failure
  60                  clean-up.  The comm may not have been opened yet.""",
  61                  exc_info=True,
  62              )
  63