Several fixes to npipe supportnpipe-advanced

- Fix _get_raw_response_socket to always return the NpipeSocket object
- Override NpipeHTTPConnectionPool._get_conn to avoid crash in urllib3
- Fix NpipeSocket.recv_into for Python 2
- Do not call select() on NpipeSocket objects

Signed-off-by: Joffrey F <joffrey@docker.com>

5 files changed, 44 insertions, 3 deletions

diff --git a/docker/client.py b/docker/client.py
index aba066b..aec78c8 100644
--- a/docker/client.py
+++ b/docker/client.py
@@ -220,7 +220,9 @@ class Client(
 
     def _get_raw_response_socket(self, response):
         self._raise_for_status(response)
-        if six.PY3:
+        if self.base_url == "http+docker://localnpipe":
+            sock = response.raw._fp.fp.raw.sock
+        elif six.PY3:
             sock = response.raw._fp.fp.raw
             if self.base_url.startswith("https://"):
                 sock = sock._sock
diff --git a/docker/transport/__init__.py b/docker/transport/__init__.py
index 04a46d9..d5560b6 100644
--- a/docker/transport/__init__.py
+++ b/docker/transport/__init__.py
@@ -2,5 +2,6 @@
 from .unixconn import UnixAdapter
 try:
     from .npipeconn import NpipeAdapter
+    from .npipesocket import NpipeSocket
 except ImportError:
     pass
diff --git a/docker/transport/npipeconn.py b/docker/transport/npipeconn.py
index 984049c..3054037 100644
--- a/docker/transport/npipeconn.py
+++ b/docker/transport/npipeconn.py
@@ -14,7 +14,6 @@ try:
 except ImportError:
     import urllib3
 
-
 RecentlyUsedContainer = urllib3._collections.RecentlyUsedContainer
 
 
@@ -46,6 +45,28 @@ class NpipeHTTPConnectionPool(urllib3.connectionpool.HTTPConnectionPool):
             self.npipe_path, self.timeout
         )
 
+    # When re-using connections, urllib3 tries to call select() on our
+    # NpipeSocket instance, causing a crash. To circumvent this, we override
+    # _get_conn, where that check happens.
+    def _get_conn(self, timeout):
+        conn = None
+        try:
+            conn = self.pool.get(block=self.block, timeout=timeout)
+
+        except AttributeError:  # self.pool is None
+            raise urllib3.exceptions.ClosedPoolError(self, "Pool is closed.")
+
+        except six.moves.queue.Empty:
+            if self.block:
+                raise urllib3.exceptions.EmptyPoolError(
+                    self,
+                    "Pool reached maximum size and no more "
+                    "connections are allowed."
+                )
+            pass  # Oh well, we'll create a new connection then
+
+        return conn or self._new_conn()
+
 
 class NpipeAdapter(requests.adapters.HTTPAdapter):
     def __init__(self, base_url, timeout=60,
diff --git a/docker/transport/npipesocket.py b/docker/transport/npipesocket.py
index 6dfc2f2..a9bf0cc 100644
--- a/docker/transport/npipesocket.py
+++ b/docker/transport/npipesocket.py
@@ -1,6 +1,7 @@
 import functools
 import io
 
+import six
 import win32file
 import win32pipe
 
@@ -115,6 +116,9 @@ class NpipeSocket(object):
 
     @check_closed
     def recv_into(self, buf, nbytes=0):
+        if six.PY2:
+            return self._recv_into_py2(buf, nbytes)
+
         readbuf = buf
         if not isinstance(buf, memoryview):
             readbuf = memoryview(buf)
@@ -125,6 +129,12 @@ class NpipeSocket(object):
         )
         return len(data)
 
+    def _recv_into_py2(self, buf, nbytes):
+        err, data = win32file.ReadFile(self._handle, nbytes or len(buf))
+        n = len(data)
+        buf[:n] = data
+        return n
+
     @check_closed
     def send(self, string, flags=0):
         err, nbytes = win32file.WriteFile(self._handle, string)
diff --git a/docker/utils/socket.py b/docker/utils/socket.py
index ed34350..164b845 100644
--- a/docker/utils/socket.py
+++ b/docker/utils/socket.py
@@ -5,6 +5,11 @@ import struct
 
 import six
 
+try:
+    from ..transport import NpipeSocket
+except ImportError:
+    NpipeSocket = type(None)
+
 
 class SocketError(Exception):
     pass
@@ -14,10 +19,12 @@ def read(socket, n=4096):
     """
     Reads at most n bytes from socket
     """
+
     recoverable_errors = (errno.EINTR, errno.EDEADLK, errno.EWOULDBLOCK)
 
     # wait for data to become available
-    select.select([socket], [], [])
+    if not isinstance(socket, NpipeSocket):
+        select.select([socket], [], [])
 
     try:
         if hasattr(socket, 'recv'):