# Path: aio_pika/tools.py
# def create_task(
#     func: Callable[..., Union[Coroutine[Any, Any, T], Awaitable[T]]],
#     *args: Any,
#     loop: Optional[asyncio.AbstractEventLoop] = None,
#     **kwargs: Any
# ) -> Awaitable[T]:
#     loop = loop or asyncio.get_event_loop()
# 
#     if iscoroutinepartial(func):
#         return loop.create_task(func(*args, **kwargs))
# 
#     def run(future: asyncio.Future) -> Optional[asyncio.Future]:
#         if future.done():
#             return None
# 
#         try:
#             future.set_result(func(*args, **kwargs))
#         except Exception as e:
#             future.set_exception(e)
# 
#         return future
# 
#     future = loop.create_future()
#     loop.call_soon(run, future)
#     return future


import abc
import asyncio
import logging
from types import TracebackType
from typing import (
    Any, AsyncContextManager, Awaitable, Callable, Coroutine, Generic, Optional,
    Set, Tuple, Type, TypeVar, Union,
)
from aiormq.tools import awaitable
from aio_pika.tools import create_task
            if self._is_overflow:
                return await self.__items.get()

            log.debug("Creating a new instance of %r", self.__constructor)
            item = await self.__constructor(*self.__constructor_args)
            self.__created += 1
            self.__item_set.add(item)
            return item

    async def _get(self) -> T:
        if self.__closed:
            raise PoolInvalidStateError("get operation on closed pool")

        if self._is_overflow:
            return await self.__items.get()

        return await self._create_item()

    def put(self, item: T) -> None:
        if self.__closed:
            raise PoolInvalidStateError("put operation on closed pool")

        self.__items.put_nowait(item)

    async def close(self) -> None:
        async with self.__lock:
            self.__closed = True
            tasks = []

            for item in self.__item_set:

# Path: aio_pika/tools.py
# def create_task(
#     func: Callable[..., Union[Coroutine[Any, Any, T], Awaitable[T]]],
#     *args: Any,
#     loop: Optional[asyncio.AbstractEventLoop] = None,
#     **kwargs: Any
# ) -> Awaitable[T]:
#     loop = loop or asyncio.get_event_loop()
# 
#     if iscoroutinepartial(func):
#         return loop.create_task(func(*args, **kwargs))
# 
#     def run(future: asyncio.Future) -> Optional[asyncio.Future]:
#         if future.done():
#             return None
# 
#         try:
#             future.set_result(func(*args, **kwargs))
#         except Exception as e:
#             future.set_exception(e)
# 
#         return future
# 
#     future = loop.create_future()
#     loop.call_soon(run, future)
#     return future


# Path: aio_pika/pool.py
import abc
import asyncio
import logging
from types import TracebackType
from typing import (
    Any, AsyncContextManager, Awaitable, Callable, Coroutine, Generic, Optional,
    Set, Tuple, Type, TypeVar, Union,
)
from aiormq.tools import awaitable
from aio_pika.tools import create_task

            if self._is_overflow:
                return await self.__items.get()

            log.debug("Creating a new instance of %r", self.__constructor)
            item = await self.__constructor(*self.__constructor_args)
            self.__created += 1
            self.__item_set.add(item)
            return item

    async def _get(self) -> T:
        if self.__closed:
            raise PoolInvalidStateError("get operation on closed pool")

        if self._is_overflow:
            return await self.__items.get()

        return await self._create_item()

    def put(self, item: T) -> None:
        if self.__closed:
            raise PoolInvalidStateError("put operation on closed pool")

        self.__items.put_nowait(item)

    async def close(self) -> None:
        async with self.__lock:
            self.__closed = True
            tasks = []

            for item in self.__item_set:

                tasks.append(create_task(item.close))