Check for EOF in the kqueue-based IOLoop to avoid incorrectly calling

the connect callback when the connection was refused.

Closes tornadoweb#223.

tornado/ioloop.py

@@ -490,7 +490,17 @@ def poll(self, timeout):
             if kevent.filter == select.KQ_FILTER_READ:
                 events[fd] = events.get(fd, 0) | IOLoop.READ
             if kevent.filter == select.KQ_FILTER_WRITE:
-                events[fd] = events.get(fd, 0) | IOLoop.WRITE
+                if kevent.flags & select.KQ_EV_EOF:
+                    # If an asynchronous connection is refused, kqueue
+                    # returns a write event with the EOF flag set.
+                    # Turn this into an error for consistency with the
+                    # other IOLoop implementations.
+                    # Note that for read events, EOF may be returned before
+                    # all data has been consumed from the socket buffer,
+                    # so we only check for EOF on write events.
+                    events[fd] = IOLoop.ERROR
+                else:
+                    events[fd] = events.get(fd, 0) | IOLoop.WRITE
             if kevent.flags & select.KQ_EV_ERROR:
                 events[fd] = events.get(fd, 0) | IOLoop.ERROR
         return events.items()

tornado/test/iostream_test.py

@@ -1,5 +1,5 @@
 from tornado.iostream import IOStream
-from tornado.testing import AsyncHTTPTestCase, LogTrapTestCase
+from tornado.testing import AsyncHTTPTestCase, LogTrapTestCase, get_unused_port
 from tornado.web import RequestHandler, Application
 import socket
 
@@ -31,3 +31,18 @@ def test_read_zero_bytes(self):
         self.stream.read_bytes(3, self.stop)
         data = self.wait()
         self.assertEqual(data, "200")
+
+    def test_connection_refused(self):
+        # When a connection is refused, the connect callback should not
+        # be run.  (The kqueue IOLoop used to behave differently from the
+        # epoll IOLoop in this respect)
+        port = get_unused_port()
+        stream = IOStream(socket.socket(), self.io_loop)
+        self.connect_called = False
+        def connect_callback():
+            self.connect_called = True
+        stream.set_close_callback(self.stop)
+        stream.connect(("localhost", port), connect_callback)
+        self.wait()
+        self.assertFalse(self.connect_called)
+