iostream.py
93 lines
| 2.9 KiB
| text/x-python
|
PythonLexer
|
r3294 | import logging | ||
|
r2754 | import sys | ||
import time | ||||
|
r4070 | from io import StringIO | ||
|
r2754 | |||
from session import extract_header, Message | ||||
|
r2938 | from IPython.utils import io | ||
|
r2754 | #----------------------------------------------------------------------------- | ||
|
r3322 | # Globals | ||
|
r2754 | #----------------------------------------------------------------------------- | ||
|
r3322 | |||
# Module-level logger | ||||
|
r3314 | logger = logging.getLogger(__name__) | ||
|
r2754 | |||
|
r3322 | #----------------------------------------------------------------------------- | ||
|
r2754 | # Stream classes | ||
#----------------------------------------------------------------------------- | ||||
class OutStream(object): | ||||
"""A file like object that publishes the stream to a 0MQ PUB socket.""" | ||||
# The time interval between automatic flushes, in seconds. | ||||
flush_interval = 0.05 | ||||
|
r3602 | topic=None | ||
|
r2754 | def __init__(self, session, pub_socket, name): | ||
self.session = session | ||||
self.pub_socket = pub_socket | ||||
self.name = name | ||||
self.parent_header = {} | ||||
self._new_buffer() | ||||
def set_parent(self, parent): | ||||
self.parent_header = extract_header(parent) | ||||
def close(self): | ||||
self.pub_socket = None | ||||
def flush(self): | ||||
|
r2938 | #io.rprint('>>>flushing output buffer: %s<<<' % self.name) # dbg | ||
|
r2754 | if self.pub_socket is None: | ||
raise ValueError(u'I/O operation on closed file') | ||||
else: | ||||
data = self._buffer.getvalue() | ||||
if data: | ||||
content = {u'name':self.name, u'data':data} | ||||
|
r3602 | msg = self.session.send(self.pub_socket, u'stream', content=content, | ||
parent=self.parent_header, ident=self.topic) | ||||
|
r3314 | logger.debug(msg) | ||
|
r3602 | |||
|
r2754 | self._buffer.close() | ||
self._new_buffer() | ||||
def isatty(self): | ||||
return False | ||||
def next(self): | ||||
raise IOError('Read not supported on a write only stream.') | ||||
def read(self, size=-1): | ||||
raise IOError('Read not supported on a write only stream.') | ||||
def readline(self, size=-1): | ||||
raise IOError('Read not supported on a write only stream.') | ||||
def write(self, string): | ||||
if self.pub_socket is None: | ||||
raise ValueError('I/O operation on closed file') | ||||
else: | ||||
|
r4070 | # Make sure that we're handling unicode | ||
if not isinstance(string, unicode): | ||||
enc = sys.stdin.encoding or sys.getdefaultencoding() | ||||
string = string.decode(enc, 'replace') | ||||
|
r2754 | self._buffer.write(string) | ||
current_time = time.time() | ||||
if self._start <= 0: | ||||
self._start = current_time | ||||
elif current_time - self._start > self.flush_interval: | ||||
self.flush() | ||||
def writelines(self, sequence): | ||||
if self.pub_socket is None: | ||||
raise ValueError('I/O operation on closed file') | ||||
else: | ||||
for string in sequence: | ||||
self.write(string) | ||||
def _new_buffer(self): | ||||
self._buffer = StringIO() | ||||
|
r2938 | self._start = -1 | ||