diff --git a/kafka/net/backend/abstract.py b/kafka/net/backend/abstract.py index b75687480..16432d08d 100644 --- a/kafka/net/backend/abstract.py +++ b/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 ---------------------------------------------------- diff --git a/kafka/net/backend/asyncio_backend.py b/kafka/net/backend/asyncio_backend.py index d14c931f6..369674474 100644 --- a/kafka/net/backend/asyncio_backend.py +++ b/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: diff --git a/kafka/net/backend/selector.py b/kafka/net/backend/selector.py index ea977cc54..2ab20c37c 100644 --- a/kafka/net/backend/selector.py +++ b/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