# Path: aio_pika/pool.py
# class Pool(Generic[T]):
#     __slots__ = (
#         "loop",
#         "__max_size",
#         "__items",
#         "__constructor",
#         "__created",
#         "__lock",
#         "__constructor_args",
#         "__item_set",
#         "__closed",
#     )
# 
#     def __init__(
#         self,
#         constructor: ConstructorType,
#         *args: Any,
#         max_size: int = None,
#         loop: Optional[asyncio.AbstractEventLoop] = None
#     ):
#         self.loop = loop or asyncio.get_event_loop()
#         self.__closed = False
#         self.__constructor: Callable[..., Awaitable[Any]] = awaitable(
#             constructor,
#         )
#         self.__constructor_args: Tuple[Any, ...] = args or ()
#         self.__created: int = 0
#         self.__item_set: Set[PoolInstance] = set()
#         self.__items: asyncio.Queue = asyncio.Queue()
#         self.__lock: asyncio.Lock = asyncio.Lock()
#         self.__max_size: Optional[int] = max_size
# 
#     @property
#     def is_closed(self) -> bool:
#         return self.__closed
# 
#     def acquire(self) -> "PoolItemContextManager[T]":
#         if self.__closed:
#             raise PoolInvalidStateError("acquire operation on closed pool")
# 
#         return PoolItemContextManager[T](self)
# 
#     @property
#     def _has_released(self) -> bool:
#         return self.__items.qsize() > 0
# 
#     @property
#     def _is_overflow(self) -> bool:
#         if self.__max_size:
#             return self.__created >= self.__max_size or self._has_released
#         return self._has_released
# 
#     async def _create_item(self) -> T:
#         if self.__closed:
#             raise PoolInvalidStateError("create item operation on closed pool")
# 
#         async with self.__lock:
#             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))
# 
#             if tasks:
#                 await asyncio.gather(*tasks, return_exceptions=True)
# 
#     async def __aenter__(self) -> "Pool":
#         return self
# 
#     async def __aexit__(
#         self,
#         exc_type: Optional[Type[BaseException]],
#         exc_val: Optional[BaseException],
#         exc_tb: Optional[TracebackType],
#     ) -> None:
#         if self.__closed:
#             return
# 
#         await asyncio.shield(self.close())


import asyncio
import aio_pika
from aio_pika.pool import Pool


async def main():
    loop = asyncio.get_event_loop()

    async def get_connection():
        return await aio_pika.connect_robust("amqp://guest:guest@localhost/")


# Path: aio_pika/pool.py
# class Pool(Generic[T]):
#     __slots__ = (
#         "loop",
#         "__max_size",
#         "__items",
#         "__constructor",
#         "__created",
#         "__lock",
#         "__constructor_args",
#         "__item_set",
#         "__closed",
#     )
# 
#     def __init__(
#         self,
#         constructor: ConstructorType,
#         *args: Any,
#         max_size: int = None,
#         loop: Optional[asyncio.AbstractEventLoop] = None
#     ):
#         self.loop = loop or asyncio.get_event_loop()
#         self.__closed = False
#         self.__constructor: Callable[..., Awaitable[Any]] = awaitable(
#             constructor,
#         )
#         self.__constructor_args: Tuple[Any, ...] = args or ()
#         self.__created: int = 0
#         self.__item_set: Set[PoolInstance] = set()
#         self.__items: asyncio.Queue = asyncio.Queue()
#         self.__lock: asyncio.Lock = asyncio.Lock()
#         self.__max_size: Optional[int] = max_size
# 
#     @property
#     def is_closed(self) -> bool:
#         return self.__closed
# 
#     def acquire(self) -> "PoolItemContextManager[T]":
#         if self.__closed:
#             raise PoolInvalidStateError("acquire operation on closed pool")
# 
#         return PoolItemContextManager[T](self)
# 
#     @property
#     def _has_released(self) -> bool:
#         return self.__items.qsize() > 0
# 
#     @property
#     def _is_overflow(self) -> bool:
#         if self.__max_size:
#             return self.__created >= self.__max_size or self._has_released
#         return self._has_released
# 
#     async def _create_item(self) -> T:
#         if self.__closed:
#             raise PoolInvalidStateError("create item operation on closed pool")
# 
#         async with self.__lock:
#             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))
# 
#             if tasks:
#                 await asyncio.gather(*tasks, return_exceptions=True)
# 
#     async def __aenter__(self) -> "Pool":
#         return self
# 
#     async def __aexit__(
#         self,
#         exc_type: Optional[Type[BaseException]],
#         exc_val: Optional[BaseException],
#         exc_tb: Optional[TracebackType],
#     ) -> None:
#         if self.__closed:
#             return
# 
#         await asyncio.shield(self.close())


# Path: docs/source/examples/pooling.py
import asyncio
import aio_pika
from aio_pika.pool import Pool



async def main():
    loop = asyncio.get_event_loop()

    async def get_connection():
        return await aio_pika.connect_robust("amqp://guest:guest@localhost/")


    connection_pool = Pool(get_connection, max_size=2, loop=loop)