/usr/lib/python3/dist-packages/ipykernel/comm/manager.py is in python3-ipykernel 4.5.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 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """Base class to manage comms"""
# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.
import sys
import logging
from traitlets.config import LoggingConfigurable
from ipython_genutils.importstring import import_item
from ipython_genutils.py3compat import string_types
from traitlets import Instance, Unicode, Dict, Any, default
from .comm import Comm
class CommManager(LoggingConfigurable):
"""Manager for Comms in the Kernel"""
kernel = Instance('ipykernel.kernelbase.Kernel')
comms = Dict()
targets = Dict()
# Public APIs
def register_target(self, target_name, f):
"""Register a callable f for a given target name
f will be called with two arguments when a comm_open message is received with `target`:
- the Comm instance
- the `comm_open` message itself.
f can be a Python callable or an import string for one.
"""
if isinstance(f, string_types):
f = import_item(f)
self.targets[target_name] = f
def unregister_target(self, target_name, f):
"""Unregister a callable registered with register_target"""
return self.targets.pop(target_name)
def register_comm(self, comm):
"""Register a new comm"""
comm_id = comm.comm_id
comm.kernel = self.kernel
self.comms[comm_id] = comm
return comm_id
def unregister_comm(self, comm):
"""Unregister a comm, and close its counterpart"""
# unlike get_comm, this should raise a KeyError
comm = self.comms.pop(comm.comm_id)
def get_comm(self, comm_id):
"""Get a comm with a particular id
Returns the comm if found, otherwise None.
This will not raise an error,
it will log messages if the comm cannot be found.
"""
try:
return self.comms[comm_id]
except KeyError:
self.log.warn("No such comm: %s", comm_id)
if self.log.isEnabledFor(logging.DEBUG):
# don't create the list of keys if debug messages aren't enabled
self.log.debug("Current comms: %s", list(self.comms.keys()))
# Message handlers
def comm_open(self, stream, ident, msg):
"""Handler for comm_open messages"""
content = msg['content']
comm_id = content['comm_id']
target_name = content['target_name']
f = self.targets.get(target_name, None)
comm = Comm(comm_id=comm_id,
primary=False,
target_name=target_name,
)
self.register_comm(comm)
if f is None:
self.log.error("No such comm target registered: %s", target_name)
else:
try:
f(comm, msg)
return
except Exception:
self.log.error("Exception opening comm with target: %s", target_name, exc_info=True)
# Failure.
try:
comm.close()
except:
self.log.error("""Could not close comm during `comm_open` failure
clean-up. The comm may not have been opened yet.""", exc_info=True)
def comm_msg(self, stream, ident, msg):
"""Handler for comm_msg messages"""
content = msg['content']
comm_id = content['comm_id']
comm = self.get_comm(comm_id)
if comm is None:
return
try:
comm.handle_msg(msg)
except Exception:
self.log.error('Exception in comm_msg for %s', comm_id, exc_info=True)
def comm_close(self, stream, ident, msg):
"""Handler for comm_close messages"""
content = msg['content']
comm_id = content['comm_id']
comm = self.get_comm(comm_id)
if comm is None:
return
del self.comms[comm_id]
try:
comm.handle_close(msg)
except Exception:
self.log.error('Exception in comm_close for %s', comm_id, exc_info=True)
__all__ = ['CommManager']
|