Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Base class to manage comms"""
3# Copyright (c) IPython Development Team.
4# Distributed under the terms of the Modified BSD License.
6import logging
8import comm.base_comm
9import traitlets
10import traitlets.config
12from .comm import Comm
14logger = logging.getLogger("ipykernel.comm")
17class CommManager(comm.base_comm.CommManager, traitlets.config.LoggingConfigurable): # type:ignore[misc]
18 """A comm manager."""
20 kernel = traitlets.Instance("ipykernel.kernelbase.Kernel")
21 comms = traitlets.Dict()
22 targets = traitlets.Dict()
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)
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
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 )