Tahinli/radioxide
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #22 from Tahinli/dev

Browse source 
Dev
This commit is contained in:
Ahmet Kaan GÜMÜŞ 2024-03-27 00:44:14 +00:00 • committed by GitHub
commit 9e836fe07f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 413 additions and 296 deletions
Download patch file Download diff file Expand all files Collapse all files

6
back/Cargo.toml
View file

@ -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"] }

1
back/src/lib.rs
View file

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

3
back/src/main.rs
View file

@ -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

5
back/src/routing.rs
View file

@ -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)

71
back/src/streaming.rs Normal file
View file

@ -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);
}

6
front/Cargo.toml
View file

@ -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"

117
front/src/components.rs Normal file
View file

@ -0,0 +1,117 @@
use crate::status::{
coin_status_check, server_status_check, Coin, CoinStatus, Server, ServerStatus,
};
use dioxus::prelude::*;
use std::time::Duration;
#[component]
pub fn server_status_renderer(server_address: String) -> Element {
let server_check_time = 1_u64;
let mut server_status = use_signal(move || ServerStatus {
status: Server::Unstable,
});
let mut server_status_watchdog = use_signal(move || false);
let mut server_status_unstable = use_signal(move || false);
let _server_status_task: Coroutine<()> = use_coroutine(|_| async move {
loop {
tokio_with_wasm::tokio::time::sleep(Duration::from_secs(server_check_time)).await;
*server_status_watchdog.write() = true;
*server_status.write() = server_status_check(server_status, &server_address).await;
*server_status_watchdog.write() = false;
}
});
let _server_status_watchdog_timer: Coroutine<()> = use_coroutine(|_| async move {
let mut watchdog_counter = 0_i8;
loop {
tokio_with_wasm::tokio::time::sleep(Duration::from_secs(2 * server_check_time + 1))
.await;
if !server_status_watchdog() {
*server_status_unstable.write() = false;
}
if server_status_watchdog() {
for _i in 0..5 {
tokio_with_wasm::tokio::time::sleep(Duration::from_secs(1)).await;
if server_status_watchdog() {
watchdog_counter += 1;
}
}
}
if watchdog_counter > 4 {
*server_status_unstable.write() = true;
}
watchdog_counter = 0;
}
});
rsx! {
if server_status_unstable() && server_status_watchdog() {
ShowServerStatus {server_status:ServerStatus{status:Server::Dead,}}
}
else {
ShowServerStatus {server_status:server_status()}
}
}
}
#[component]
pub fn coin_status_renderer(server_address: String) -> Element {
let is_loading = use_signal(|| false);
let coin_result = use_signal(|| CoinStatus { status: Coin::Head });
let call_coin = move |_| {
spawn({
to_owned![is_loading];
to_owned![coin_result];
to_owned![server_address];
is_loading.set(true);
async move {
match coin_status_check(&server_address).await {
Ok(coin_status) => {
is_loading.set(false);
coin_result.set(coin_status);
}
Err(_) => {
is_loading.set(false);
}
}
}
});
};
rsx! {
div {
button {
disabled: is_loading(),
onclick: call_coin,
"style":"width: 80px; height: 50px;",
if is_loading() {
"Loading"
}else {
"Coin Flip"
}
}
div {
ShowCoinStatus{ coin_status: coin_result().clone() }
}
}
}
}
#[component]
fn ShowServerStatus(server_status: ServerStatus) -> Element {
rsx! {
div {
div { class: "flex items-center",
span { "Server Status: " }
span { { server_status.status.to_string() } }
}
}
}
}
#[component]
fn ShowCoinStatus(coin_status: CoinStatus) -> Element {
rsx! {
div {
div { class: "flex items-center",
span { "Coin Status: " }
span { { coin_status.status.to_string() } }
}
}
}
}

3
front/src/lib.rs Normal file
View file

@ -0,0 +1,3 @@
pub mod components;
pub mod status;
pub mod streaming;

301
front/src/main.rs
View file

