def __init__(
        self,
        source_generator: AsyncGenerator[T, None],
        queue_size: int = 1000,
        logger: Optional[logging.Logger] = None,
    ):
        """Initialize the async source stream.

        Args:
            source_generator: The source async generator
            queue_size: Size of the queue connecting producer and consumer
            logger: Optional contextualized logger, falls back to global logger if not provided
        """
        self.source_generator = source_generator
        # Queue is used to buffer entities and implement backpressure
        self.queue: asyncio.Queue[Optional[T]] = asyncio.Queue(maxsize=queue_size)
        self.producer_task = None
        self.is_running = True
        self.producer_done = asyncio.Event()
        self.producer_exception = None
        if logger is None:
            self.logger = logging.getLogger(__name__)
        else:
            self.logger = logger
Initialize the async source stream.

        Args:
            source_generator: The source async generator
            queue_size: Size of the queue connecting producer and consumer
            logger: Optional contextualized logger, falls back to global logger if not provided
        
backend/airweave/platform/sync/stream.py