Add PipeIOStream

diff --git a/tornado/iostream.py b/tornado/iostream.py
index 94f00f9660cd9379244f17ecdbb72e6863147768..2f716a728cb2729046abed92c7d121c3f1be0665 100644 (file)
--- a/tornado/iostream.py
+++ b/tornado/iostream.py
@@ -21,6 +21,7 @@ Contents:
 * `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 @@ class SSLIOStream(IOStream):
             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

diff --git a/tornado/test/iostream_test.py b/tornado/test/iostream_test.py
index d06c69b3a33db128a1dfc722fdf763b6336bffc5..9f5b5cf09ccce2624047605a353bf911a8b08fe3 100644 (file)
--- a/tornado/test/iostream_test.py
+++ b/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 @@ class TestIOStreamSSL(TestIOStreamMixin, AsyncTestCase):
     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()

diff --git a/website/sphinx/iostream.rst b/website/sphinx/iostream.rst
index d37a3793c78675255a1b4e5924d5497b91c04675..85b4d28aa957001706239ab565c1029922536f8c 100644 (file)
--- a/website/sphinx/iostream.rst
+++ b/website/sphinx/iostream.rst
@@ -39,3 +39,6 @@
 
    .. autoclass:: SSLIOStream
       :members:
+
+   .. autoclass:: PipeIOStream
+      :members: