ssip-client/tests/fifo_async_tests.rs

// Copyright (c) 2022 Laurent Pelecq
//
// Licensed under the Apache License, Version 2.0
// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. All files in the project carrying such notice may not be copied,
// modified, or distributed except according to those terms.

#[cfg(feature = "async-mio")]
use mio::{Events, Poll, Token};
#[cfg(feature = "async-mio")]
use std::{slice::Iter, time::Duration};

#[cfg(feature = "async-mio")]
use ssip_client::*;

#[cfg(feature = "async-mio")]
mod server;

#[cfg(feature = "async-mio")]
use server::Server;

#[cfg(feature = "async-mio")]
enum Answer {
    Str(&'static str),
    Int(i8),
}

#[cfg(feature = "async-mio")]
struct State<'a, 'b> {
    pub done: bool,
    pub countdown: usize,
    pub writable: bool,
    pub start_get: bool,
    pub iter_requests: Iter<'a, Request>,
    pub iter_answers: Iter<'b, Answer>,
}

#[cfg(feature = "async-mio")]
impl<'a, 'b> State<'a, 'b> {
    fn new(iter_requests: Iter<'a, Request>, iter_answers: Iter<'b, Answer>) -> Self {
        State {
            done: false,
            countdown: 50,
            writable: false,
            start_get: false,
            iter_requests,
            iter_answers,
        }
    }

    fn terminated(&self) -> bool {
        self.done || self.countdown == 0
    }

    fn must_send(&self) -> bool {
        self.writable && self.countdown > 0
    }

    fn next_request(&mut self) -> Option<&Request> {
        if self.start_get {
            self.iter_requests.next()
        } else {
            None
        }
    }

    fn assert_string(&mut self, val: &str) {
        match self.iter_answers.next() {
            Some(Answer::Str(expected_val)) => assert_eq!(expected_val, &val),
            Some(Answer::Int(expected_val)) => panic!(
                "expecting integer {} instead of string '{}'",
                expected_val, val
            ),
            None => panic!("no more answers"),
        }
    }

    fn assert_integer(&mut self, val: i8) {
        match self.iter_answers.next() {
            Some(Answer::Int(expected_val)) => assert_eq!(expected_val, &val),
            Some(Answer::Str(expected_val)) => panic!(
                "expecting string '{}' instead of integer {}",
                expected_val, val
            ),
            None => panic!("no more answers"),
        }
    }
}

#[test]
#[cfg(feature = "async-mio")]
fn basic_async_communication() -> ClientResult<()> {
    const COMMUNICATION: [(&str, &str); 5] = [
        (
            "SET self CLIENT_NAME test:test:main\r\n",
            "208 OK CLIENT NAME SET\r\n",
        ),
        ("SET self LANGUAGE en\r\n", "201 OK LANGUAGE SET\r\n"),
        ("STOP self\r\n", "210 OK STOPPED\r\n"),
        (
            "GET OUTPUT_MODULE\r\n",
            "251-espeak\r\n251 OK GET RETURNED\r\n",
        ),
        ("GET RATE\r\n", "251-10\r\n251 OK GET RETURNED\r\n"),
    ];

    let get_requests = vec![Request::GetOutputModule, Request::GetRate];
    let get_answers = vec![Answer::Str("espeak"), Answer::Int(10)];
    let mut state = State::new(get_requests.iter(), get_answers.iter());

    let socket_dir = tempfile::tempdir()?;
    let socket_path = socket_dir.path().join("basic_async_communication.socket");
    assert!(!socket_path.exists());
    let handle = Server::run(&socket_path, &COMMUNICATION);
    let mut poll = Poll::new()?;
    let mut events = Events::with_capacity(128);
    let mut client = AsyncClient::new(fifo::Builder::new().path(&socket_path).build()?);
    let input_token = Token(0);
    let output_token = Token(1);
    let timeout = Duration::new(0, 500 * 1000 * 1000 /* 500 ms */);
    client.register(&poll, input_token, output_token).unwrap();
    client.push(Request::SetName(ClientName::new("test", "test")));
    while !state.terminated() {
        if !state.writable || !client.has_next() {
            poll.poll(&mut events, Some(timeout))?;
        }
        state.countdown -= 1;
        for event in &events {
            let token = event.token();
            if token == input_token {
                match dbg!(client.receive_next()?) {
                    Response::ClientNameSet => {
                        client.push(Request::SetLanguage(ClientScope::Current, "en".to_string()))
                    }
                    Response::LanguageSet => client.push(Request::Stop(MessageScope::Last)),
                    Response::Stopped => state.start_get = true,
                    Response::GetString(val) => state.assert_string(&val),
                    Response::GetInteger(val) => state.assert_integer(val),
                    result => panic!("Unexpected response: {:?}", result),
                }
                if let Some(request) = state.next_request() {
                    client.push(request.clone());
                } else if state.start_get {
                    state.done = true; // No more get request
                }
            } else if token == output_token {
                state.writable = true;
            }
        }
        if state.must_send() {
            match client.send_next() {
                Ok(()) => (),
                Err(ClientError::NotReady) => state.writable = false,
                err => return err,
            }
        }
    }
    handle.join().unwrap().unwrap();
    assert!(state.countdown > 0);
    socket_dir.close()?;
    Ok(())
}