Merge pull request #83 from graingert/fix-ipc-connection-close-exception

 fix on_recv called when None after close

pytest.ini

@@ -2,4 +2,5 @@
 python_files = test_*.py
 testpaths =
     src/hupper
-    tests
+    tests
+filterwarnings = error

src/hupper/ipc.py

@@ -1,3 +1,4 @@
+import errno
 import io
 import os
 import pickle
@@ -135,9 +136,12 @@ def activate(self, on_recv):
         self.reader_thread.start()
 
     def close(self):
-        close_fd(self.r_fd)
-        close_fd(self.w_fd)
-        self.on_recv = None
+        self.on_recv = lambda _: None
+        self.r_fd, r_fd = -1, self.r_fd
+        self.w_fd, w_fd = -1, self.w_fd
+
+        close_fd(w_fd)
+        close_fd(r_fd)
 
     def _recv_packet(self):
         buf = io.BytesIO()
@@ -166,6 +170,9 @@ def _read_loop(self):
                 self.on_recv(packet)
         except EOFError:
             pass
+        except OSError as e:
+            if e.errno != errno.EBADF:
+                raise
         self.on_recv(None)
 
     def _write_packet(self, data):

tests/test_ipc.py

@@ -0,0 +1,36 @@
+import queue
+
+from hupper.ipc import Pipe, spawn
+
+
+def echo(pipe):
+    q = queue.Queue()
+    pipe.activate(q.put)
+    msg = q.get()
+    while msg is not None:
+        pipe.send(msg)
+        msg = q.get()
+    pipe.close()
+    pipe.reader_thread.join()
+
+
+def test_ipc_close():
+    c1, c2 = Pipe()
+    c1_q = queue.Queue()
+    c1.activate(c1_q.put)
+
+    with spawn(
+        __name__ + '.echo',
+        kwargs={"pipe": c2},
+        pass_fds=[c2.r_fd, c2.w_fd],
+    ) as proc:
+        try:
+            c2.close()
+
+            c1.send("hello world")
+            assert c1_q.get() == "hello world"
+
+            c1.close()
+            c1.reader_thread.join()
+        finally:
+            proc.terminate()