title: Websockets bridge for audio stream in FastAPI
body: <h2>Objective</h2>
  <p>My objective is to consume an audio stream. Logically, this is my objective:</p>
  <ol>
  <li>Audio stream comes through WebSocket A (<code>FastAPI</code> endpoint)</li>
  <li>Audio stream is bridged to a different WebSocket, B, which will return a JSON (<a href="https://www.rev.ai/docs/streaming#section/WebSocket-Endpoint" rel="nofollow noreferrer">Rev-ai</a>'s WebSocket)</li>
  <li>Json results are sent back through WebSocket A, in real-time. Thus, <strong>while the audio stream is still coming in.</strong></li>
  </ol>
  <h2>Possible solution</h2>
  <p>To solve this problem, I've had quite a few ideas, but ultimately I've been trying to bridge <code>WebSocket A</code> to <code>WebSocket B</code>. My attempt so far involves a <code>ConnectionManager</code> class, which contains a <code>Queue.queue</code>. The chunks of the audio stream are added to this queue so that we do not consume directly from <code>WebSocket A</code>.</p>
  <p>The <code>ConnectionManager</code> also contains a generator method to yield all values from the queue.</p>
  <p>My FastAPI implementation consumes from <code>websocket A</code> like this:</p>
  <pre><code>@app.websocket(&quot;/ws&quot;)
  async def predict_feature(websocket: WebSocket):
      await manager.connect(websocket)
      try:
          while True:
              chunk = await websocket.receive_bytes()
              manager.add_to_buffer(chunk)
      except KeyboardInterrupt:
          manager.disconnect()
  </code></pre>
  <p>Concurrent to this ingestion, I'd like to have a task that would bridge our audio stream to <code>WebSocket B</code>, and send the obtained values to <code>WebSocket A</code>. The audio stream could be consumed through the aforementioned <code>generator</code> method.</p>
  <p>The generator method is necessary due to how WebSocket B consumes messages, as shown in Rev-ai's <a href="https://github.com/revdotcom/revai-examples/blob/master/python_examples/generator_streaming_example.py" rel="nofollow noreferrer">examples</a>:</p>
  <pre><code>streamclient = RevAiStreamingClient(access_token, config)
  response_generator = streamclient.start(MEDIA_GENERATOR)
  for response in response_generator:
      # return through websocket A this value
      print(response)
  </code></pre>
  <p>This is one of the biggest challenges, as we need to be consuming data into a generator and getting the results in real-time.</p>
  <h2>Latest attempts</h2>
  <p>I've been trying my luck with <code>asyncio</code>; from what i'm understanding, a possibility would be to create a coroutine that would run in the background. I've been unsuccessful with this, but it sounded promising.</p>
  <p>I've thought about triggering this through the <code>FastAPI</code> startup event, but I'm having trouble achieving concurrency. I tried to use <code>event_loops</code>, but it gave me a <code>nested event loop</code> related error.</p>
  <h2>Caveat</h2>
  <p>FastAPI can be optional if your insight deems so, and in a way so is WebSocket A. At the end of the day, the ultimate objective is to receive an audio stream through our own API endpoint, run it through Rev.ai's WebSocket, do some extra processing, and send the results back.</p>