Why does SpeechAsyncClient streaming recognise method hang

Question

Overview

I have implemented a simple client-server WebSocket connection using websocket.asyncio. My goal is to stream audio from the client to the server and to transcribe that audio using Google Cloud's SpeechAsyncClient. Here is the code for the server and client respectively:

Server

import re

from google.cloud import speech_v1p1beta1 as speech
import google.api_core.retry_async as retries
import google.api_core.exceptions as core_exceptions

import asyncio
from websockets.asyncio.server import serve

streaming_config = speech.StreamingRecognitionConfig()
streaming_config.interim_results = True
streaming_config.config.encoding = speech.RecognitionConfig.AudioEncoding.LINEAR16
streaming_config.config.sample_rate_hertz = 16000
streaming_config.config.language_code = "en-US"
streaming_config.config.audio_channel_count = 1
streaming_config.config.enable_automatic_punctuation = True
streaming_config.config.profanity_filter = True

retry = retries.AsyncRetry(
    initial=0.1,
    maximum=60.0,
    multiplier=1.3,
    predicate=retries.if_exception_type(
        core_exceptions.DeadlineExceeded,
        core_exceptions.ServiceUnavailable,
    ),
    deadline=5000.0,
)


class Server:
    def __init__(self) -> None:
        self._is_streaming = False
        self._audio_queue = asyncio.Queue()
        self._speech_client = speech.SpeechAsyncClient()

    async def _read_audio(self):
        print("Reading audio")

        config_request = speech.StreamingRecognizeRequest()
        config_request.streaming_config = streaming_config
        yield config_request

        while self._is_streaming:
            chunk = await self._audio_queue.get()
            if chunk is None:
                return
            data = [chunk]

            while True:
                try:
                    chunk = await self._audio_queue.get_nowait()
                    if chunk is None:
                        return
                    data.append(chunk)
                except asyncio.QueueEmpty:
                    break

            request = speech.StreamingRecognizeRequest()
            request.audio_content = b"".join(data)
            yield request

    async def _build_requests(self):
        print("Building requests")
        audio_generator = self._read_audio()
        responses = await self._speech_client.streaming_recognize(
            requests=audio_generator,
            retry=retry,
        )
        print("Done")
        await self._listen_print_loop(responses)

    async def _handler(self, websocket):
        print("Connection")
        asyncio.create_task(self._build_requests())
        self._is_streaming = True
        try:
            async for message in websocket:
                await self._audio_queue.put(message)
        except Exception as e:
            print(f"Failed: {e}")

    async def launch_server(self):
        print("Waiting for connection...", end=" ")
        
        server = await serve(
            self._handler,
            host="localhost",
            port=8080,
        )

        await server.serve_forever()

    async def _listen_print_loop(self, responses) -> str:
        num_chars_printed = 0
        transcript = ""
        async for response in responses:
            if not response.results:
                continue

            result = response.results[0]
            if not result.alternatives:
                continue

            transcript = result.alternatives[0].transcript
            overwrite_chars = " " * (num_chars_printed - len(transcript))

            if not result.is_final:
                print(transcript + overwrite_chars)
                num_chars_printed = len(transcript)

            else:
                print(transcript + overwrite_chars)
                if re.search(r"\b(exit|quit)\b", transcript, re.I):
                    print("Exiting..")
                    break

                num_chars_printed = 0

        return transcript


if __name__ == "__main__":
    server = Server()
    asyncio.run(server.launch_server())

Client

from websockets.asyncio.client import connect
import asyncio
import numpy as np
import sounddevice as sd
import queue


# Audio recording parameters
RATE = 16000
CHUNK = int(RATE / 10)  # 100ms


class MicrophoneStream:
    """Opens a recording stream as a generator yielding the audio chunks."""

    def __init__(self, rate: int = RATE, chunk: int = CHUNK):
        """The audio -- and generator -- is guaranteed to be on the main thread."""
        self._rate = rate
        self._chunk = chunk

        # Create a thread-safe buffer of audio data
        self._buff = queue.Queue()
        self.closed = True

    def __enter__(self) -> "MicrophoneStream":
        self.closed = False

        # Start the audio stream
        self._stream = sd.InputStream(
            samplerate=self._rate,
            channels=1,
            dtype='int16',
            blocksize=self._chunk,
            callback=self._fill_buffer
        )
        self._stream.start()

        return self

    def __exit__(self, type, value, traceback) -> None:
        """Closes the stream, regardless of whether the connection was lost or not."""
        self._stream.stop()
        self._stream.close()
        self.closed = True
        self._buff.put(None)

    def _fill_buffer(
            self, in_data: np.ndarray, frames: int, time, status
    ) -> None:
        """Continuously collect data from the audio stream, into the buffer.

        Args:
            in_data: The audio data as a NumPy array
            frames: The number of frames captured
            time: The time information
            status: The status flags
        """
        self._buff.put(in_data.tobytes())

    def generator(self):
        while not self.closed:
            # Use a blocking get() to ensure there's at least one chunk of
            # data, and stop iteration if the chunk is None, indicating the
            # end of the audio stream.
            chunk = self._buff.get()
            if chunk is None:
                return
            data = [chunk]

            while True:
                try:
                    chunk = self._buff.get(block=False)
                    if chunk is None:
                        return
                    data.append(chunk)
                except queue.Empty:
                    break

            yield b"".join(data)


async def start_client():
    uri = "ws://localhost:8080"
    async with connect(uri) as websocket:
        with MicrophoneStream(RATE, CHUNK) as stream:
            for chunk in stream.generator():
                await websocket.send(chunk)

if __name__ == "__main__":
    asyncio.run(start_client())

Problem

This sends audio from the client side continuously filling the audio buffer and the _build_requests method instantiates the async generator used by the SpeechAsyncClient.streaming_recognise method. However, this method hangs and doesn't get to the print("Done") line and never calls the generator _read_audio. Is there an issue in how I have used asyncio?