tcp audio

audio.rs

use fon::{stereo::Stereo32, Audio, Sink};
use pasts::{exec, wait};
use wavy::{Microphone, MicrophoneStream, Speakers, SpeakersSink};

const SAMPLE_RATE: f32 = 48_000.;

/// An event handled by the event loop.
enum Event<'a> {
    /// Speaker is ready to play more audio.
    Play(SpeakersSink<'a, Stereo32>),
    /// Microphone has recorded some audio.
    Record(MicrophoneStream<'a, Stereo32>),
}

/// Shared state between tasks on the thread.
struct State {
    /// Temporary buffer for holding real-time audio samples.
    buffer: Audio<Stereo32>,
}

impl State {
    /// Event loop.  Return false to stop program.
    fn event(&mut self, event: Event<'_>) {
        match event {
            Event::Play(mut speakers) => speakers.stream(self.buffer.drain()),
            Event::Record(microphone) => send(microphone),
        }
    }
}

fn send(microphone: MicrophoneStream<'_, Stereo32>) {
    use std::io::prelude::*;
    use std::net::TcpStream;

    let mut stream = TcpStream::connect("127.0.0.1:35254").unwrap();

    let data = Audio::with_frames(SAMPLE_RATE, microphone.collect::<Vec<Stereo32>>())
        .as_f32_slice()
        .iter()
        .map(|f| f.to_le_bytes().to_vec())
        .flatten()
        .collect::<Vec<u8>>();

    stream.write_all(&data).unwrap();
}

fn server() -> (impl std::future::Future, async_channel::Receiver<Vec<u8>>) {
    use std::io::Read;
    use std::net::TcpListener;
    let listener = TcpListener::bind("127.0.0.1:35254").unwrap();

    let (tx, rx) = async_channel::unbounded();

    let async_server = async move {
        let mut data = vec![];
        for stream in listener.incoming() {
            stream.unwrap().read_to_end(&mut data).unwrap();
            tx.send(data.to_vec()).await.unwrap();
            data.clear();
        }
    };
    (async_server, rx)
}

/// Program start.
fn main() {
    let op = std::env::args().nth(1);
    if let Some(op) = op {
        match op.as_str() {
            "send" => {
                // state is actually ignored when sending
                let mut state = State {
                    buffer: Audio::with_silence(SAMPLE_RATE, 0),
                };
                let mut microphone = Microphone::default();

                exec!(state.event(wait! {
                    Event::Record(microphone.record().await),
                }));
            }
            "recv" => {
                let mut state = State {
                    buffer: Audio::with_silence(SAMPLE_RATE, 0),
                };
                let mut speakers = Speakers::default();
                let (serv, rx) = server();

                std::thread::spawn(move || {
                    pasts::block_on(async move {
                        serv.await;
                    });
                });
                pasts::block_on(async move {
                    wait!({
                        loop {
                            state.event(Event::Play(speakers.play().await));
                            let data = rx.recv().await.unwrap();
                            let data: Vec<f32> = data
                                .chunks_exact(4)
                                .map(|array| std::convert::TryFrom::try_from(array).unwrap())
                                .map(f32::from_le_bytes)
                                .collect();
                            let mut audio: Audio<Stereo32> =
                                Audio::with_f32_buffer(SAMPLE_RATE, data);
                            state.buffer.extend(audio.drain());
                        }
                    });
                });
            }
            _ => (),
        }
    }
}