##// END OF EJS Templates
Starting work on a Markdown cell.
Starting work on a Markdown cell.

File last commit:

r4348:97f1ce7d
r4507:f5f61980
Show More
sessionmanager.py
90 lines | 3.0 KiB | text/x-python | PythonLexer
"""A manager for session and channels for a single kernel."""
#-----------------------------------------------------------------------------
# Copyright (C) 2011 The IPython Development Team
#
# Distributed under the terms of the BSD License. The full license is in
# the file COPYING.txt, distributed as part of this software.
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
import zmq
from zmq.eventloop.zmqstream import ZMQStream
from IPython.utils.traitlets import Instance, Dict, CBytes, Bool
from IPython.zmq.session import SessionFactory
class SessionManagerRunningError(Exception):
pass
#-----------------------------------------------------------------------------
# Classes
#-----------------------------------------------------------------------------
class SessionManager(SessionFactory):
"""Manages a session for a kernel.
The object manages a variety of things for a connection session to
a running kernel:
* The set of channels or connected ZMQ streams to the kernel.
* An IPython.zmq.session.Session object that manages send/recv logic
for those channels.
"""
kernel_manager = Instance('IPython.frontend.html.notebook.kernelmanager.KernelManager')
kernel_id = CBytes(b'')
_session_streams = Dict()
_running = Bool(False)
def __init__(self, **kwargs):
kernel_id = kwargs.pop('kernel_id')
super(SessionManager, self).__init__(**kwargs)
self.kernel_id = kernel_id
self.start()
def __del__(self):
self.stop()
def start(self):
if not self._running:
ports = self.kernel_manager.get_kernel_ports(self.kernel_id)
iopub_stream = self.create_connected_stream(ports['iopub_port'], zmq.SUB)
iopub_stream.socket.setsockopt(zmq.SUBSCRIBE, b'')
shell_stream = self.create_connected_stream(ports['shell_port'], zmq.XREQ)
self._session_streams = dict(
iopub_stream = iopub_stream,
shell_stream = shell_stream
)
self._running = True
else:
raise SessionManagerRunningError(
'Session manager is already running, call stop() before start()'
)
def stop(self):
if self._running:
for name, stream in self._session_streams.items():
stream.close()
self._session_streams = {}
self._running = False
def create_connected_stream(self, port, socket_type):
sock = self.context.socket(socket_type)
addr = "tcp://%s:%i" % (self.kernel_manager.get_kernel_ip(self.kernel_id), port)
self.log.info("Connecting to: %s" % addr)
sock.connect(addr)
return ZMQStream(sock)
def get_iopub_stream(self):
return self._session_streams['iopub_stream']
def get_shell_stream(self):
return self._session_streams['shell_stream']