Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Add PipeIOStream

  • Loading branch information...
commit 6449189d4e43f5b7aa3c39a4e66255f50cf87ef6 1 parent e88f02e
@bdarnell bdarnell authored
View
35 tornado/iostream.py
@@ -21,6 +21,7 @@
* `BaseIOStream`: Generic interface for reading and writing.
* `IOStream`: Implementation of BaseIOStream using non-blocking sockets.
* `SSLIOStream`: SSL-aware version of IOStream.
+* `PipeIOStream`: Pipe-based IOStream implementation.
"""
from __future__ import absolute_import, division, with_statement
@@ -796,6 +797,40 @@ def read_from_fd(self):
return None
return chunk
+class PipeIOStream(BaseIOStream):
+ """Pipe-based IOStream implementation.
+
+ The constructor takes an integer file descriptor (such as one returned
+ by `os.pipe`) rather than an open file object.
+ """
+ def __init__(self, fd, *args, **kwargs):
+ from tornado.platform.posix import _set_nonblocking
+ self.fd = fd
+ _set_nonblocking(fd)
+ super(PipeIOStream, self).__init__(*args, **kwargs)
+
+ def fileno(self):
+ return self.fd
+
+ def close_fd(self):
+ os.close(self.fd)
+
+ def write_to_fd(self, data):
+ return os.write(self.fd, data)
+
+ def read_from_fd(self):
+ try:
+ chunk = os.read(self.fd, self.read_chunk_size)
+ except (IOError, OSError), e:
+ if e.args[0] in (errno.EWOULDBLOCK, errno.EAGAIN):
+ return None
+ else:
+ raise
+ if not chunk:
+ self.close()
+ return None
+ return chunk
+
def _double_prefix(deque):
"""Grow by doubling, but don't split the second chunk just because the
View
28 tornado/test/iostream_test.py
@@ -1,7 +1,7 @@
from __future__ import absolute_import, division, with_statement
from tornado import netutil
from tornado.ioloop import IOLoop
-from tornado.iostream import IOStream, SSLIOStream
+from tornado.iostream import IOStream, SSLIOStream, PipeIOStream
from tornado.log import gen_log
from tornado.testing import AsyncHTTPTestCase, AsyncHTTPSTestCase, AsyncTestCase, bind_unused_port, ExpectLog
from tornado.test.util import unittest
@@ -389,3 +389,29 @@ def _make_server_iostream(self, connection, **kwargs):
def _make_client_iostream(self, connection, **kwargs):
return SSLIOStream(connection, io_loop=self.io_loop, **kwargs)
TestIOStreamSSL = skipIfNoSSL(TestIOStreamSSL)
+
+class TestPipeIOStream(AsyncTestCase):
+ def test_pipe_iostream(self):
+ r, w = os.pipe()
+
+ rs = PipeIOStream(r, io_loop=self.io_loop)
+ ws = PipeIOStream(w, io_loop=self.io_loop)
+
+ ws.write(b("hel"))
+ ws.write(b("lo world"))
+
+ rs.read_until(b(' '), callback=self.stop)
+ data = self.wait()
+ self.assertEqual(data, b("hello "))
+
+ rs.read_bytes(3, self.stop)
+ data = self.wait()
+ self.assertEqual(data, b("wor"))
+
+ ws.close()
+
+ rs.read_until_close(self.stop)
+ data = self.wait()
+ self.assertEqual(data, b("ld"))
+
+ rs.close()
View
3  website/sphinx/iostream.rst
@@ -39,3 +39,6 @@
.. autoclass:: SSLIOStream
:members:
+
+ .. autoclass:: PipeIOStream
+ :members:
Please sign in to comment.
Something went wrong with that request. Please try again.