test single channel

Cargo.lock

@@ -121,7 +121,7 @@ dependencies = [
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "aws-types",
- "bytes",
+ "bytes 1.5.0",
  "fastrand",
  "hex",
  "http",
@@ -154,7 +154,7 @@ dependencies = [
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "aws-types",
- "bytes",
+ "bytes 1.5.0",
  "http",
  "http-body",
  "pin-project-lite",
@@ -200,7 +200,7 @@ dependencies = [
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "aws-types",
- "bytes",
+ "bytes 1.5.0",
  "http",
  "regex",
  "tracing",
@@ -222,7 +222,7 @@ dependencies = [
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "aws-types",
- "bytes",
+ "bytes 1.5.0",
  "http",
  "regex",
  "tracing",
@@ -244,7 +244,7 @@ dependencies = [
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "aws-types",
- "bytes",
+ "bytes 1.5.0",
  "http",
  "regex",
  "tracing",
@@ -291,7 +291,7 @@ dependencies = [
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "aws-types",
- "bytes",
+ "bytes 1.5.0",
  "http",
  "hyper",
  "regex",
@@ -314,7 +314,7 @@ dependencies = [
  "aws-smithy-runtime-api",
  "aws-smithy-types",
  "aws-types",
- "bytes",
+ "bytes 1.5.0",
  "fastrand",
  "http",
  "regex",
@@ -332,7 +332,7 @@ dependencies = [
  "aws-smithy-http",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
- "bytes",
+ "bytes 1.5.0",
  "form_urlencoded",
  "hex",
  "hmac",
@@ -363,7 +363,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1c669e1e5fc0d79561bf7a122b118bd50c898758354fe2c53eb8f2d31507cbc3"
 dependencies = [
  "aws-smithy-types",
- "bytes",
+ "bytes 1.5.0",
  "crc32fast",
 ]
 
@@ -376,7 +376,7 @@ dependencies = [
  "aws-smithy-eventstream",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
- "bytes",
+ "bytes 1.5.0",
  "bytes-utils",
  "futures-core",
  "http",
@@ -417,7 +417,7 @@ dependencies = [
  "aws-smithy-http",
  "aws-smithy-runtime-api",
  "aws-smithy-types",
- "bytes",
+ "bytes 1.5.0",
  "fastrand",
  "http",
  "http-body",
@@ -439,7 +439,7 @@ checksum = "4d27c3235d4972ed976b5c1a82286e7c4457f618f3c2ae6d4ae44f081dd24575"
 dependencies = [
  "aws-smithy-async",
  "aws-smithy-types",
- "bytes",
+ "bytes 1.5.0",
  "http",
  "pin-project-lite",
  "tokio",
@@ -454,7 +454,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d2fc32035dc0636a8583cf0c6dd7f1e6d5404103b836d26228b8730907a88d9f"
 dependencies = [
  "base64-simd",
- "bytes",
+ "bytes 1.5.0",
  "bytes-utils",
  "futures-core",
  "http",
@@ -604,6 +604,12 @@ version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
 
+[[package]]
+name = "bytes"
+version = "0.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0e4cec68f03f32e44924783795810fa50a7035d8c8ebe78580ad7e6c703fba38"
+
 [[package]]
 name = "bytes"
 version = "1.5.0"
@@ -616,7 +622,7 @@ version = "0.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7dafe3a8757b027e2be6e4e5601ed563c55989fcf1546e933c66c8eb3a058d35"
 dependencies = [
- "bytes",
+ "bytes 1.5.0",
  "either",
 ]
 
@@ -845,6 +851,21 @@ version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 
+[[package]]
+name = "futures"
+version = "0.3.28"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "23342abe12aba583913b2e62f22225ff9c950774065e4bfb61a19cd9770fec40"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
 [[package]]
 name = "futures-channel"
 version = "0.3.29"
@@ -852,6 +873,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ff4dd66668b557604244583e3e1e1eada8c5c2e96a6d0d6653ede395b78bbacb"
 dependencies = [
  "futures-core",
+ "futures-sink",
 ]
 
 [[package]]
@@ -860,6 +882,23 @@ version = "0.3.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "eb1d22c66e66d9d72e1758f0bd7d4fd0bee04cad842ee34587d68c07e45d088c"
 
+[[package]]
+name = "futures-executor"
+version = "0.3.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0f4fb8693db0cf099eadcca0efe2a5a22e4550f98ed16aba6c48700da29597bc"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
+
+[[package]]
+name = "futures-io"
+version = "0.3.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8bf34a163b5c4c52d0478a4d757da8fb65cabef42ba90515efee0f6f9fa45aaa"
+
 [[package]]
 name = "futures-macro"
 version = "0.3.29"
@@ -889,15 +928,30 @@ version = "0.3.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a19526d624e703a3179b3d322efec918b6246ea0fa51d41124525f00f1cc8104"
 dependencies = [
+ "futures-channel",
  "futures-core",
+ "futures-io",
  "futures-macro",
  "futures-sink",
  "futures-task",
+ "memchr",
  "pin-project-lite",
  "pin-utils",
  "slab",
 ]
 
+[[package]]
+name = "futures_codec"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ce54d63f8b0c75023ed920d46fd71d0cbbb830b0ee012726b5b4f506fb6dea5b"
+dependencies = [
+ "bytes 0.5.6",
+ "futures",
+ "memchr",
+ "pin-project",
+]
+
 [[package]]
 name = "fvad"
 version = "0.1.3"
@@ -946,7 +1000,7 @@ version = "0.3.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4d6250322ef6e60f93f9a2162799302cd6f68f79f6e5d85c8c16f14d1d958178"
 dependencies = [
- "bytes",
+ "bytes 1.5.0",
  "fnv",
  "futures-core",
  "futures-sink",
@@ -981,7 +1035,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "06683b93020a07e3dbcf5f8c0f6d40080d725bea7936fc01ad345c01b97dc270"
 dependencies = [
  "base64 0.21.5",
- "bytes",
+ "bytes 1.5.0",
  "headers-core",
  "http",
  "httpdate",
@@ -1037,13 +1091,19 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "hound"
+version = "3.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "62adaabb884c94955b19907d60019f4e145d091c75345379e70d1ee696f7854f"
+
 [[package]]
 name = "http"
 version = "0.2.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8947b1a6fad4393052c7ba1f4cd97bed3e953a95c79c92ad9b051a04611d9fbb"
 dependencies = [
- "bytes",
+ "bytes 1.5.0",
  "fnv",
  "itoa",
 ]
@@ -1054,7 +1114,7 @@ version = "0.4.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d5f38f16d184e36f2408a55281cd658ecbd3ca05cce6d6510a176eca393e26d1"
 dependencies = [
- "bytes",
+ "bytes 1.5.0",
  "http",
  "pin-project-lite",
 ]
@@ -1083,7 +1143,7 @@ version = "0.14.27"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ffb1cfd654a8219eaef89881fdb3bb3b1cdc5fa75ded05d6933b2b382e395468"
 dependencies = [
- "bytes",
+ "bytes 1.5.0",
  "futures-channel",
  "futures-core",
  "futures-util",
@@ -1476,6 +1536,26 @@ dependencies = [
  "sha2",
 ]
 
+[[package]]
+name = "pin-project"
+version = "0.4.30"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3ef0f924a5ee7ea9cbcea77529dba45f8a9ba9f622419fe3386ca581a3ae9d5a"
+dependencies = [
+ "pin-project-internal",
+]
+
+[[package]]
+name = "pin-project-internal"
+version = "0.4.30"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "851c8d0ce9bebe43790dedfc86614c23494ac9f423dd618d3a61fc693eafe61e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "pin-project-lite"
 version = "0.2.13"
@@ -1496,7 +1576,7 @@ checksum = "504774c97b0744c1ee108a37e5a65a9745a4725c4c06277521dabc28eb53a904"
 dependencies = [
  "async-trait",
  "base64 0.21.5",
- "bytes",
+ "bytes 1.5.0",
  "futures-util",
  "headers",
  "http",
@@ -1515,8 +1595,10 @@ dependencies = [
  "serde_json",
  "serde_urlencoded",
  "smallvec",
+ "sse-codec",
  "thiserror",
  "tokio",
+ "tokio-stream",
  "tokio-tungstenite",
  "tokio-util",
  "tracing",
@@ -1549,6 +1631,7 @@ dependencies = [
  "base64 0.21.5",
  "config",
  "futures-util",
+ "hound",
  "lazy_static",
  "once_cell",
  "poem",
@@ -1557,6 +1640,7 @@ dependencies = [
  "serde_yaml",
  "tokio",
  "tokio-stream",
+ "tokio-tungstenite",
  "tracing",
  "tracing-subscriber",
  "tracing-test",
@@ -2025,6 +2109,18 @@ version = "0.9.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67"
 
+[[package]]
+name = "sse-codec"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "84a59f811350c44b4a037aabeb72dc6a9591fc22aa95a036db9a96297c58085a"
+dependencies = [
+ "bytes 0.5.6",
+ "futures-io",
+ "futures_codec",
+ "memchr",
+]
+
 [[package]]
 name = "strsim"
 version = "0.8.0"
@@ -2157,7 +2253,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d0c014766411e834f7af5b8f4cf46257aab4036ca95e9d2c144a10f59ad6f5b9"
 dependencies = [
  "backtrace",
- "bytes",
+ "bytes 1.5.0",
  "libc",
  "mio",
  "num_cpus",
@@ -2218,8 +2314,9 @@ version = "0.7.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5419f34732d9eb6ee4c3578b7989078579b7f039cbbb9ca2c4da015749371e15"
 dependencies = [
- "bytes",
+ "bytes 1.5.0",
  "futures-core",
+ "futures-io",
  "futures-sink",
  "pin-project-lite",
  "tokio",
@@ -2355,7 +2452,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9e3dac10fd62eaf6617d3a904ae222845979aec67c615d1c842b4002c7666fb9"
 dependencies = [
  "byteorder",
- "bytes",
+ "bytes 1.5.0",
  "data-encoding",
  "http",
  "httparse",

Cargo.toml

@@ -39,3 +39,6 @@ whisper = { path="whisper", features = ["cuda"], optional = true }
 
 [dev-dependencies]
 tracing-test = "*"
+poem = { version = "1.3", features = ["websocket", "static-files", "test"] }
+tokio-tungstenite = "*"
+hound = "*"

src/asr/aws.rs

@@ -12,7 +12,7 @@ use tokio::select;
 use tokio::sync::broadcast::Receiver;
 use tokio_stream::Stream;
 use futures_util::TryStreamExt;
-use crate::asr::{ASR, Event};
+use crate::asr::{ASR, Event, slice_i16_to_u8};
 
 pub struct AwsAsr {
     speaker_voice_channel: tokio::sync::mpsc::Sender<Vec<i16>>,
@@ -82,16 +82,6 @@ impl AwsAsr {
     }
 }
 
-#[allow(dead_code)]
-fn slice_i16_to_u8(slice: &[i16]) -> Vec<u8> {
-    slice
-        .iter()
-        .flat_map(|&sample| {
-            [sample as u8, (sample >> 8) as u8]
-        })
-        .collect()
-}
-
 impl Drop for AwsAsr {
     fn drop(&mut self) {
         if let Some(drop_handler) = self.drop_handler.take() {

src/asr/mod.rs

@@ -16,3 +16,15 @@ pub(crate) trait ASR {
     async fn frame(&mut self, frame: Vec<i16>) -> anyhow::Result<()>;
     fn subscribe(&mut self) -> Receiver<Event>;
 }
+
+
+#[allow(dead_code)]
+pub(crate) fn slice_i16_to_u8(slice: &[i16]) -> Vec<u8> {
+    slice
+        .iter()
+        .flat_map(|&sample| {
+            [sample as u8, (sample >> 8) as u8]
+        })
+        .collect()
+}
+

src/main.rs

@@ -276,7 +276,7 @@ async fn stream_listener(
     })
 }
 
-#[derive(Serialize, Debug)]
+#[derive(Serialize, Deserialize, Debug)]
 enum SingleEvent {
     #[serde(rename = "original")]
     Transcription {
@@ -403,4 +403,127 @@ fn u8_to_i16(input: &[u8]) -> Vec<i16> {
             i16::from_le_bytes(buf)
         })
         .collect::<Vec<i16>>()
+}
+
+#[cfg(test)]
+mod test {
+    use std::cell::{Cell};
+    use std::time::Duration;
+    use async_stream::stream;
+    use poem::listener::{Acceptor, Listener};
+    use tokio::pin;
+    use tokio::time::sleep;
+    use tokio_stream::StreamExt;
+    use tokio_tungstenite::{
+        connect_async,
+        tungstenite::Message,
+    };
+    use tracing::{info, error};
+    use crate::asr::slice_i16_to_u8;
+    use super::*;
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    #[tracing_test::traced_test]
+    async fn test_single() {
+        let shared_config = aws_config::load_defaults(BehaviorVersion::latest()).await;
+        let ctx = Context {
+            lessons_manager: LessonsManager::new(&shared_config),
+        };
+
+        let query = SingleQuery {
+            id: 1,
+            from: "zh-CN".to_string(),
+            to: "en-US".to_string(),
+            voice: None,
+        };
+
+        let acceptor = TcpListener::bind("[::]:0")
+            .into_acceptor()
+            .await
+            .unwrap();
+        let addr = acceptor
+            .local_addr()
+            .remove(0)
+            .as_socket_addr()
+            .cloned()
+            .unwrap();
+        let server = Server::new_with_acceptor(acceptor);
+        let handle = tokio::spawn(async move {
+            let _ = server.run(
+                Route::new()
+                    .at("/ws/voice", get(stream_single))
+                    .data(ctx)
+            ).await;
+        });
+
+        let url = format!(
+            "ws://{}/ws/voice?id={}&from={}&to={}&voice={}",
+            addr, query.id, query.from, query.to, query.voice.unwrap_or("Amy".to_string())
+        );
+        let (mut client_stream, _) = connect_async(url)
+            .await
+            .unwrap();
+
+        client_stream
+            .send(Message::Binary(Vec::new()))
+            .await
+            .unwrap();
+
+
+        let wav = hound::WavReader::open("whisper/samples/samples_jfk.wav")
+            .expect("failed to open wav");
+        let spec = wav.spec();
+        info!("{:?}", spec);
+        let samples = wav
+            .into_samples::<i16>()
+            .map(|s| s.unwrap())
+            .collect::<Vec<i16>>();
+        let chunks = samples.chunks(1600)
+            .map(|chunk| chunk.to_vec())
+            .into_iter();
+
+        let audio_stream = stream! {
+            for chunk in chunks {
+                yield slice_i16_to_u8(&chunk);
+                sleep(Duration::from_millis(10)).await;
+            }
+        };
+        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?;
+            }
+            info!("sent all voice chunks");
+
+            while let Some(next_msg) = client_stream.next().await {
+                debug!(?next_msg);
+                let msg = next_msg?;
+                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
+                }
+            }
+
+            Ok(()) as anyhow::Result<()>
+        };
+
+        select! {
+            res = recv_fut => {
+                if let Err(e) = res {
+                    error!("Error: {:?}", e);
+                }
+            }
+            _ = sleep(Duration::from_secs(10)) => {
+                error!("timeout")
+            }
+        }
+
+        handle.abort();
+
+        assert!(voice_flag.get(), "voice not received");
+    }
 }