def _stream_with_heartbeat(
        self, stream: Callable[[], AsyncGenerator[str, None]], separator: str
    ) -> Callable[[], AsyncGenerator[str, None]]:
        """Add heartbeat messages to a GraphQL stream to prevent connection timeouts.

        This method wraps an async stream generator with heartbeat functionality by:
        1. Creating a queue to coordinate between data and heartbeat messages
        2. Running two concurrent tasks: one for original stream data, one for heartbeats
        3. Merging both message types into a single output stream

        Messages in the queue are tuples of (raised, done, data) where:
        - raised (bool): True if this contains an exception to be re-raised
        - done (bool): True if this is the final signal indicating stream completion
        - data: The actual message content to yield, or exception if raised=True
               Note: data is always None when done=True and can be ignored

        Note: This implementation addresses two critical concerns:

        1. Race condition: There's a potential race between checking task.done() and
           processing the final message. We solve this by having the drain task send
           an explicit (False, True, None) completion signal as its final action.
           Without this signal, we might exit before processing the final boundary.

           Since the queue size is 1 and the drain task will only complete after
           successfully queueing the done signal, task.done() guarantees the done
           signal is either in the queue or has already been processed. This ensures
           we never miss the final boundary.

        2. Flow control: The queue has maxsize=1, which is essential because:
           - It provides natural backpressure between producers and consumer
           - Prevents heartbeat messages from accumulating when drain is active
           - Ensures proper task coordination without complex synchronization
           - Guarantees the done signal is queued before drain task completes

        Heartbeats are sent every 5 seconds when the drain task isn't sending data.

        Note: Due to the asynchronous nature of the heartbeat task, an extra heartbeat
        message may be sent after the final stream boundary message. This is safe because
        both the MIME specification (RFC 2046) and Apollo's GraphQL Multipart HTTP protocol
        require clients to ignore any content after the final boundary marker. Additionally,
        Apollo's protocol defines heartbeats as empty JSON objects that clients must
        silently ignore.
        """
        queue: asyncio.Queue[tuple[bool, bool, Any]] = asyncio.Queue(
            maxsize=1,  # Critical: maxsize=1 for flow control.
        )
        cancelling = False

        async def drain() -> None:
            try:
                async for item in stream():
                    await queue.put((False, False, item))
            except Exception as e:
                if not cancelling:
                    await queue.put((True, False, e))
                else:
                    raise
            # Send completion signal to prevent race conditions. The queue.put()
            # blocks until space is available (due to maxsize=1), guaranteeing that
            # when task.done() is True, the final stream message has been dequeued.
            await queue.put((False, True, None))  # Always use None with done=True

        async def heartbeat() -> None:
            while True:
                item = self.encode_multipart_data({}, separator)
                await queue.put((False, False, item))

                await asyncio.sleep(5)

        async def merged() -> AsyncGenerator[str, None]:
            heartbeat_task = asyncio.create_task(heartbeat())
            task = asyncio.create_task(drain())

            async def cancel_tasks() -> None:
                nonlocal cancelling
                cancelling = True
                task.cancel()

                with contextlib.suppress(asyncio.CancelledError):
                    await task

                heartbeat_task.cancel()

                with contextlib.suppress(asyncio.CancelledError):
                    await heartbeat_task

            try:
                # When task.done() is True, the final stream message has been
                # dequeued due to queue size 1 and the blocking nature of queue.put().
                while not task.done():
                    raised, done, data = await queue.get()

                    if done:
                        # Received done signal (data is None), stream is complete.
                        # Note that we may not get here because of the race between
                        # task.done() and queue.get(), but that's OK because if
                        # task.done() is True, the actual final message (including any
                        # exception) has been consumed. The only intent here is to
                        # ensure that data=None is not yielded.
                        break

                    if raised:
                        await cancel_tasks()
                        raise data

                    yield data
            finally:
                await cancel_tasks()

        return merged
Add heartbeat messages to a GraphQL stream to prevent connection timeouts.

        This method wraps an async stream generator with heartbeat functionality by:
        1. Creating a queue to coordinate between data and heartbeat messages
        2. Running two concurrent tasks: one for original stream data, one for heartbeats
        3. Merging both message types into a single output stream

        Messages in the queue are tuples of (raised, done, data) where:
        - raised (bool): True if this contains an exception to be re-raised
        - done (bool): True if this is the final signal indicating stream completion
        - data: The actual message content to yield, or exception if raised=True
               Note: data is always None when done=True and can be ignored

        Note: This implementation addresses two critical concerns:

        1. Race condition: There's a potential race between checking task.done() and
           processing the final message. We solve this by having the drain task send
           an explicit (False, True, None) completion signal as its final action.
           Without this signal, we might exit before processing the final boundary.

           Since the queue size is 1 and the drain task will only complete after
           successfully queueing the done signal, task.done() guarantees the done
           signal is either in the queue or has already been processed. This ensures
           we never miss the final boundary.

        2. Flow control: The queue has maxsize=1, which is essential because:
           - It provides natural backpressure between producers and consumer
           - Prevents heartbeat messages from accumulating when drain is active
           - Ensures proper task coordination without complex synchronization
           - Guarantees the done signal is queued before drain task completes

        Heartbeats are sent every 5 seconds when the drain task isn't sending data.

        Note: Due to the asynchronous nature of the heartbeat task, an extra heartbeat
        message may be sent after the final stream boundary message. This is safe because
        both the MIME specification (RFC 2046) and Apollo's GraphQL Multipart HTTP protocol
        require clients to ignore any content after the final boundary marker. Additionally,
        Apollo's protocol defines heartbeats as empty JSON objects that clients must
        silently ignore.
        
strawberry/http/async_base_view.py