[package]

name = "reowolf_rs"

version = "0.1.4"

authors = [

	"Christopher Esterhuyse <esterhuy@cwi.nl, christopher.esterhuyse@gmail.com>",

	"Hans-Dieter Hiep <hdh@cwi.nl>"

]

edition = "2018"

[dependencies]

# convenience macros

maplit = "1.0.2"

derive_more = "0.99.2"

# runtime

serde = { version = "1.0.114", features = ["derive"] }

getrandom = "0.1.14" # tiny crate. used to guess controller-id

# network

# integer-encoding = "1.1.5"

# byteorder = "1.3.4"

mio = { version = "0.7.0", package = "mio", features = ["tcp", "os-poll"] }

# protocol

backtrace = "0.3"

[dev-dependencies]

# test-generator = "0.3.0"

crossbeam-utils = "0.7.2"

lazy_static = "1.4.0"

[lib]

# compile target: dynamically linked library using C ABI

crate-type = ["cdylib"]

[features]

default = ["ffi"]

ffi = [] # no feature dependencies

macro_rules! endptlog {

    ($logger:expr, $($arg:tt)*) => {{

    }};

}

macro_rules! log {

    ($logger:expr, $($arg:tt)*) => {{

        let _ = writeln!($logger.line_writer(), $($arg)*);

    }};

}

use super::*;

struct MonitoredReader<R: Read> {

    bytes: usize,

    r: R,

}

#[derive(Debug)]

enum TryRecyAnyError {

    Timeout,

    PollFailed,

    EndpointError { error: EndpointError, index: usize },

}

/////////////////////

impl Endpoint {

    pub(super) fn try_recv<T: serde::de::DeserializeOwned>(

        &mut self,

    ) -> Result<Option<T>, EndpointError> {

        use EndpointError::*;

        // populate inbox as much as possible

        'read_loop: loop {

            let res = self.stream.read_to_end(&mut self.inbox);

            endptlog!(logger, "Stream read to end {:?}", &res);

            match res {

                Err(e) if would_block(&e) => break 'read_loop,

                Ok(0) => break 'read_loop,

                Ok(_) => (),

                Err(_e) => return Err(BrokenEndpoint),

            }

        }

        let mut monitored = MonitoredReader::from(&self.inbox[..]);

            Ok(msg) => {

                let msg_size = monitored.bytes_read();

                self.inbox.drain(0..(msg_size.try_into().unwrap()));

                Ok(Some(msg))

            }

            Err(e) => match *e {

                bincode::ErrorKind::Io(k) if k.kind() == std::io::ErrorKind::UnexpectedEof => {

                    Ok(None)

                }

                _ => Err(MalformedMessage),

            },

        }

    }

    pub(super) fn send<T: serde::ser::Serialize>(&mut self, msg: &T) -> Result<(), EndpointError> {

    }

}

impl EndpointManager {

    pub(super) fn index_iter(&self) -> Range<usize> {

        0..self.num_endpoints()

    }

    pub(super) fn num_endpoints(&self) -> usize {

        self.endpoint_exts.len()

    }

    pub(super) fn send_to_comms(&mut self, index: usize, msg: &Msg) -> Result<(), SyncError> {

        let endpoint = &mut self.endpoint_exts[index].endpoint;

        endpoint.send(msg).map_err(|_| SyncError::BrokenEndpoint(index))

    }

    pub(super) fn send_to_setup(&mut self, index: usize, msg: &Msg) -> Result<(), ConnectError> {

        let endpoint = &mut self.endpoint_exts[index].endpoint;

        endpoint.send(msg).map_err(|err| {

            ConnectError::EndpointSetupError(endpoint.stream.local_addr().unwrap(), err)

        })

    }

    pub(super) fn send_to(&mut self, index: usize, msg: &Msg) -> Result<(), EndpointError> {

        self.endpoint_exts[index].endpoint.send(msg)

    }

    pub(super) fn try_recv_any_comms(

        f.debug_struct("Predicate")

            .field("Trues", &MySet(self, true))

            .field("Falses", &MySet(self, false))

            .finish()

    }

}

impl serde::Serialize for SerdeProtocolDescription {

    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>

    where

        S: serde::Serializer,

    {

        let inner: &ProtocolDescription = &self.0;

        inner.serialize(serializer)

    }

}

impl<'de> serde::Deserialize<'de> for SerdeProtocolDescription {

    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>

    where

        D: serde::Deserializer<'de>,

