scaproust 0.3.2

// Copyright (c) 2015-2017 Contributors as noted in the AUTHORS file.
//
// 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.
// This file may not be copied, modified, or distributed except according to those terms.

use std::collections::HashMap;
use std::io;
use std::thread;
use std::sync::mpsc;

use mio_extras;

use super::*;
use transport::Transport;
use reactor;
use reactor::dispatcher;
use core::session::{Request, Reply};
use core::socket::{Protocol, ProtocolCtor};
use core;
use io_error::*;

#[doc(hidden)]
type ReplyReceiver = mpsc::Receiver<Reply>;

#[doc(hidden)]
struct RequestSender {
    req_tx: EventLoopRequestSender
}

impl RequestSender {
    fn new(tx: EventLoopRequestSender) -> RequestSender {
        RequestSender { req_tx: tx }
    }
    fn socket_sender(&self, socket_id: core::SocketId) -> socket::RequestSender {
        socket::RequestSender::new(self.req_tx.clone(), socket_id)
    }
    fn device_sender(&self, device_id: core::DeviceId) -> device::RequestSender {
        device::RequestSender::new(self.req_tx.clone(), device_id)
    }
    fn probe_sender(&self, probe_id: core::ProbeId) -> probe::RequestSender {
        probe::RequestSender::new(self.req_tx.clone(), probe_id)
    }
    fn send(&self, req: Request) -> io::Result<()> {
        self.req_tx.send(reactor::Request::Session(req)).map_err(from_send_error)
    }
}

/// Creates the session and starts the I/O thread.
#[derive(Default)]
pub struct SessionBuilder {
    transports: HashMap<String, Box<Transport + Send>, core::BuildIdHasher>
}

impl SessionBuilder {

    pub fn new() -> SessionBuilder {
        SessionBuilder {
            transports: HashMap::with_hasher(core::BuildIdHasher)
        }
    }

    pub fn with<T>(mut self, scheme: &str, transport: T)  -> SessionBuilder
    where T : Transport + Send + 'static {
        self.transports.insert(scheme.into(), Box::new(transport));
        self
    }

    pub fn build(self) -> io::Result<Session> {

        let (reply_tx, reply_rx) = mpsc::channel();
        let (request_tx, request_rx) = mio_extras::channel::channel();
        let session = Session::new(RequestSender::new(request_tx), reply_rx);

        thread::spawn(move || dispatcher::Dispatcher::dispatch(self.transports, request_rx, reply_tx));

        Ok(session)
    }
}

/// Creates sockets and devices.
pub struct Session {
    request_sender: RequestSender,
    reply_receiver: ReplyReceiver
}

impl Session {

    fn new(request_tx: RequestSender, reply_rx: ReplyReceiver) -> Session {
        Session {
            request_sender: request_tx,
            reply_receiver: reply_rx
        }
    }

/*****************************************************************************/
/*                                                                           */
/* Create socket                                                             */
/*                                                                           */
/*****************************************************************************/

    /// Creates a socket with the specified protocol, which in turn determines its exact semantics.
    /// See [the proto module](proto/index.html) for a list of built-in protocols.
    /// The newly created socket is initially not associated with any endpoints.
    /// In order to establish a message flow at least one endpoint has to be added to the socket 
    /// using [connect](struct.Socket.html#method.connect) and [bind](struct.Socket.html#method.bind) methods.
    pub fn create_socket<T>(&mut self) -> io::Result<socket::Socket>
    where T : Protocol + From<mpsc::Sender<core::socket::Reply>> + 'static
    {
        let protocol_ctor = Session::create_protocol_ctor::<T>();
        let request = Request::CreateSocket(protocol_ctor);

        self.call(request, |reply| self.on_create_socket_reply(reply))
    }

    fn create_protocol_ctor<T>() -> ProtocolCtor 
    where T : Protocol + From<mpsc::Sender<core::socket::Reply>> + 'static
    {
        Box::new(move |sender: mpsc::Sender<core::socket::Reply>| {
            Box::new(T::from(sender)) as Box<Protocol>
        })
    }

    fn on_create_socket_reply(&self, reply: Reply) -> io::Result<socket::Socket> {
        match reply {
            Reply::SocketCreated(id, rx) => {
                let sender = self.request_sender.socket_sender(id);
                let sock = socket::Socket::new(sender, rx);
                
                Ok(sock)
            },
            Reply::Err(e) => Err(e),
            _ => self.unexpected_reply()
        }
    }

/*****************************************************************************/
/*                                                                           */
/* Create device                                                             */
/*                                                                           */
/*****************************************************************************/

    /// Creates a loopback device that loops and sends any messages received from the socket back to itself.
    pub fn create_relay_device(&self, socket: socket::Socket) -> io::Result<Box<device::Device>> {
        Ok(Box::new(device::Relay::new(socket)))
    }

    /// Creates a bridge device to forward messages between two sockets. 
    /// It loops and sends any messages received from `left` to `right` and vice versa.
    pub fn create_bridge_device(&mut self, left: socket::Socket, right: socket::Socket) -> io::Result<Box<device::Device>> {
        let request = Request::CreateDevice(left.id(), right.id());

        self.call(request, |reply| self.on_create_device_reply(reply, left, right))
    }

    fn on_create_device_reply(&self, reply: Reply, left: socket::Socket, right: socket::Socket) -> io::Result<Box<device::Device>> {
        match reply {
            Reply::DeviceCreated(id, rx) => {
                let sender = self.request_sender.device_sender(id);
                let bridge = device::Bridge::new(sender, rx, left, right);
                
                Ok(Box::new(bridge))
            },
            Reply::Err(e) => Err(e),
            _ => self.unexpected_reply()
        }
    }

/*****************************************************************************/
/*                                                                           */
/* Create probe                                                              */
/*                                                                           */
/*****************************************************************************/

    /// Creates a probe for polling sockets
    pub fn create_probe(&mut self, poll_opts: Vec<core::PollReq>) -> io::Result<probe::Probe> {
        let request = Request::CreateProbe(poll_opts);

        self.call(request, |reply| self.on_create_probe_reply(reply))
    }

    fn on_create_probe_reply(&self, reply: Reply) -> io::Result<probe::Probe> {
        match reply {
            Reply::ProbeCreated(id, rx) => {
                let sender = self.request_sender.probe_sender(id);
                let probe = probe::Probe::new(sender, rx);

                Ok(probe)
            },
            Reply::Err(e) => Err(e),
            _ => self.unexpected_reply()
        }
    }

/*****************************************************************************/
/*                                                                           */
/* backend                                                                   */
/*                                                                           */
/*****************************************************************************/

    fn unexpected_reply<T>(&self) -> io::Result<T> {
        Err(other_io_error("unexpected reply"))
    }

    fn call<T, F : FnOnce(Reply) -> io::Result<T>>(&self, request: Request, process: F) -> io::Result<T> {
        self.execute_request(request).and_then(process)
    }

    fn execute_request(&self, request: Request) -> io::Result<Reply> {
        self.send_request(request).and_then(|_| self.recv_reply())
    }

    fn send_request(&self, request: Request) -> io::Result<()> {
        self.request_sender.send(request)
    }

    fn recv_reply(&self) -> io::Result<Reply> {
        self.reply_receiver.receive()
    }
}

impl Drop for Session {
    fn drop(&mut self) {
        let _ = self.send_request(Request::Shutdown);
        let _ = self.recv_reply();
    }
}