From 0f225e518b6889f604cb440f14824b21ed49bf37 Mon Sep 17 00:00:00 2001 From: Eskil Q Date: Sat, 2 Jan 2021 11:24:59 +0100 Subject: make a bunch of functions sync --- mumd/src/audio.rs | 55 ++++++++++++++++++------------------------------- mumd/src/audio/input.rs | 1 - mumd/src/main.rs | 2 +- mumd/src/state.rs | 4 ++-- 4 files changed, 23 insertions(+), 39 deletions(-) (limited to 'mumd') diff --git a/mumd/src/audio.rs b/mumd/src/audio.rs index a8af82d..9c3c143 100644 --- a/mumd/src/audio.rs +++ b/mumd/src/audio.rs @@ -22,7 +22,6 @@ use futures::task::{Context, Poll}; use std::pin::Pin; use tokio_stream::StreamExt; use std::future::{Future}; -use std::mem; //TODO? move to mumlib pub const EVENT_SOUNDS: &[(&'static [u8], NotificationEvents)] = &[ @@ -88,7 +87,7 @@ pub struct Audio { } impl Audio { - pub async fn new(input_volume: f32, output_volume: f32) -> Self { + pub fn new(input_volume: f32, output_volume: f32) -> Self { let sample_rate = SampleRate(SAMPLE_RATE); let host = cpal::default_host(); @@ -205,7 +204,7 @@ impl Audio { 4, input_config.sample_rate.0, input_config.channels as usize, - StreamingSignalExt::into_interleaved_samples(from_interleaved_samples_stream::<_, f32>(sample_receiver).await)); //TODO attach a noise gate + StreamingSignalExt::into_interleaved_samples(from_interleaved_samples_stream::<_, f32>(sample_receiver))); //TODO attach a noise gate //TODO group frames correctly output_stream.play().unwrap(); @@ -564,17 +563,16 @@ impl Stream for IntoInterleavedSamples } } -struct FromStream { +struct FromStream { stream: S, - next: Option, + underlying_exhausted: bool, } -async fn from_stream(mut stream: S) -> FromStream +fn from_stream(mut stream: S) -> FromStream where S: Stream + Unpin, S::Item: Frame { - let next = stream.next().await; - FromStream { stream, next } + FromStream { stream, underlying_exhausted: false } } impl StreamingSignal for FromStream @@ -585,20 +583,23 @@ impl StreamingSignal for FromStream fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { let s = self.get_mut(); - if s.next.is_none() { + if s.underlying_exhausted { return Poll::Ready(::EQUILIBRIUM); } match S::poll_next(Pin::new(&mut s.stream), cx) { - Poll::Ready(val) => { - let ret = mem::replace(&mut s.next, val); - Poll::Ready(ret.unwrap()) + Poll::Ready(Some(val)) => { + Poll::Ready(val) + } + Poll::Ready(None) => { + s.underlying_exhausted = true; + return Poll::Ready(::EQUILIBRIUM); } Poll::Pending => Poll::Pending, } } fn is_exhausted(&self) -> bool { - self.next.is_none() + self.underlying_exhausted } } @@ -607,37 +608,19 @@ struct FromInterleavedSamplesStream where F: Frame { stream: S, - next_frame: Option, next_buf: Vec, underlying_exhausted: bool, } -async fn from_interleaved_samples_stream(mut stream: S) -> FromInterleavedSamplesStream +fn from_interleaved_samples_stream(stream: S) -> FromInterleavedSamplesStream where S: Stream + Unpin, S::Item: Sample, F: Frame { - let mut i = 0; - let mut buf = Vec::new(); - let next = loop { - if i == F::CHANNELS { - let mut iter = buf.into_iter(); - break F::from_samples(&mut iter); - } - match stream.next().await { - Some(v) => { - buf.push(v); - } - None => break None, - } - - i += 1; - }; FromInterleavedSamplesStream { stream, next_buf: Vec::new(), underlying_exhausted: false, - next_frame: next, } } @@ -660,7 +643,6 @@ impl StreamingSignal for FromInterleavedSamplesStream } Poll::Ready(None) => { s.underlying_exhausted = true; - s.next_frame = None; return Poll::Ready(F::EQUILIBRIUM); } Poll::Pending => return Poll::Pending, @@ -670,8 +652,11 @@ impl StreamingSignal for FromInterleavedSamplesStream let mut data = s.next_buf.iter().cloned(); let n = F::from_samples(&mut data).unwrap(); s.next_buf.clear(); - let ret = mem::replace(&mut s.next_frame, Some(n)).unwrap(); - Poll::Ready(ret) + Poll::Ready(n) + } + + fn is_exhausted(&self) -> bool { + self.underlying_exhausted } } diff --git a/mumd/src/audio/input.rs b/mumd/src/audio/input.rs index 8f0fe6e..febcb17 100644 --- a/mumd/src/audio/input.rs +++ b/mumd/src/audio/input.rs @@ -2,7 +2,6 @@ use cpal::{InputCallbackInfo, Sample}; use std::collections::VecDeque; use std::sync::{Arc, Mutex}; use tokio::sync::watch; -use log::*; use futures::Stream; use std::pin::Pin; use std::task::{Context, Poll}; diff --git a/mumd/src/main.rs b/mumd/src/main.rs index 4d6f148..db6d2ef 100644 --- a/mumd/src/main.rs +++ b/mumd/src/main.rs @@ -58,7 +58,7 @@ async fn main() { let (response_sender, response_receiver) = mpsc::unbounded_channel(); let (ping_request_sender, ping_request_receiver) = mpsc::unbounded_channel(); - let state = State::new(packet_sender, connection_info_sender).await; + let state = State::new(packet_sender, connection_info_sender); let state = Arc::new(Mutex::new(state)); let (_, _, _, e, _) = join!( diff --git a/mumd/src/state.rs b/mumd/src/state.rs index 1421691..85e5449 100644 --- a/mumd/src/state.rs +++ b/mumd/src/state.rs @@ -64,7 +64,7 @@ pub struct State { } impl State { - pub async fn new( + pub fn new( packet_sender: mpsc::UnboundedSender>, connection_info_sender: watch::Sender>, ) -> Self { @@ -72,7 +72,7 @@ impl State { let audio = Audio::new( config.audio.input_volume.unwrap_or(1.0), config.audio.output_volume.unwrap_or(1.0), - ).await; + ); let mut state = Self { config, server: None, -- cgit v1.2.1