buttplug 0.2.0

// Buttplug Rust Source Code File - See https://buttplug.io for more info.
//
// Copyright 2016-2019 Nonpolynomial Labs LLC. All rights reserved.
//
// Licensed under the BSD 3-Clause license. See LICENSE file in the project root
// for full license information.

//! Handling of communication with Buttplug Server.
pub mod messagesorter;
#[cfg(any(feature = "client-ws", feature = "client-ws-ssl"))]
pub mod websocket;

#[cfg(feature = "server")]
use crate::server::{ButtplugInProcessServerWrapper, ButtplugServerWrapper, ButtplugServer};
use crate::{
    core::{
        messages::{ButtplugClientInMessage, ButtplugClientOutMessage, create_message_validator, ButtplugSpecV2OutMessage},
        errors::{ButtplugError, ButtplugMessageError}
    },
    util::{
        future::{
            ButtplugFuture, ButtplugFutureState, ButtplugFutureStateShared, ButtplugMessageStateShared,
        }
    },
};
use async_std::sync::{channel, Receiver};
#[cfg(feature = "serialize_json")]
use async_std::{
    prelude::{FutureExt, StreamExt},
    sync::Sender,
};
use async_trait::async_trait;
#[cfg(feature = "serialize_json")]
use futures::future::Future;
#[cfg(feature = "serialize_json")]
use messagesorter::ClientConnectorMessageSorter;
use std::{error::Error, fmt};

pub type ButtplugClientConnectionState =
    ButtplugFutureState<Result<(), ButtplugClientConnectorError>>;
pub type ButtplugClientConnectionStateShared =
    ButtplugFutureStateShared<Result<(), ButtplugClientConnectorError>>;
pub type ButtplugClientConnectionFuture = ButtplugFuture<Result<(), ButtplugClientConnectorError>>;

#[derive(Debug, Clone)]
pub struct ButtplugClientConnectorError {
    pub message: String,
}

impl ButtplugClientConnectorError {
    pub fn new(msg: &str) -> Self {
        Self {
            message: msg.to_owned(),
        }
    }
}

impl fmt::Display for ButtplugClientConnectorError {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(f, "Init Error: {}", self.message)
    }
}

impl Error for ButtplugClientConnectorError {
    fn description(&self) -> &str {
        self.message.as_str()
    }

    fn source(&self) -> Option<&(dyn Error + 'static)> {
        None
    }
}

// Not real sure if this is sync, since there may be state that could get weird
// in connectors implementing this trait, but Send should be ok.
#[async_trait]
pub trait ButtplugClientConnector: Send {
    async fn connect(&mut self) -> Result<(), ButtplugClientConnectorError>;
    async fn disconnect(&mut self) -> Result<(), ButtplugClientConnectorError>;
    async fn send(&mut self, msg: ButtplugClientInMessage, state: &ButtplugMessageStateShared);
    fn get_event_receiver(&mut self) -> Receiver<ButtplugClientOutMessage>;
}

#[cfg(feature = "server")]
pub struct ButtplugEmbeddedClientConnector {
    server: ButtplugInProcessServerWrapper,
    recv: Option<Receiver<ButtplugClientOutMessage>>,
}

#[cfg(feature = "server")]
impl<'a> ButtplugEmbeddedClientConnector{
    pub fn new(name: &str, max_ping_time: u128) -> Self {
        let (server, recv) = ButtplugInProcessServerWrapper::new(&name, max_ping_time);
        Self {
            recv: Some(recv),
            server
        }
    }

    pub fn server_ref(&'a mut self) -> &'a mut ButtplugServer
    {
        self.server.server_ref()
    }
}

#[cfg(feature = "server")]
#[async_trait]
impl ButtplugClientConnector for ButtplugEmbeddedClientConnector {
    async fn connect(&mut self) -> Result<(), ButtplugClientConnectorError> {
        Ok(())
    }

    async fn disconnect(&mut self) -> Result<(), ButtplugClientConnectorError> {
        Ok(())
    }

    async fn send(&mut self, msg: ButtplugClientInMessage, state: &ButtplugMessageStateShared) {
        let ret_msg = self.server.parse_message(msg).await;
        let mut waker_state = state.lock().unwrap();
        waker_state.set_reply(ret_msg);
    }

    fn get_event_receiver(&mut self) -> Receiver<ButtplugClientOutMessage> {
        // This will panic if we've already taken the receiver.
        self.recv.take().unwrap()
    }
}

// The embedded connector is used heavily in the client unit tests, so we can
// assume code coverage there and omit specific tests here.

pub trait ButtplugRemoteClientConnectorSender: Sync + Send {
    fn send(&self, msg: ButtplugClientInMessage);
    fn close(&self);
}

pub enum ButtplugRemoteClientConnectorMessage {
    Sender(Box<dyn ButtplugRemoteClientConnectorSender>),
    Connected(),
    Text(String),
    Error(String),
    ClientClose(String),
    Close(String),
}