@ -1,48 +1,8 @@
use std::time::Duration;
use tokio_with_wasm::tokio;
use dioxus::prelude::*;
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use cpal::{FromSample, Sample, SizedSample};
use serde::Deserialize;
const SERVER_ADDRESS: &str = "https://tahinli.com.tr:2323";
#[derive(Debug, Clone, PartialEq, Deserialize)]
enum Server{
Alive,
Unstable,
Dead,
}
impl Server {
fn to_string(&mut self) -> String{
match self {
Self::Alive => {String::from("Alive")},
Self::Unstable => {String::from("Unstable")},
Self::Dead => {String::from("Dead")},
}
}
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
enum Coin{
Tail,
Head,
}
impl Coin {
fn to_string(&mut self) -> String {
match self {
Self::Head => {String::from("Head")},
Self::Tail => {String::from("Tail")},
}
}
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
struct ServerStatus {
status:Server,
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
struct CoinStatus {
status:Coin,
}
use front::{
components::{coin_status_renderer, server_status_renderer},
streaming::start_listening,
};
fn main() {
println!("Hello, world!");
@ -50,48 +10,23 @@ fn main() {
launch(app);
}
async fn server_status_check(mut server_status:Signal<ServerStatus>) ->ServerStatus {
match reqwest::get(SERVER_ADDRESS).await {
Ok(response) => {
match response.json::<ServerStatus>().await {
Ok(_) => {
*server_status.write() = ServerStatus{status:Server::Alive,};
ServerStatus{status:Server::Alive,}
}
Err(err_val) => {
*server_status.write() = ServerStatus{status:Server::Dead,};
log::info!("{}", err_val);
ServerStatus{status:Server::Dead,}
}
}
}
Err(err_val) => {
*server_status.write() = ServerStatus{status:Server::Dead,};
log::info!("{}", err_val);
ServerStatus{status:Server::Dead,}
}
}
}
async fn coin_status_check() -> Result<CoinStatus, reqwest::Error> {
Ok(reqwest::get(format!("{}{}", SERVER_ADDRESS, "/coin")).await.unwrap().json::<CoinStatus>().await.unwrap())
}
fn app() -> Element {
let server_address = "https://tahinli.com.tr:2323".to_string();
rsx! {
page_base {}
audio_stream_renderer {}
div {
button {
onclick: move |_| record(),
onclick: move |_| start_listening(),
"style":"width: 80px; height: 50px;",
"Sinusoidal"
"Listen"
}
}
coin_status_renderer {}
server_status_renderer {}
coin_status_renderer {server_address:server_address.clone()}
server_status_renderer {server_address:server_address.clone()}
}
}
fn page_base() ->Element {
fn page_base() -> Element {
rsx! {
h1 {
"Radioxide"
@ -102,221 +37,7 @@ fn page_base() ->Element {
span {
}
}
}
}
}
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;
// 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()?;
tokio::time::sleep(Duration::from_secs(3)).await;
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| {
log::info!("{:?}", data);
//I need to do something here, I think
},
move |_err| {
},
None).unwrap();
stream.play().unwrap();
tokio::time::sleep(Duration::from_secs(10)).await;
stream.pause().unwrap();*/
}
fn server_status_renderer() -> Element {
let server_check_time = 1_u64;
let mut server_status = use_signal(move || ServerStatus{status:Server::Unstable,});
let mut server_status_watchdog = use_signal(move|| false);
let mut server_status_unstable = use_signal(move|| false);
let _server_status_task:Coroutine<()> = use_coroutine(|_| async move {
loop {
tokio::time::sleep(Duration::from_secs(server_check_time)).await;
*server_status_watchdog.write() = true;
*server_status.write() = server_status_check(server_status).await;
*server_status_watchdog.write() = false;
};
});
let _server_status_watchdog_timer:Coroutine<()> = use_coroutine(|_| async move {
let mut watchdog_counter = 0_i8;
loop {
tokio::time::sleep(Duration::from_secs(2*server_check_time+1)).await;
if !server_status_watchdog() {
*server_status_unstable.write() = false;
}
if server_status_watchdog() {
for _i in 0..5 {
tokio::time::sleep(Duration::from_secs(1)).await;
if server_status_watchdog() {
watchdog_counter += 1;
}
}
}
if watchdog_counter > 4 {
*server_status_unstable.write() = true;
}
watchdog_counter = 0;
}
});
rsx! {
if server_status_unstable() && server_status_watchdog() {
ShowServerStatus {server_status:ServerStatus{status:Server::Dead,}}
}
else {
ShowServerStatus {server_status:server_status()}
}
}
}
fn coin_status_renderer() -> Element {
let is_loading = use_signal(|| false);
let coin_result = use_signal(|| CoinStatus{status:Coin::Head,});
let call_coin = move |_| {
spawn({
to_owned![is_loading];
to_owned![coin_result];
is_loading.set(true);
async move {
match coin_status_check().await {
Ok(coin_status) => {
is_loading.set(false);
coin_result.set(coin_status);
}
Err(_) => {
is_loading.set(false);
}
}
}
});
};
rsx! {
div {
button {
disabled: is_loading(),
onclick: call_coin,
"style":"width: 80px; height: 50px;",
if is_loading() {
"Loading"
}else {
"Coin Flip"
}
}
div {
ShowCoinStatus{ coin_status: coin_result().clone() }
}
}
}
}
fn audio_stream_renderer() -> Element {
rsx! {
div {
audio{
src:"https://tahinli.com.tr:2323/stream",
controls:true,
autoplay: true,
muted:false,
r#loop:true,
}
}
}
}
#[component]
fn ShowServerStatus(server_status: ServerStatus) -> Element {
rsx! {
div {
div { class: "flex items-center",
span { "Server Status: " }
span { { server_status.status.to_string() } }
}
}
}
}
#[component]
fn ShowCoinStatus(coin_status: CoinStatus) -> Element {
rsx! {
div {
div { class: "flex items-center",
span { "Coin Status: " }
span { { coin_status.status.to_string() } }
}
}
}
}

83
front/src/status.rs Normal file
View file

@ -0,0 +1,83 @@
use dioxus::signals::{Signal, Writable};
use serde::Deserialize;
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub enum Server {
Alive,
Unstable,
Dead,
}
impl Server {
pub fn to_string(&mut self) -> String {
match self {
Self::Alive => String::from("Alive"),
Self::Unstable => String::from("Unstable"),
Self::Dead => String::from("Dead"),
}
}
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub enum Coin {
Tail,
Head,
}
impl Coin {
pub fn to_string(&mut self) -> String {
match self {
Self::Head => String::from("Head"),
Self::Tail => String::from("Tail"),
}
}
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct ServerStatus {
pub status: Server,
}
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct CoinStatus {
pub status: Coin,
}
pub async fn server_status_check(
mut server_status: Signal<ServerStatus>,
server_address: &String,
) -> ServerStatus {
match reqwest::get(server_address).await {
Ok(response) => match response.json::<ServerStatus>().await {
Ok(_) => {
*server_status.write() = ServerStatus {
status: Server::Alive,
};
ServerStatus {
status: Server::Alive,
}
}
Err(err_val) => {
*server_status.write() = ServerStatus {
status: Server::Dead,
};
log::info!("{}", err_val);
ServerStatus {
status: Server::Dead,
}
}
},
Err(err_val) => {
*server_status.write() = ServerStatus {
status: Server::Dead,
};
log::info!("{}", err_val);
ServerStatus {
status: Server::Dead,
}
}
}
}
pub async fn coin_status_check(server_address: &String) -> Result<CoinStatus, reqwest::Error> {
Ok(reqwest::get(format!("{}{}", server_address, "/coin"))
.await
.unwrap()
.json::<CoinStatus>()
.await
.unwrap())
}

109
front/src/streaming.rs Normal file
View file

@ -0,0 +1,109 @@
use std::{mem::MaybeUninit, sync::Arc, time::Duration};
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use dioxus::hooks::{use_coroutine, Coroutine};
use futures_util::StreamExt;
use ringbuf::{Consumer, HeapRb, Producer, SharedRb};
use tokio_tungstenite_wasm::WebSocketStream;
use tokio_with_wasm::tokio::time::sleep;
pub 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();
let _sound_stream: Coroutine<()> = use_coroutine(|_| async move {
sound_stream(stream, producer).await;
});
tokio_with_wasm::tokio::time::sleep(Duration::from_secs(1)).await;
let _listen: Coroutine<()> = use_coroutine(|_| async move {
listen(consumer).await;
});
}
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 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 output_stream = output_device
.build_output_stream(&config, output_data_fn, err_fn, None)
.unwrap();
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();
// 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| {
// },
// None).unwrap();
// stream.play().unwrap();
// tokio::time::sleep(Duration::from_secs(10)).await;
// stream.pause().unwrap();
}
fn err_fn(err: cpal::StreamError) {
eprintln!("Something Happened: {}", err);
}

4
streamer/src/main.rs
View file

@ -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");
}
};