Feature: urllib3 instead of curl

kombu/asynchronous/http/__init__.py

@@ -1,24 +1,19 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
 from kombu.asynchronous import get_event_loop
-from kombu.asynchronous.http.base import Headers, Request, Response
+from kombu.asynchronous.http.base import BaseClient, Headers, Request, Response
 from kombu.asynchronous.hub import Hub
 
-if TYPE_CHECKING:
-    from kombu.asynchronous.http.curl import CurlClient
-
 __all__ = ('Client', 'Headers', 'Response', 'Request')
 
 
-def Client(hub: Hub | None = None, **kwargs: int) -> CurlClient:
+def Client(hub: Hub | None = None, **kwargs: int) -> BaseClient:
     """Create new HTTP client."""
-    from .curl import CurlClient
-    return CurlClient(hub, **kwargs)
+    from .urllib3_client import Urllib3Client
+    return Urllib3Client(hub, **kwargs)
 
 
-def get_client(hub: Hub | None = None, **kwargs: int) -> CurlClient:
+def get_client(hub: Hub | None = None, **kwargs: int) -> BaseClient:
     """Get or create HTTP client bound to the current event loop."""
     hub = hub or get_event_loop()
     try:

kombu/asynchronous/http/urllib3_client.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+from collections import deque
+from io import BytesIO
+
+import urllib3
+
+from kombu.asynchronous.hub import Hub, get_event_loop
+from kombu.exceptions import HttpError
+
+from .base import BaseClient
+
+__all__ = ('Urllib3Client',)
+
+DEFAULT_USER_AGENT = 'Mozilla/5.0 (compatible; urllib3)'
+EXTRA_METHODS = frozenset(['DELETE', 'OPTIONS', 'PATCH'])
+
+
+class Urllib3Client(BaseClient):
+    """Urllib3 HTTP Client."""
+
+    def __init__(self, hub: Hub | None = None, max_clients: int = 10):
+        hub = hub or get_event_loop()
+        super().__init__(hub)
+        self.max_clients = max_clients
+        self._http = urllib3.PoolManager(maxsize=max_clients)
+        self._pending = deque()
+        self._timeout_check_tref = self.hub.call_repeatedly(
+            1.0, self._timeout_check,
+        )
+
+    def close(self):
+        self._timeout_check_tref.cancel()
+        self._http.clear()
+
+    def add_request(self, request):
+        self._pending.append(request)
+        self._process_queue()
+        return request
+
+    def _timeout_check(self):
+        self._process_pending_requests()
+
+    def _process_pending_requests(self):
+        while self._pending:
+            request = self._pending.popleft()
+            self._process_request(request)
+
+    def _process_request(self, request):
+        method = request.method
+        url = request.url
+        headers = request.headers
+        body = request.body
+
+        try:
+            response = self._http.request(
+                method,
+                url,
+                headers=headers,
+                body=body,
+                preload_content=False
+            )
+            buffer = BytesIO(response.data)
+            response_obj = self.Response(
+                request=request,
+                code=response.status,
+                headers=response.headers,
+                buffer=buffer,
+                effective_url=response.geturl(),
+                error=None
+            )
+        except urllib3.exceptions.HTTPError as e:
+            response_obj = self.Response(
+                request=request,
+                code=599,
+                headers={},
+                buffer=None,
+                effective_url=None,
+                error=HttpError(599, str(e))
+            )
+
+        request.on_ready(response_obj)
+
+    def _process_queue(self):
+        self._process_pending_requests()
+
+    def on_readable(self, fd):
+        pass
+
+    def on_writable(self, fd):
+        pass
+
+    def _setup_request(self, curl, request, buffer, headers):
+        pass

t/unit/asynchronous/http/test_http.py

@@ -147,7 +147,7 @@ def test_as_context(self):
 class test_Client:
 
     def test_get_client(self, hub):
-        pytest.importorskip('pycurl')
+        pytest.importorskip('urllib3')
         client = http.get_client()
         assert client.hub is hub
         client2 = http.get_client(hub)

t/unit/asynchronous/http/test_urllib3.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+from io import BytesIO
+from unittest.mock import Mock, patch
+
+import pytest
+
+import t.skip
+from kombu.asynchronous.http.urllib3_client import Urllib3Client
+
+
+@t.skip.if_pypy
+@pytest.mark.usefixtures('hub')
+class test_Urllib3Client:
+    class Client(Urllib3Client):
+        Urllib3 = Mock(name='Urllib3')
+
+    def test_max_clients_set(self):
+        x = self.Client(max_clients=303)
+        assert x.max_clients == 303
+
+    def test_init(self):
+        with patch(
+                'kombu.asynchronous.http.urllib3_client.urllib3.PoolManager'
+        ) as _PoolManager:
+            x = self.Client()
+            assert x._http is not None
+            assert x._pending is not None
+            assert x._timeout_check_tref
+
+            _PoolManager.assert_called_with(maxsize=x.max_clients)
+
+    def test_close(self):
+        with patch(
+                'kombu.asynchronous.http.urllib3_client.urllib3.PoolManager'
+        ):
+            x = self.Client()
+            x._timeout_check_tref = Mock(name='timeout_check_tref')
+            x.close()
+            x._timeout_check_tref.cancel.assert_called_with()
+            x._http.clear.assert_called_with()
+
+    def test_add_request(self):
+        with patch(
+                'kombu.asynchronous.http.urllib3_client.urllib3.PoolManager'
+        ):
+            x = self.Client()
+            x._process_queue = Mock(name='_process_queue')
+            request = Mock(name='request')
+            x.add_request(request)
+            assert request in x._pending
+            x._process_queue.assert_called_with()
+
+    def test_timeout_check(self):
+        with patch(
+                'kombu.asynchronous.http.urllib3_client.urllib3.PoolManager'
+        ):
+            hub = Mock(name='hub')
+            x = self.Client(hub)
+            x._process_pending_requests = Mock(name='process_pending')
+            x._timeout_check()
+            x._process_pending_requests.assert_called_with()
+
+    def test_process_request(self):
+        with (patch(
+                'kombu.asynchronous.http.urllib3_client.urllib3.PoolManager'
+        ) as _PoolManager):
+            x = self.Client()
+            request = Mock(
+                name='request',
+                method='GET',
+                url='http://example.com',
+                headers={},
+                body=None
+            )
+            response = Mock(
+                name='response',
+                status=200,
+                headers={},
+                data=b'content'
+            )
+            response.geturl.return_value = 'http://example.com'
+            _PoolManager.return_value.request.return_value = response
+
+            x._process_request(request)
+            response_obj = x.Response(
+                request=request,
+                code=200,
+                headers={},
+                buffer=BytesIO(b'content'),
+                effective_url='http://example.com',
+                error=None
+            )
+            request.on_ready.assert_called()
+            called_response = request.on_ready.call_args[0][0]
+            assert called_response.code == response_obj.code
+            assert called_response.headers == response_obj.headers
+            assert (
+                    called_response.buffer.getvalue() ==
+                    response_obj.buffer.getvalue()
+            )
+            assert called_response.effective_url == response_obj.effective_url
+            assert called_response.error == response_obj.error
+
+    def test_on_readable_on_writable(self):
+        x = self.Client()
+        x.on_readable(Mock(name='fd'))
+        x.on_writable(Mock(name='fd'))