Move call_soon_with_future / _invoke -> NetBackend base class (#3143)

Author: dpkp

kafka/net/backend/abstract.py

@@ -49,6 +49,7 @@
 """
 import abc
 import importlib
+import inspect
 from typing import Any, Callable, Optional, Protocol, Sequence, Tuple, runtime_checkable
 
 import kafka.errors as Errors
@@ -207,10 +208,6 @@ def call_soon(self, task: Any) -> Any:
         deferred-handle box).
         """
 
-    @abc.abstractmethod
-    def call_soon_with_future(self, coro: Any, *args: Any) -> NetBackendFuture:
-        """Schedule ``coro`` and return a future that resolves with its result."""
-
     @abc.abstractmethod
     def call_at(self, when: float, task: Any) -> Any:
         """Schedule ``task`` to run at absolute monotonic time ``when``."""
@@ -334,6 +331,33 @@ def wait_for(self, future: Any, timeout_ms: Optional[float], raise_error: bool=T
             if raise_error:
                 raise
 
+    async def _invoke(self, coro, *args):
+        """Invoke coro/awaitable/function and fully resolve the result."""
+        if inspect.iscoroutinefunction(coro):
+            result = await coro(*args)
+        elif hasattr(coro, '__await__'):
+            result = await coro
+        else:
+            result = coro(*args)
+        if inspect.iscoroutine(result) or hasattr(result, '__await__'):
+            result = await result
+        while isinstance(result, NetBackendFuture):
+            result = await result
+        return result
+
+    def call_soon_with_future(self, coro: Any, *args: Any) -> NetBackendFuture:
+        """Schedule ``coro`` and return a future that resolves with its result."""
+        if hasattr(coro, '__await__') and args:
+            raise ValueError('initiated coroutine does not accept args')
+        future = self.create_future()
+        async def wrapper():
+            try:
+                future.success(await self._invoke(coro, *args))
+            except BaseException as exc:
+                future.failure(exc)
+        self.call_soon(wrapper)
+        return future
+
 
 # --- backend selection ----------------------------------------------------
 

kafka/net/backend/asyncio_backend.py

@@ -286,57 +286,6 @@ def wakeup(self):
     def create_future(self):
         return AsyncioFuture(self._loop)
 
-    async def _resolve_future(self, fut):
-        """Await any kafka.future.Future (plain or AsyncioFuture) to its value."""
-        if fut.is_done:
-            if fut.exception is not None:
-                raise fut.exception
-            return fut.value
-        aio = self._loop.create_future()
-
-        def _cb(_=None):
-            if aio.done():
-                return
-            if fut.exception is not None:
-                aio.set_exception(fut.exception)
-            else:
-                aio.set_result(fut.value)
-
-        fut.add_both(_cb)
-        return await aio
-
-    async def _invoke(self, coro, *args):
-        """Invoke coro/awaitable/function and fully resolve the result.
-
-        Mirrors NetworkSelector._invoke, but bridges any trailing kafka Future
-        through _resolve_future (a plain Future isn't awaitable under asyncio).
-        """
-        if inspect.iscoroutinefunction(coro):
-            result = await coro(*args)
-        elif hasattr(coro, '__await__'):
-            result = await coro
-        else:
-            result = coro(*args)
-        if inspect.iscoroutine(result) or hasattr(result, '__await__'):
-            result = await result
-        while isinstance(result, Future):
-            result = await self._resolve_future(result)
-        return result
-
-    def call_soon_with_future(self, coro, *args):
-        if hasattr(coro, '__await__') and args:
-            raise ValueError('initiated coroutine does not accept args')
-        future = AsyncioFuture(self._loop)
-
-        async def wrapper():
-            try:
-                future.success(await self._invoke(coro, *args))
-            except BaseException as exc:
-                future.failure(exc)
-
-        self.call_soon(wrapper)
-        return future
-
     # --- cross-thread bridge ---------------------------------------------
     def run(self, coro, *args, timeout_ms=None):
         if self._closed:

kafka/net/backend/selector.py

@@ -501,37 +501,6 @@ def call_soon(self, task):
             self.wakeup()
         return task
 
-    def call_soon_with_future(self, coro, *args):
-        if hasattr(coro, '__await__'):
-            if args:
-                raise ValueError('initiated coroutine does not accept args')
-        future = SelectorFuture()  # returned to callers who await it
-        async def wrapper():
-            try:
-                future.success(await self._invoke(coro, *args))
-            except BaseException as exc:
-                future.failure(exc)
-        self.call_soon(wrapper)
-        return future
-
-    async def _invoke(self, coro, *args):
-        """Invoke coro/awaitable/function and fully resolve the result.
-
-        If the result is itself a Future (e.g. send() returning an unresolved
-        Future), it is awaited so callers receive the resolved value.
-        """
-        if inspect.iscoroutinefunction(coro):
-            result = await coro(*args)
-        elif hasattr(coro, '__await__'):
-            result = await coro
-        else:
-            result = coro(*args)
-        if inspect.iscoroutine(result) or hasattr(result, '__await__'):
-            result = await result
-        while isinstance(result, Future):
-            result = await result
-        return result
-
     def _unschedule(self, task):
         assert task.state is TaskState.SCHEDULED
         assert task.scheduled_at is not None