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
|
"""Blocking channels
Useful for test suites and blocking terminal interfaces.
"""
#-----------------------------------------------------------------------------
# Copyright (C) 2013 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
#-----------------------------------------------------------------------------
try:
from queue import Queue, Empty # Py 3
except ImportError:
from Queue import Queue, Empty # Py 2
from IPython.kernel.channels import IOPubChannel, HBChannel, \
ShellChannel, StdInChannel
#-----------------------------------------------------------------------------
# Blocking kernel manager
#-----------------------------------------------------------------------------
class BlockingChannelMixin(object):
def __init__(self, *args, **kwds):
super(BlockingChannelMixin, self).__init__(*args, **kwds)
self._in_queue = Queue()
def call_handlers(self, msg):
self._in_queue.put(msg)
def get_msg(self, block=True, timeout=None):
""" Gets a message if there is one that is ready. """
if timeout is None:
# Queue.get(timeout=None) has stupid uninteruptible
# behavior, so wait for a week instead
timeout = 604800
return self._in_queue.get(block, timeout)
def get_msgs(self):
""" Get all messages that are currently ready. """
msgs = []
while True:
try:
msgs.append(self.get_msg(block=False))
except Empty:
break
return msgs
def msg_ready(self):
""" Is there a message that has been received? """
return not self._in_queue.empty()
class BlockingIOPubChannel(BlockingChannelMixin, IOPubChannel):
pass
class BlockingShellChannel(BlockingChannelMixin, ShellChannel):
pass
class BlockingStdInChannel(BlockingChannelMixin, StdInChannel):
pass
class BlockingHBChannel(HBChannel):
# This kernel needs quicker monitoring, shorten to 1 sec.
# less than 0.5s is unreliable, and will get occasional
# false reports of missed beats.
time_to_dead = 1.
def call_handlers(self, since_last_heartbeat):
""" Pause beating on missed heartbeat. """
pass
|
