send audio in single message of websocket

src/lesson.rs

@@ -395,11 +395,9 @@ impl InnerVoiceLesson {
                 while let Ok(translated) = translate_rx.recv().await {
                     let res = synthesize_speech(&client, translated, shared_voice_id.clone()).await;
                     match res {
-                        Ok((vec, mut audio_stream)) => {
+                        Ok((vec, audio)) => {
                             let _ = shared_lip_sync_tx.send(vec);
-                            while let Some(Ok(bytes)) = audio_stream.next().await {
-                                let _ = &shared_voice_lesson.send(bytes.to_vec());
-                            }
+                            let _ = &shared_voice_lesson.send(audio);
                         }
                         Err(e) => {
                             return Err(e);
@@ -453,7 +451,7 @@ async fn synthesize_speech(
     client: &aws_sdk_polly::Client,
     text: String,
     voice_id: VoiceId,
-) -> Result<(Vec<Viseme>, ByteStream), SynthesizeError> {
+) -> Result<(Vec<Viseme>, Vec<u8>), SynthesizeError> {
     let audio_fut = client
         .synthesize_speech()
         .engine(Engine::Neural)
@@ -469,10 +467,16 @@ async fn synthesize_speech(
         .speech_mark_types(SpeechMarkType::Viseme)
         .output_format(OutputFormat::Json)
         .send();
-    let (audio, visemes) = try_join(audio_fut, visemes_fut)
+    let (audio_out, visemes_out) = try_join(audio_fut, visemes_fut)
         .await
         .map_err(|e| SynthesizeError::Polly(e.into()))?;
-    let visemes = visemes
+    let audio = audio_out
+        .audio_stream
+        .collect()
+        .await
+        .map_err(|e| SynthesizeError::Transmitting(e.into()))?
+        .to_vec();
+    let visemes = visemes_out
         .audio_stream
         .collect()
         .await
@@ -482,5 +486,5 @@ async fn synthesize_speech(
         .lines()
         .flat_map(|line| Ok::<Viseme, anyhow::Error>(serde_json::from_str::<Viseme>(&line?)?))
         .collect();
-    Ok((parsed, audio.audio_stream))
+    Ok((parsed, audio))
 }

src/main.rs

@@ -407,7 +407,6 @@ fn u8_to_i16(input: &[u8]) -> Vec<i16> {
 
 #[cfg(test)]
 mod test {
-    use std::cell::{Cell};
     use std::time::Duration;
     use async_stream::stream;
     use poem::listener::{Acceptor, Listener};
@@ -490,7 +489,6 @@ mod test {
         };
         pin!(audio_stream);
 
-        let voice_flag = Cell::new(false);
         let recv_fut = async {
             while let Some(voice_slice) = audio_stream.next().await {
                 client_stream.send(Message::Binary(voice_slice)).await?;
@@ -503,8 +501,7 @@ mod test {
                 let Message::Text(json_str) = msg else { continue };
                 let Ok(evt) = serde_json::from_str::<SingleEvent>(&json_str) else { continue };
                 if let SingleEvent::Voice { .. } = evt {
-                    voice_flag.replace(true);
-                    break
+                    return Ok(())
                 }
             }
 
@@ -515,15 +512,14 @@ mod test {
             res = recv_fut => {
                 if let Err(e) = res {
                     error!("Error: {:?}", e);
+                    assert!(false, "Error: {}", e);
                 }
             }
             _ = sleep(Duration::from_secs(10)) => {
-                error!("timeout")
+                assert!(false, "timeout");
             }
-        }
+        };
 
         handle.abort();
-
-        assert!(voice_flag.get(), "voice not received");
     }
 }