From 669dbcc0e867f8f361d27c0bb1bb782edb8c6342 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ahmet=20Kaan=20G=C3=9CM=C3=9C=C5=9E?= <96421894+Tahinli@users.noreply.github.com> Date: Sun, 24 Mar 2024 16:29:43 +0300 Subject: [PATCH] feat: :sparkles: First microphone stream from server to client --- back/Cargo.toml | 6 +- back/src/lib.rs | 1 + back/src/main.rs | 3 +- back/src/routing.rs | 5 +- back/src/streaming.rs | 71 +++++++++++++ front/Cargo.toml | 6 +- front/src/main.rs | 225 ++++++++++++++++++++++++++---------------- streamer/src/main.rs | 4 +- 8 files changed, 229 insertions(+), 92 deletions(-) create mode 100644 back/src/streaming.rs diff --git a/back/Cargo.toml b/back/Cargo.toml index 5d80e63..8537955 100644 --- a/back/Cargo.toml +++ b/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"] } diff --git a/back/src/lib.rs b/back/src/lib.rs index 317af45..4b4db3c 100644 --- a/back/src/lib.rs +++ b/back/src/lib.rs @@ -1,6 +1,7 @@ use serde::{Deserialize, Serialize}; pub mod routing; +pub mod streaming; #[derive(Debug, Clone)] pub struct AppState{ diff --git a/back/src/main.rs b/back/src/main.rs index 1e3054f..6841203 100644 --- a/back/src/main.rs +++ b/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::().unwrap()); + tokio::spawn(streaming::start()); axum_server::bind_rustls(addr, config) .serve(app.into_make_service()) .await diff --git a/back/src/routing.rs b/back/src/routing.rs index 3892e00..ab1acb4 100644 --- a/back/src/routing.rs +++ b/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) diff --git a/back/src/streaming.rs b/back/src/streaming.rs new file mode 100644 index 0000000..c0dc91d --- /dev/null +++ b/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::::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, mut consumer: Consumer>>>>) { + 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>>>>) { + 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); +} diff --git a/front/Cargo.toml b/front/Cargo.toml index 81e1b9c..cd8f583 100644 --- a/front/Cargo.toml +++ b/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" diff --git a/front/src/main.rs b/front/src/main.rs index b539b41..6662d85 100644 --- a/front/src/main.rs +++ b/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) ->ServerSta async fn coin_status_check() -> Result { Ok(reqwest::get(format!("{}{}", SERVER_ADDRESS, "/coin")).await.unwrap().json::().await.unwrap()) } +async fn sound_stream(mut stream:WebSocketStream, mut producer: Producer>>>>) { + while let Some(msg) = stream.next().await { + match msg.unwrap().to_string().parse::() { + 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::::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(device: &cpal::Device, config: &cpal::StreamConfig) -> Result<(), anyhow::Error> -where - T: SizedSample + FromSample, -{ - let sample_rate = config.sample_rate.0 as f32; - let channels = config.channels as usize; +// pub async fn run(consumer: Consumer>>>>, device: &cpal::Device, config: &cpal::StreamConfig) -> Result<(), anyhow::Error> +// where +// T: SizedSample + FromSample, +// { +// 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(output: &mut [T], channels: usize, next_sample: &mut dyn FnMut() -> f32) +// where +// T: Sample + FromSample, +// { +// 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>>>>) { + + 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(output: &mut [T], channels: usize, next_sample: &mut dyn FnMut() -> f32) -where - T: Sample + FromSample, -{ - 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::(&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::(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(); } diff --git a/streamer/src/main.rs b/streamer/src/main.rs index c67c07f..764f4f9 100644 --- a/streamer/src/main.rs +++ b/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"); } };