make a bunch of functions sync

4 files changed, 23 insertions, 39 deletions

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<S> Stream for IntoInterleavedSamples<S>
     }
 }
 
-struct FromStream<S: Stream> {
+struct FromStream<S> {
     stream: S,
-    next: Option<S::Item>,
+    underlying_exhausted: bool,
 }
 
-async fn from_stream<S>(mut stream: S) -> FromStream<S>
+fn from_stream<S>(mut stream: S) -> FromStream<S>
     where
         S: Stream + Unpin,
         S::Item: Frame {
-    let next = stream.next().await;
-    FromStream { stream, next }
+    FromStream { stream, underlying_exhausted: false }
 }
 
 impl<S> StreamingSignal for FromStream<S>
@@ -585,20 +583,23 @@ impl<S> StreamingSignal for FromStream<S>
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Frame> {
         let s = self.get_mut();
-        if s.next.is_none() {
+        if s.underlying_exhausted {
             return Poll::Ready(<Self::Frame as Frame>::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(<Self::Frame as Frame>::EQUILIBRIUM);
             }
             Poll::Pending => Poll::Pending,
         }
     }
 
     fn is_exhausted(&self) -> bool {
-        self.next.is_none()
+        self.underlying_exhausted
     }
 }
 
@@ -607,37 +608,19 @@ struct FromInterleavedSamplesStream<S, F>
     where
         F: Frame {
     stream: S,
-    next_frame: Option<F>,
     next_buf: Vec<F::Sample>,
     underlying_exhausted: bool,
 }
 
-async fn from_interleaved_samples_stream<S, F>(mut stream: S) -> FromInterleavedSamplesStream<S, F>
+fn from_interleaved_samples_stream<S, F>(stream: S) -> FromInterleavedSamplesStream<S, F>
     where
         S: Stream + Unpin,
         S::Item: Sample,
         F: Frame<Sample = S::Item> {
-    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<S, F> StreamingSignal for FromInterleavedSamplesStream<S, F>
                 }
                 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<S, F> StreamingSignal for FromInterleavedSamplesStream<S, F>
         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<ControlPacket<Serverbound>>,
         connection_info_sender: watch::Sender<Option<ConnectionInfo>>,
     ) -> 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,