feat: ✨ First microphone stream from server to client

back/Cargo.toml

@@ -6,12 +6,16 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-axum = "0.7.4"
+axum = { version = "0.7.4", features = ["macros"] }
 axum-server = { version = "0.6.0", features = ["tls-rustls"] }
+cpal = "0.15.3"
+futures-util = "0.3.30"
 rand = "0.8.5"
+ringbuf = "0.3.3"
 serde = { version = "1.0.197", features = ["derive"] }
 serde_json = "1.0.114"
 tokio = { version = "1.36.0", features = ["full"] }
+tokio-tungstenite = "0.21.0"
 #tokio-stream = { version = "0.1.15", features = ["full"] }
 tokio-util = { version = "0.7.10", features = ["full"] }
 tower-http = { version = "0.5.2", features = ["full"] }

back/src/lib.rs

@@ -1,6 +1,7 @@
 use serde::{Deserialize, Serialize};
 
 pub mod routing;
+pub mod streaming;
 
 #[derive(Debug, Clone)]
 pub struct AppState{

back/src/main.rs

@@ -1,4 +1,4 @@
-use back::{routing, AppState};
+use back::{AppState, routing, streaming};
 use std::{env, net::SocketAddr};
 use axum_server::tls_rustls::RustlsConfig;
 
@@ -23,6 +23,7 @@ async fn main() {
     };
     let app = routing::routing(axum::extract::State(state)).await;
     let addr = SocketAddr::from(take_args().parse::<SocketAddr>().unwrap());
+    tokio::spawn(streaming::start());
     axum_server::bind_rustls(addr, config)
         .serve(app.into_make_service())
         .await

back/src/routing.rs

@@ -1,4 +1,4 @@
-use crate::{AppState, ServerStatus, CoinStatus};
+use crate::{AppState, ServerStatus, CoinStatus, streaming};
 use axum::{body::Body, extract::State, http::StatusCode, response::IntoResponse, routing::get, Json, Router};
 use tokio::fs::File;
 use tokio_util::io::ReaderStream;
@@ -36,7 +36,10 @@ async fn flip_coin() -> impl IntoResponse {
     (StatusCode::OK, Json(coin_json))
 }
 
+#[axum::debug_handler]
 async fn stream() -> impl IntoResponse {
+    println!("Stream");
+    streaming::start().await;
     let file = File::open("audios/audio.mp3").await.unwrap();
     let stream = ReaderStream::new(file);
     Body::from_stream(stream)

back/src/streaming.rs

@@ -0,0 +1,71 @@
+use std::{mem::MaybeUninit, sync::Arc};
+
+use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
+use ringbuf::{Consumer, HeapRb, Producer, SharedRb};
+use tokio::net::{TcpListener, TcpStream};
+use futures_util::SinkExt;
+use tokio_tungstenite::WebSocketStream;
+
+pub async fn start() {
+    let socket = TcpListener::bind("127.0.0.1:2424").await.unwrap();
+    while let Ok((tcp_stream, _)) = socket.accept().await {
+        println!("Dude Someone Triggered");
+        let ring = HeapRb::<f32>::new(1000000);
+        let (producer, consumer) = ring.split();
+        let ws_stream = tokio_tungstenite::accept_async(tcp_stream).await.unwrap();
+        
+        tokio::spawn(record(producer));
+        std::thread::sleep(std::time::Duration::from_secs(3));
+        tokio::spawn(stream(ws_stream, consumer));
+    }
+}
+
+pub async fn stream(mut ws_stream:WebSocketStream<TcpStream>, mut consumer: Consumer<f32, Arc<SharedRb<f32, Vec<MaybeUninit<f32>>>>>) {
+    println!("Waiting");
+    loop {
+        if !consumer.is_empty() {
+            match consumer.pop() {
+                Some(data) => {
+                    ws_stream.send(data.to_string().into()).await.unwrap();       
+                }
+                None => {
+                    //ws_stream.send(0.0.to_string().into()).await.unwrap();
+                }
+            }
+            ws_stream.flush().await.unwrap();
+        }           
+    }
+}
+
+pub async fn record(mut producer: Producer<f32, Arc<SharedRb<f32, Vec<MaybeUninit<f32>>>>>) {
+    println!("Hello, world!");
+    let host = cpal::default_host();
+    let input_device = host.default_input_device().unwrap();
+
+    println!("Input Device: {}", input_device.name().unwrap());
+
+    let config:cpal::StreamConfig = input_device.default_input_config().unwrap().into();
+
+    let input_data_fn = move |data: &[f32], _:&cpal::InputCallbackInfo| {
+        for &sample in data {
+            match producer.push(sample) {
+                Ok(_) => {},
+                Err(_) => {},
+            }
+            println!("{}", sample);
+        }
+    };
+    
+
+    let input_stream = input_device.build_input_stream(&config, input_data_fn, err_fn, None).unwrap();
+    
+
+    println!("STREAMIN");
+    input_stream.play().unwrap();
+    std::thread::sleep(std::time::Duration::from_secs(100));
+    println!("DONE I HOPE");
+}
+
+fn err_fn(err: cpal::StreamError) {
+    eprintln!("Something Happened: {}", err);
+}

front/Cargo.toml

@@ -8,9 +8,13 @@ edition = "2021"
 [dependencies]
 anyhow = "1.0.81"
 cpal = { version = "0.15.3", features = ["wasm-bindgen"] }
-dioxus = { version = "0.5.0-alpha.0", features = ["web"] }
+dioxus = { version = "0.5.0-alpha.2", features = ["web"] }
+futures-core = "0.3.30"
+futures-util = { version = "0.3.30", features = ["futures-sink", "sink"] }
 log = "0.4.21"
 reqwest = { version = "0.11.24", features = ["json"] }
+ringbuf = "0.3.3"
 serde = { version = "1.0.197", features = ["derive"] }
+tokio-tungstenite-wasm = "0.3.1"
 tokio_with_wasm = "0.4.3"
 wasm-logger = "0.2.0"

front/src/main.rs

@@ -1,9 +1,14 @@
+use std::mem::MaybeUninit;
+use std::sync::Arc;
 use std::time::Duration;
+use ringbuf::{Consumer, HeapRb, Producer, SharedRb};
 use tokio_with_wasm::tokio;
+use tokio_tungstenite_wasm::*;
+use futures_util::*;
 use dioxus::prelude::*;
 use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
-use cpal::{FromSample, Sample, SizedSample};
 use serde::Deserialize;
+use tokio_with_wasm::tokio::time::sleep;
 
 const SERVER_ADDRESS: &str = "https://tahinli.com.tr:2323";
 
@@ -75,15 +80,39 @@ async fn server_status_check(mut server_status:Signal<ServerStatus>) ->ServerSta
 async fn coin_status_check() -> Result<CoinStatus, reqwest::Error> {
     Ok(reqwest::get(format!("{}{}", SERVER_ADDRESS, "/coin")).await.unwrap().json::<CoinStatus>().await.unwrap())
 }
+async fn sound_stream(mut stream:WebSocketStream, mut producer: Producer<f32, Arc<SharedRb<f32, Vec<MaybeUninit<f32>>>>>) {
+    while let Some(msg) = stream.next().await {
+        match msg.unwrap().to_string().parse::<f32>() {
+            Ok(sound_data) => {
+                match producer.push(sound_data) {
+                    Ok(_) => {},
+                    Err(_) => {},
+                }
+            }
+            Err(_) =>{}
+        };
+    }
+    log::info!("Connection Lost Sir");
+}
+async fn start_listening() {
+    log::info!("Trying Sir");
+    let connect_addr = "ws://127.0.0.1:2424";
+    let stream = tokio_tungstenite_wasm::connect(connect_addr).await.unwrap();
+    let ring = HeapRb::<f32>::new(1000000);
+    let (producer, consumer) = ring.split();
+    tokio_with_wasm::tokio::spawn(sound_stream(stream, producer));
+    tokio_with_wasm::tokio::time::sleep(Duration::from_secs(1));
+    tokio_with_wasm::tokio::spawn(listen(consumer));
+}
 fn app() -> Element {
     rsx! {
         page_base {}
-        audio_stream_renderer {}
+        //audio_stream_renderer {}
         div {
             button {
-                onclick: move |_| record(),
+                onclick: move |_| start_listening(),
                 "style":"width: 80px; height: 50px;",
-                "Sinusoidal"
+                "Listen"
             }
         }
         coin_status_renderer {}
@@ -107,97 +136,121 @@ fn page_base() ->Element {
     }
 }
 
-pub async fn run<T>(device: &cpal::Device, config: &cpal::StreamConfig) -> Result<(), anyhow::Error>
-where
-    T: SizedSample + FromSample<f32>,
-{
-    let sample_rate = config.sample_rate.0 as f32;
-    let channels = config.channels as usize;
+// pub async fn run<T>(consumer: Consumer<f32, Arc<SharedRb<f32, Vec<MaybeUninit<f32>>>>>, device: &cpal::Device, config: &cpal::StreamConfig) -> Result<(), anyhow::Error>
+// where
+//     T: SizedSample + FromSample<f32>,
+// {
+//     let sample_rate = config.sample_rate.0 as f32;
+//     let channels = config.channels as usize;
 
-    // Produce a sinusoid of maximum amplitude.
-    let mut sample_clock = 0f32;
-    let mut next_value = move || {
-        sample_clock = (sample_clock + 1.0) % sample_rate;
-        (sample_clock * 440.0 * 2.0 * std::f32::consts::PI / sample_rate).sin()
+//     // Produce a sinusoid of maximum amplitude.
+//     let mut sample_clock = 0f32;
+//     let mut next_value = move || {
+//         sample_clock = (sample_clock + 1.0) % sample_rate;
+//         (sample_clock * 440.0 * 2.0 * std::f32::consts::PI / sample_rate).sin()
+//     };
+
+//     let err_fn = |err| eprintln!("an error occurred on stream: {}", err);
+
+//     let stream = device.build_output_stream(
+//         config,
+//         move |data: &mut [T], _: &cpal::OutputCallbackInfo| {
+//             write_data(data, channels, &mut next_value)
+//         },
+//         err_fn,
+//         None,
+//     )?;
+//     stream.play()?;
+
+//     loop {
+        
+//     }
+
+//     //Ok(())
+// }
+
+// fn write_data<T>(output: &mut [T], channels: usize, next_sample: &mut dyn FnMut() -> f32)
+// where
+//     T: Sample + FromSample<f32>,
+// {
+//     for frame in output.chunks_mut(channels) {
+//         let value: T = T::from_sample(next_sample());
+//         for sample in frame.iter_mut() {
+//             *sample = value;
+//         }
+//     }
+//}
+
+
+fn err_fn(err: cpal::StreamError) {
+    eprintln!("Something Happened: {}", err);
+}
+pub async fn listen(mut consumer: Consumer<f32, Arc<SharedRb<f32, Vec<MaybeUninit<f32>>>>>) {
+
+    log::info!("Hi");
+    let host = cpal::default_host();
+    let output_device = host.default_output_device().unwrap();
+    let config:cpal::StreamConfig = output_device.default_output_config().unwrap().into();
+
+    let output_data_fn = move |data: &mut [f32], _:&cpal::OutputCallbackInfo| {
+        for sample in data {
+            *sample = match consumer.pop() {
+                Some(s) => s,
+                None => {0.0},
+            };
+        }
     };
 
-    let err_fn = |err| eprintln!("an error occurred on stream: {}", err);
 
-    let stream = device.build_output_stream(
-        config,
-        move |data: &mut [T], _: &cpal::OutputCallbackInfo| {
-            write_data(data, channels, &mut next_value)
-        },
-        err_fn,
-        None,
-    )?;
-    stream.play()?;
+    let output_stream = output_device.build_output_stream(&config, output_data_fn, err_fn, None).unwrap();
 
-    tokio::time::sleep(Duration::from_secs(3)).await;
+    output_stream.play().unwrap();
+    sleep(Duration::from_secs(100)).await;
+    output_stream.pause().unwrap();
+    // let host = cpal::default_host();
+    // let devices = host.devices().unwrap();
+    // for (_derive_index, device) in devices.enumerate() {
+    //     log::info!("{:?}", device.name());
+    // }
+    // let device = host.default_output_device().unwrap();
 
-    Ok(())
-}
-
-fn write_data<T>(output: &mut [T], channels: usize, next_sample: &mut dyn FnMut() -> f32)
-where
-    T: Sample + FromSample<f32>,
-{
-    for frame in output.chunks_mut(channels) {
-        let value: T = T::from_sample(next_sample());
-        for sample in frame.iter_mut() {
-            *sample = value;
-        }
-    }
-}
-
-
-
-pub async fn record() {
-    log::info!("mic");
-    let host = cpal::default_host();
-    let devices = host.devices().unwrap();
-    for (_derive_index, device) in devices.enumerate() {
-        log::info!("{:?}", device.name());
-    }
-    let device = host.default_output_device().unwrap();
-
-    let mut supported_config = device.supported_output_configs().unwrap();
-    let config = supported_config.next().unwrap().with_max_sample_rate();
-    log::info!("{:?}", config);
-    match config.sample_format() {
-        cpal::SampleFormat::I8 => {log::info!("i8")},
-        cpal::SampleFormat::I16 => {log::info!("i16")},
-        //cpal::SampleFormat::I24 => {log::info!("i24")},
-        cpal::SampleFormat::I32 => {log::info!("i32")},
-        //cpal::SampleFormat::I48 => {log::info!("i48")},
-        cpal::SampleFormat::I64 => {log::info!("i64")},
-        cpal::SampleFormat::U8 => {log::info!("u8")},
-        cpal::SampleFormat::U16 => {log::info!("u16")},
-        //cpal::SampleFormat::U24 => {log::info!("u24")},
-        cpal::SampleFormat::U32 => {log::info!("u32")},
-        //cpal::SampleFormat::U48 => {log::info!("u48")},
-        cpal::SampleFormat::U64 => {log::info!("u64")},
-        cpal::SampleFormat::F32 => {log::info!("f32");
-        run::<f32>(&device, &config.clone().into()).await.unwrap();},
-        cpal::SampleFormat::F64 => {log::info!("f64")},
-        sample_format => panic!("Unsupported sample format '{sample_format}'"),
-    }
-    /*let config:StreamConfig = config.into();
-    let stream = device.build_output_stream(
-        &config, 
-        move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
+    // let mut supported_config = device.supported_output_configs().unwrap();
+    // let config = supported_config.next().unwrap().with_max_sample_rate();
+    // log::info!("{:?}", config);
+    // match config.sample_format() {
+    //     cpal::SampleFormat::I8 => {log::info!("i8")},
+    //     cpal::SampleFormat::I16 => {log::info!("i16")},
+    //     //cpal::SampleFormat::I24 => {log::info!("i24")},
+    //     cpal::SampleFormat::I32 => {log::info!("i32")},
+    //     //cpal::SampleFormat::I48 => {log::info!("i48")},
+    //     cpal::SampleFormat::I64 => {log::info!("i64")},
+    //     cpal::SampleFormat::U8 => {log::info!("u8")},
+    //     cpal::SampleFormat::U16 => {log::info!("u16")},
+    //     //cpal::SampleFormat::U24 => {log::info!("u24")},
+    //     cpal::SampleFormat::U32 => {log::info!("u32")},
+    //     //cpal::SampleFormat::U48 => {log::info!("u48")},
+    //     cpal::SampleFormat::U64 => {log::info!("u64")},
+    //     cpal::SampleFormat::F32 => {log::info!("f32");
+    //     run::<f32>(consumer, &device, &config.clone().into()).await.unwrap();},
+    //     cpal::SampleFormat::F64 => {log::info!("f64")},
+    //     sample_format => panic!("Unsupported sample format '{sample_format}'"),
+    // }
+    // let config:StreamConfig = config.into();
+    // let stream = device.build_output_stream(
+    //     &config, 
+    //     move |data: &mut [f32], _: &cpal::OutputCallbackInfo| {
             
-            log::info!("{:?}", data);
-            //I need to do something here, I think
-        }, 
-        move |_err| {
+    //         log::info!("{:?}", data);
+    //         //I need to do something here, I think
+    //     }, 
+    //     move |_err| {
             
-        }, 
-        None).unwrap();
+    //     }, 
+    //     None).unwrap();
     
-    stream.play().unwrap();
-    tokio::time::sleep(Duration::from_secs(10)).await;
-    stream.pause().unwrap();*/
+    // stream.play().unwrap();
+    // tokio::time::sleep(Duration::from_secs(10)).await;
+    // stream.pause().unwrap();
 
     
 }

streamer/src/main.rs

@@ -33,7 +33,7 @@ async fn main() {
             }
         }
         if output_fell_behind {
-            eprintln!("Output consumed all, increase delay");
+            eprintln!("Too fast friend");
         }
     };
 
@@ -49,7 +49,7 @@ async fn main() {
             };
         }
         if input_fell_behind {
-            eprintln!("Input can't be fast enough, increase delay");
+            eprintln!("Too fast");
         }
     };