#[cfg(feature = "serialize_json")]
pub struct ButtplugRemoteClientConnectorHelper {
    // Channel send/recv pair for applications wanting to send out through the
    // remote connection. Receiver will be send to task on creation.
    internal_send: Sender<(ButtplugClientInMessage, ButtplugMessageStateShared)>,
    internal_recv: Option<Receiver<(ButtplugClientInMessage, ButtplugMessageStateShared)>>,
    // Channel send/recv pair for remote connection sending information to the
    // application. Receiver will be sent to task on creation.
    remote_send: Sender<ButtplugRemoteClientConnectorMessage>,
    remote_recv: Option<Receiver<ButtplugRemoteClientConnectorMessage>>,
    event_send: Option<Sender<ButtplugClientOutMessage>>,
}

#[cfg(feature = "serialize_json")]
unsafe impl Send for ButtplugRemoteClientConnectorHelper {}
#[cfg(feature = "serialize_json")]
unsafe impl Sync for ButtplugRemoteClientConnectorHelper {}

#[cfg(feature = "serialize_json")]
impl ButtplugRemoteClientConnectorHelper {
    pub fn new(event_sender: Sender<ButtplugClientOutMessage>) -> Self {
        let (internal_send, internal_recv) = channel::<(ButtplugClientInMessage, ButtplugMessageStateShared)>(256);
        let (remote_send, remote_recv) = channel(256);
        Self {
            event_send: Some(event_sender),
            remote_send,
            remote_recv: Some(remote_recv),
            internal_send,
            internal_recv: Some(internal_recv),
        }
    }

    pub fn get_remote_send(&self) -> Sender<ButtplugRemoteClientConnectorMessage> {
        self.remote_send.clone()
    }

    pub async fn send(&mut self, msg: &ButtplugClientInMessage, state: &ButtplugMessageStateShared) {
        self.internal_send.send((msg.clone(), state.clone())).await;
    }

    pub async fn close(&self) {
        // Emulate a close from the connector side, which will cause us to
        // close.
        self.remote_send
            .send(ButtplugRemoteClientConnectorMessage::Close(
                "Client requested close.".to_owned(),
            ))
            .await;
    }

    pub fn get_recv_future(&mut self) -> impl Future {
        // Set up a way to get futures in and out of the sorter, which will live
        // in our connector task.
        let event_send = self.event_send.take().unwrap();

        // Remove the receivers we need to move into the task.
        let mut remote_recv = self.remote_recv.take().unwrap();
        let mut internal_recv = self.internal_recv.take().unwrap();
        let message_validator = create_message_validator();
        async move {
            let mut sorter = ClientConnectorMessageSorter::default();
            // Our in-task remote sender, which is a wrapped version of whatever
            // bus specific sender (websocket, tcp, etc) we'll be using.
            let mut remote_send: Option<Box<dyn ButtplugRemoteClientConnectorSender>> = None;

            enum StreamValue {
                NoValue,
                Incoming(ButtplugRemoteClientConnectorMessage),
                Outgoing((ButtplugClientInMessage, ButtplugMessageStateShared)),
            }

            loop {
                // We use two Options instead of an enum because we may never
                // get anything.
                let mut stream_return: StreamValue = async {
                    match remote_recv.next().await {
                        Some(msg) => StreamValue::Incoming(msg),
                        None => StreamValue::NoValue,
                    }
                }
                .race(async {
                    match internal_recv.next().await {
                        Some(msg) => StreamValue::Outgoing(msg),
                        None => StreamValue::NoValue,
                    }
                })
                .await;
                match stream_return {
                    StreamValue::NoValue => break,
                    StreamValue::Incoming(remote_msg) => {
                        match remote_msg {
                            ButtplugRemoteClientConnectorMessage::Sender(s) => {
                                remote_send = Some(s);
                            }
                            ButtplugRemoteClientConnectorMessage::Text(t) => {
                                match message_validator.validate(&t.clone()) {
                                    Ok(_) => {
                                        let array: Vec<ButtplugClientOutMessage> =
                                           serde_json::from_str(&t.clone()).unwrap();
                                        for smsg in array {
                                            if !sorter.maybe_resolve_message(&smsg) {
                                                debug!("Sending event!");
                                                // Send notification through event channel
                                                event_send.send(smsg).await;
                                            }
                                        }
                                    },
                                    Err(e) => {
                                        let error_str = format!("Got invalid messages from remote Buttplug Server: {:?}", e);
                                        error!("{}", error_str);
                                        event_send.send(ButtplugSpecV2OutMessage::Error(ButtplugError::ButtplugMessageError(ButtplugMessageError::new(&error_str)).into())).await;
                                    }
                                }

                            }
                            ButtplugRemoteClientConnectorMessage::ClientClose(s) => {
                                info!("Client closing connection {}", s);
                                if let Some(ref mut remote_sender) = remote_send {
                                    remote_sender.close();
                                } else {
                                    panic!("Can't send message yet!");
                                }
                            }
                            ButtplugRemoteClientConnectorMessage::Close(s) => {
                                info!("Connector closing connection {}", s);
                                break;
                            }
                            _ => {
                                panic!("UNHANDLED BRANCH");
                            }
                        }
                    }
                    StreamValue::Outgoing(ref mut buttplug_fut_msg) => {
                        // Create future sets our message ID, so make sure this
                        // happens before we send out the message.
                        sorter.register_future(&mut buttplug_fut_msg.0, &buttplug_fut_msg.1);
                        if let Some(ref mut remote_sender) = remote_send {
                            remote_sender.send(buttplug_fut_msg.0.clone());
                        } else {
                            panic!("Can't send message yet!");
                        }
                    }
                }
            }
        }
    }
}