snippet of code to process stream of raw PCM audio and encode for opus broadcast

mumble_say.rs

pub fn say<'a>(vox_out_rx: futures::sync::mpsc::Receiver<Vec<u8>>,
               udp_tx: futures::sync::mpsc::Sender<Vec<u8>>,
               crypt_state: Arc<Mutex<ocbaes128::CryptState>>)
               -> impl Future<Item = (), Error = Error> + 'a {

    // Hz * channel * ms / 1000
    let sample_channels: u32 = 1;
    let sample_rate: u32 = 16000;
    let sample_ms: u32 = 10;
    let sample_size: u32 = sample_rate * sample_channels * sample_ms / 1000;

    println!("sample channels: {} rate: {} ms: {} size: {}",
             sample_channels,
             sample_rate,
             sample_ms,
             sample_size);

    let mut encoder =
        opus::Encoder::new(sample_rate, opus::Channels::Mono, opus::Application::Voip).unwrap();

    let mut sequence = chrono::UTC::now().timestamp() as u64;

    vox_out_rx.map(|segment| futures::stream::iter_ok::<_, ()>(segment))
        .flatten()
        .chunks(2)
        .and_then(|raw| {
            let pcm = (&raw[..]).read_i16::<LittleEndian>().unwrap();
            ok::<i16, ()>(pcm)
        })
        .chunks(sample_size as usize)
        .fold(udp_tx, move |udp_tx, chunk| {
            let mut chunk = Vec::from(chunk);
            chunk.resize(sample_size as usize, 0);
            let frame = encoder.encode_vec(&chunk[..], 4000).unwrap();

            sequence = sequence + 1;

            let done = false;

            let aud_header = 0b100 << 5;
            let mut data = Vec::<u8>::new();
            data.write_u8(aud_header).unwrap();
            data.write_varint(sequence).unwrap();
            let opus_len = if done {
                frame.len() as u64 | 0x2000
            } else {
                frame.len() as u64
            };
            data.write_varint(opus_len).unwrap();
            data.write_all(&frame).unwrap();

            let buf = {
                let mut crypt_state = crypt_state.lock().unwrap();
                crypt_state.encrypt(&data[..])
            };

            udp_tx.send(buf)
                .map_err(|_| ())
        })
        .map(|_| ())
        .map_err(|_| Error::new(ErrorKind::Other, "vox out"))
}