    {

        let inner: ProtocolDescription = ProtocolDescription::deserialize(deserializer)?;

        Ok(Self(Arc::new(inner)))

    }

}

use crate as reowolf;

use crossbeam_utils::thread::scope;

use reowolf::{

    error::*,

    EndpointPolarity::{Active, Passive},

    Polarity::{Getter, Putter},

    *,

};

use std::{fs::File, net::SocketAddr, path::Path, sync::Arc, time::Duration};

//////////////////////////////////////////

fn next_test_addr() -> SocketAddr {

    use std::{

        net::{Ipv4Addr, SocketAddrV4},

        sync::atomic::{AtomicU16, Ordering::SeqCst},

    };

    static TEST_PORT: AtomicU16 = AtomicU16::new(5_000);

    let port = TEST_PORT.fetch_add(1, SeqCst);

    SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into()

}

fn file_logged_connector(connector_id: ConnectorId, dir_path: &Path) -> Connector {

    let _ = std::fs::create_dir(dir_path); // we will check failure soon

    let path = dir_path.join(format!("cid_{:?}.txt", connector_id));

    let file = File::create(path).unwrap();

    let file_logger = Box::new(FileLogger::new(connector_id, file));

    Connector::new(file_logger, MINIMAL_PROTO.clone(), connector_id, 8)

}

static MINIMAL_PDL: &'static [u8] = b"

primitive together(in ia, in ib, out oa, out ob){

  while(true) synchronous() {

    if(fires(ia)) {

      put(oa, get(ia));

      put(ob, get(ib));

    }

  } 

}

";

lazy_static::lazy_static! {

    static ref MINIMAL_PROTO: Arc<ProtocolDescription> = {

        Arc::new(reowolf::ProtocolDescription::parse(MINIMAL_PDL).unwrap())

    };

}

lazy_static::lazy_static! {

    static ref TEST_MSG: Payload = {

        Payload::from(b"hello" as &[u8])

    };

}

//////////////////////////////////////////

#[test]

            let _ = c.new_net_port(Putter, sock_addr, Active).unwrap();

            c.connect(Some(Duration::from_secs(1))).unwrap();

            c.sync(Some(Duration::from_millis(300))).unwrap_err(); // timeout

        });

        s.spawn(|_| {

            // parent

            let mut c = file_logged_connector(999, test_log_path);

            let g = c.new_net_port(Getter, sock_addr, Passive).unwrap();

            c.connect(Some(Duration::from_secs(1))).unwrap();

            c.get(g).unwrap(); // not matched by put

            c.sync(None).unwrap_err(); // no timeout

        });

    })

    .unwrap();

}

#[test]

fn chain_connect() {

    let test_log_path = Path::new("./logs/chain_connect");

    let sock_addrs = [next_test_addr(), next_test_addr(), next_test_addr(), next_test_addr()];

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            c.new_net_port(Putter, sock_addrs[0], Passive).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(10, test_log_path);

            c.new_net_port(Getter, sock_addrs[0], Active).unwrap();

            c.new_net_port(Putter, sock_addrs[1], Passive).unwrap();

        });

        s.spawn(|_| {

            // LEADER

            let mut c = file_logged_connector(7, test_log_path);

            c.new_net_port(Getter, sock_addrs[1], Active).unwrap();

            c.new_net_port(Putter, sock_addrs[2], Passive).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(4, test_log_path);

            c.new_net_port(Getter, sock_addrs[2], Active).unwrap();

            c.new_net_port(Putter, sock_addrs[3], Passive).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            c.new_net_port(Getter, sock_addrs[3], Active).unwrap();

        });

    })

    .unwrap();

}

#[test]

fn net_self_loop() {

    let test_log_path = Path::new("./logs/net_self_loop");

    let sock_addr = next_test_addr();

    let mut c = file_logged_connector(0, test_log_path);

    let p = c.new_net_port(Putter, sock_addr, Active).unwrap();

    let g = c.new_net_port(Getter, sock_addr, Passive).unwrap();

    c.connect(Some(Duration::from_secs(1))).unwrap();

    c.put(p, TEST_MSG.clone()).unwrap();

    c.get(g).unwrap();

    c.sync(Some(Duration::from_millis(500))).unwrap();

}

#[test]

fn nobody_connects_active() {

    let test_log_path = Path::new("./logs/nobody_connects_active");

    let sock_addr = next_test_addr();

    let mut c = file_logged_connector(0, test_log_path);

    let _g = c.new_net_port(Getter, sock_addr, Active).unwrap();