Merge pull request #603 from dpkp/windows

Updates to support Windows

5 files changed, 83 insertions, 11 deletions

diff --git a/build_integration.sh b/build_integration.sh
index 4785028..0babfa1 100755
--- a/build_integration.sh
+++ b/build_integration.sh
@@ -54,7 +54,16 @@ pushd servers
         fi
         if [ ! -f "../$kafka/kafka-bin/bin/kafka-run-class.sh" ]; then
           echo "Downloading kafka ${kafka} tarball"
-          wget -N https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tgz || wget -N https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tar.gz
+          if hash wget 2>/dev/null; then
+            wget -N https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tgz || wget -N https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tar.gz
+          else
+            echo "wget not found... using curl"
+            if [ -f "${KAFKA_ARTIFACT}.tar.gz" ]; then
+              echo "Using cached artifact: ${KAFKA_ARTIFACT}.tar.gz"
+            else
+              curl -f https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tgz -o ${KAFKA_ARTIFACT}.tar.gz || curl -f https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tar.gz -o ${KAFKA_ARTIFACT}.tar.gz
+            fi
+          fi
           echo
           echo "Extracting kafka ${kafka} binaries"
           tar xzvf ${KAFKA_ARTIFACT}.t* -C ../$kafka/
diff --git a/kafka/client_async.py b/kafka/client_async.py
index 57aea66..ae9dbb4 100644
--- a/kafka/client_async.py
+++ b/kafka/client_async.py
@@ -4,9 +4,9 @@ import copy
 import heapq
 import itertools
 import logging
-import os
 import random
 import select
+import socket
 import time
 
 import six
@@ -18,6 +18,7 @@ from .conn import BrokerConnection, ConnectionStates, collect_hosts
 from .future import Future
 from .protocol.metadata import MetadataRequest
 from .protocol.produce import ProduceRequest
+from . import socketpair
 from .version import __version__
 
 if six.PY2:
@@ -97,11 +98,11 @@ class KafkaClient(object):
         self._last_bootstrap = 0
         self._bootstrap_fails = 0
         self._bootstrap(collect_hosts(self.config['bootstrap_servers']))
-        self._wake_r, self._wake_w = os.pipe()
+        self._wake_r, self._wake_w = socket.socketpair()
 
     def __del__(self):
-        os.close(self._wake_r)
-        os.close(self._wake_w)
+        self._wake_r.close()
+        self._wake_w.close()
 
     def _bootstrap(self, hosts):
         # Exponential backoff if bootstrap fails
@@ -674,14 +675,15 @@ class KafkaClient(object):
         return version
 
     def wakeup(self):
-        os.write(self._wake_w, b'x')
+        if self._wake_w.send(b'x') != 1:
+            log.warning('Unable to send to wakeup socket!')
 
     def _clear_wake_fd(self):
         while True:
             fds, _, _ = select.select([self._wake_r], [], [], 0)
             if not fds:
                 break
-            os.read(self._wake_r, 1)
+            self._wake_r.recv(1)
 
 
 class DelayedTaskQueue(object):
diff --git a/kafka/conn.py b/kafka/conn.py
index 015bf23..65451f9 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -90,9 +90,10 @@ class BrokerConnection(object):
                 pass
             self.last_attempt = time.time()
 
-            if not ret or ret is errno.EISCONN:
+            if not ret or ret == errno.EISCONN:
                 self.state = ConnectionStates.CONNECTED
-            elif ret in (errno.EINPROGRESS, errno.EALREADY):
+            # WSAEINVAL == 10022, but errno.WSAEINVAL is not available on non-win systems
+            elif ret in (errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK, 10022):
                 self.state = ConnectionStates.CONNECTING
             else:
                 log.error('Connect attempt to %s returned error %s.'
@@ -114,9 +115,10 @@ class BrokerConnection(object):
                     ret = self._sock.connect_ex((self.host, self.port))
                 except socket.error as ret:
                     pass
-                if not ret or ret is errno.EISCONN:
+                if not ret or ret == errno.EISCONN:
                     self.state = ConnectionStates.CONNECTED
-                elif ret is not errno.EALREADY:
+                # WSAEINVAL == 10022, but errno.WSAEINVAL is not available on non-win systems
+                elif ret not in (errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK, 10022):
                     log.error('Connect attempt to %s returned error %s.'
                               ' Disconnecting.', self, ret)
                     self.close()
diff --git a/kafka/socketpair.py b/kafka/socketpair.py
new file mode 100644
index 0000000..6a87c4d
--- /dev/null
+++ b/kafka/socketpair.py
@@ -0,0 +1,57 @@
+# pylint: skip-file
+# vendored from https://github.com/mhils/backports.socketpair
+
+import sys
+import socket
+import errno
+
+_LOCALHOST    = '127.0.0.1'
+_LOCALHOST_V6 = '::1'
+
+if not hasattr(socket, "socketpair"):
+    # Origin: https://gist.github.com/4325783, by Geert Jansen.  Public domain.
+    def socketpair(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0):
+        if family == socket.AF_INET:
+            host = _LOCALHOST
+        elif family == socket.AF_INET6:
+            host = _LOCALHOST_V6
+        else:
+            raise ValueError("Only AF_INET and AF_INET6 socket address families "
+                             "are supported")
+        if type != socket.SOCK_STREAM:
+            raise ValueError("Only SOCK_STREAM socket type is supported")
+        if proto != 0:
+            raise ValueError("Only protocol zero is supported")
+
+        # We create a connected TCP socket. Note the trick with
+        # setblocking(False) that prevents us from having to create a thread.
+        lsock = socket.socket(family, type, proto)
+        try:
+            lsock.bind((host, 0))
+            lsock.listen(min(socket.SOMAXCONN, 128))
+            # On IPv6, ignore flow_info and scope_id
+            addr, port = lsock.getsockname()[:2]
+            csock = socket.socket(family, type, proto)
+            try:
+                csock.setblocking(False)
+                if sys.version_info >= (3, 0):
+                    try:
+                        csock.connect((addr, port))
+                    except (BlockingIOError, InterruptedError):
+                        pass
+                else:
+                    try:
+                        csock.connect((addr, port))
+                    except socket.error as e:
+                        if e.errno != errno.WSAEWOULDBLOCK:
+                            raise
+                csock.setblocking(True)
+                ssock, _ = lsock.accept()
+            except:
+                csock.close()
+                raise
+        finally:
+            lsock.close()
+        return (ssock, csock)
+
+    socket.socketpair = socketpair
diff --git a/test/test_consumer.py b/test/test_consumer.py
index 2c9561b..e664292 100644
--- a/test/test_consumer.py
+++ b/test/test_consumer.py
@@ -1,3 +1,4 @@
+import sys
 
 from mock import MagicMock, patch
 from . import unittest
@@ -17,6 +18,7 @@ class TestKafkaConsumer(unittest.TestCase):
 
 
 class TestMultiProcessConsumer(unittest.TestCase):
+    @unittest.skipIf(sys.platform.startswith('win'), 'test mocking fails on windows')
     def test_partition_list(self):
         client = MagicMock()
         partitions = (0,)