Changeset - 4c844c3751d1
[Not reviewed]
0 2 0
Christopher Esterhuyse - 5 years ago 2020-05-27 13:20:02
christopher.esterhuyse@gmail.com
imposed a standard set of component defs. this approach is simple but temporary
2 files changed with 37 insertions and 40 deletions:
0 comments (0 inline, 0 general)
src/protocol/inputsource.rs
Show inline comments
 
use std::fmt;
 
use std::fs::File;
 
use std::io;
 
use std::path::Path;
 

	
 
use backtrace::Backtrace;
 

	
 
#[derive(Clone)]
 
pub struct InputSource {
 
    filename: String,
 
    input: Vec<u8>,
 
    line: usize,
 
    column: usize,
 
    offset: usize,
 
}
 

	
 
static STD_LIB_PDL: &'static [u8] = b"
 
primitive forward(in i, out o) {
 
    while(true) synchronous() put(o, get(i));
 
}
 
primitive sync(in i, out o) {
 
    while(true) synchronous() if(fires(i)) put(o, get(i));
 
}
 
primitive alternator_2(in i, out a, out b) {
 
    while(true) {
 
        synchronous() put(a, get(i));
 
        synchronous() put(b, get(i));
 
    }
 
}
 
primitive replicator_2(in i, out l, out r) {
 
    while(true) synchronous() if(fires(i)) {
 
        msg m = get(i);
 
        put(l, m);
 
        put(r, m);
 
    }
 
}
 
primitive merger_2(in l, in r, out o) {
 
    while(true) synchronous {
 
        if(fires(l)) put(o, get(l));
 
        else         put(o, get(r));
 
    }
 
}";
 

	
 
impl InputSource {
 
    // Constructors
 
    pub fn new<A: io::Read, S: ToString>(filename: S, reader: &mut A) -> io::Result<InputSource> {
 
        let mut vec = Vec::new();
 
        let mut vec = STD_LIB_PDL.to_vec();
 
        reader.read_to_end(&mut vec)?;
 
        Ok(InputSource {
 
            filename: filename.to_string(),
 
            input: vec,
 
            line: 1,
 
            column: 1,
 
            offset: 0,
 
        })
 
    }
 
    // Constructor helpers
 
    pub fn from_file(path: &Path) -> io::Result<InputSource> {
 
        let filename = path.file_name();
 
        match filename {
 
            Some(filename) => {
 
                let mut f = File::open(path)?;
 
                InputSource::new(filename.to_string_lossy(), &mut f)
 
            }
 
            None => Err(io::Error::new(io::ErrorKind::NotFound, "Invalid path")),
 
        }
 
    }
 
    pub fn from_string(string: &str) -> io::Result<InputSource> {
 
        let buffer = Box::new(string);
 
        let mut bytes = buffer.as_bytes();
 
        InputSource::new(String::new(), &mut bytes)
 
    }
 
    pub fn from_buffer(buffer: &[u8]) -> io::Result<InputSource> {
 
        InputSource::new(String::new(), &mut Box::new(buffer))
 
    }
 
    // Internal methods
 
    pub fn pos(&self) -> InputPosition {
 
        InputPosition { line: self.line, column: self.column, offset: self.offset }
 
    }
 
    pub fn error<S: ToString>(&self, message: S) -> ParseError {
 
        self.pos().parse_error(message)
 
    }
 
    pub fn is_eof(&self) -> bool {
 
        self.next() == None
 
    }
 
    pub fn next(&self) -> Option<u8> {
 
        if self.offset < self.input.len() {
 
            Some((*self.input)[self.offset])
 
        } else {
 
            None
 
        }
 
    }
 
    pub fn lookahead(&self, pos: usize) -> Option<u8> {
 
        if let Some(x) = usize::checked_add(self.offset, pos) {
 
            if x < self.input.len() {
 
                return Some((*self.input)[x]);
 
            }
 
        }
 
        None
 
    }
 
    pub fn consume(&mut self) {
 
        match self.next() {
 
            Some(x) if x == b'\r' && self.lookahead(1) != Some(b'\n') || x == b'\n' => {
 
                self.line += 1;
 
                self.offset += 1;
 
                self.column = 1;
 
            }
 
            Some(_) => {
 
                self.offset += 1;
 
                self.column += 1;
 
            }
 
            None => {}
 
        }
 
    }
 
}
 

	
 
impl fmt::Display for InputSource {
 
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
 
        self.pos().fmt(f)
 
    }
 
}
 

	
 
#[derive(Debug, Clone, Copy, Default)]
 
pub struct InputPosition {
 
    line: usize,
 
    column: usize,
 
    offset: usize,
 
}
 

	
 
impl InputPosition {
 
    fn context<'a>(&self, source: &'a InputSource) -> &'a [u8] {
 
        let start = self.offset - (self.column - 1);
 
        let mut end = self.offset;
 
        while end < source.input.len() {
 
            let cur = (*source.input)[end];
 
            if cur == b'\n' || cur == b'\r' {
 
                break;
 
            }
 
            end += 1;
 
        }
 
        &source.input[start..end]
 
    }
 
    fn parse_error<S: ToString>(&self, message: S) -> ParseError {
src/test/connector.rs
Show inline comments
 
extern crate test_generator;
 

	
 
use super::*;
 

	
 
use crate::common::*;
 
use crate::runtime::{errors::*, PortBinding::*};
 

	
 
static PDL: &[u8] = b"
 
primitive forward_once(in i, out o) {
 
    synchronous() put(o, get(i));
 
}
 
primitive blocked(in i, out o) {
 
    while(true) synchronous {}
 
}
 
primitive forward(in i, out o) {
 
    while(true) synchronous {
 
        put(o, get(i));
 
    }
 
}
 
primitive sync(in i, out o) {
 
    while(true) synchronous {
 
        if (fires(i)) put(o, get(i));
 
    }
 
}
 
primitive alternator_2(in i, out a, out b) {
 
    while(true) {
 
        synchronous { put(a, get(i)); }
 
        synchronous { put(b, get(i)); } 
 
    }
 
}
 
composite sync_2(in i, out o) {
 
    channel x -> y;
 
    new sync(i, x);
 
    new sync(y, o);
 
}
 
primitive exchange(in ai, out ao, in bi, out bo) {
 
    // Note the implicit causal relationship
 
    while(true) synchronous {
 
        if(fires(ai)) {
 
            put(bo, get(ai));
 
            put(ao, get(bi));
 
        }
 
    }
 
}
 
primitive filter(in i, out ok, out err) {
 
    while(true) synchronous {
 
        if (fires(i)) {
 
            msg m = get(i);
 
            if(m.length > 0) {
 
                put(ok, m);
 
            } else {
 
                put(err, m);
 
            } 
 
        }
 
    }
 
}
 
primitive token_spout(out o) {
 
    while(true) synchronous {
 
        put(o, create(0));
 
    }
 
}
 
primitive wait_n(int to_wait, out o) {
 
    while(to_wait > 0) synchronous() to_wait -= 1;
 
    synchronous { put(o, create(0)); }
 
}
 
composite wait_10(out o) {
 
    new wait_n(10, o);
 
}
 
primitive fifo_1(msg m, in i, out o) {
 
    while(true) synchronous {
 
        if (m == null && fires(i)) {
 
            m = get(i);
 
        } else if (m != null && fires(o)) {
 
            put(o, m);
 
            m = null;
 
        }
 
    }
 
}
 
composite fifo_1_e(in i, out o) {
 
    new fifo_1(null, i, o);
 
}
 
primitive samelen(in a, in b, out c) {
 
    synchronous {
 
        msg m = get(a);
 
        msg n = get(b);
 
        assert(m.length == n.length);
 
        put(c, m);
 
    }
 
}
 
primitive repl2(in a, out b, out c) {
 
    synchronous {
 
        msg m = get(a);
 
        put(b, m);
 
        put(c, m);
 
    }
 
}
 
composite samelen_repl(in a, out b) {
 
    channel c -> d;   
 
    channel e -> f;
 
    new samelen(a, f, c);
 
    new repl2(d, b, e);
 
    new replicator_2(d, b, e);
 
}
 
";
 

	
 
#[test]
 
fn connector_connects_ok() {
 
    // Test if we can connect natives using the given PDL
 
    /*
 
    Alice -->silence--P|A-->silence--> Bob
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    let addrs = [next_addr()];
 
    assert!(run_connector_set(&[
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"blocked").unwrap();
 
            x.configure(PDL, b"forward").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.bind_port(1, Passive(addrs[0])).unwrap();
 
            x.connect(timeout).unwrap();
 
        },
 
        &|x| {
 
            // Bob
 
            x.configure(PDL, b"blocked").unwrap();
 
            x.configure(PDL, b"forward").unwrap();
 
            x.bind_port(0, Active(addrs[0])).unwrap();
 
            x.bind_port(1, Native).unwrap();
 
            x.connect(timeout).unwrap();
 
        },
 
    ]));
 
}
 

	
 
#[test]
 
fn connector_connected_but_silent_natives() {
 
    // Test if we can connect natives and have a trivial sync round
 
    /*
 
    Alice -->silence--P|A-->silence--> Bob
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    let addrs = [next_addr()];
 
    assert!(run_connector_set(&[
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"blocked").unwrap();
 
            x.configure(PDL, b"sync").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.bind_port(1, Passive(addrs[0])).unwrap();
 
            x.connect(timeout).unwrap();
 
            assert_eq!(Ok(0), x.sync(timeout));
 
        },
 
        &|x| {
 
            // Bob
 
            x.configure(PDL, b"blocked").unwrap();
 
            x.configure(PDL, b"sync").unwrap();
 
            x.bind_port(0, Active(addrs[0])).unwrap();
 
            x.bind_port(1, Native).unwrap();
 
            x.connect(timeout).unwrap();
 
            assert_eq!(Ok(0), x.sync(timeout));
 
        },
 
    ]));
 
}
 

	
 
#[test]
 
fn connector_self_forward_ok() {
 
    // Test a deterministic system
 
    // where a native has no network bindings
 
    // and sends messages to itself
 
    /*
 
        /-->\
 
    Alice   forward
 
        \<--/
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    const N: usize = 5;
 
    static MSG: &[u8] = b"Echo!";
 
    assert!(run_connector_set(&[
 
        //
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"forward").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.bind_port(1, Native).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                x.put(0, MSG.to_vec().into()).unwrap();
 
                x.get(1).unwrap();
 
                assert_eq!(Ok(0), x.sync(timeout));
 
                assert_eq!(Ok(MSG), x.read_gotten(1));
 
            }
 
        },
 
    ]));
 
}
 
#[test]
 
fn connector_token_spout_ok() {
 
    // Test a deterministic system where the proto
 
    // creates token messages
 
    /*
 
    Alice<--token_spout
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    const N: usize = 5;
 
    assert!(run_connector_set(&[
 
        //
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"token_spout").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                x.get(0).unwrap();
 
                assert_eq!(Ok(0), x.sync(timeout));
 
                assert_eq!(Ok(&[] as &[u8]), x.read_gotten(0));
 
            }
 
        },
 
    ]));
 
}
 

	
 
#[test]
 
fn connector_waiter_ok() {
 
    // Test a stateful proto that blocks port 0 for 10 rounds
 
    // and then sends a single token on the 11th
 
    /*
 
    Alice<--token_spout
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    assert!(run_connector_set(&[
 
        //
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"wait_10").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _ in 0..10 {
 
                assert_eq!(Ok(0), x.sync(timeout));
 
                assert_eq!(Err(ReadGottenErr::DidNotGet), x.read_gotten(0));
 
            }
 
            x.get(0).unwrap();
 
            assert_eq!(Ok(0), x.sync(timeout));
 
            assert_eq!(Ok(&[] as &[u8]), x.read_gotten(0));
 
        },
 
    ]));
 
}
 

	
 
#[test]
 
fn connector_self_forward_timeout() {
 
    // Test a deterministic system
 
    // where a native has no network bindings
 
    // and sends messages to itself
 
    /*
 
        /-->\
 
    Alice   forward
 
        \<--/
 
    */
 
    let timeout = Duration::from_millis(500);
 
    static MSG: &[u8] = b"Echo!";
 
    assert!(run_connector_set(&[
 
        //
 
        &|x| {
 
            // Sender
 
            x.configure(PDL, b"forward").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.bind_port(1, Native).unwrap();
 
            x.connect(timeout).unwrap();
 
            x.put(0, MSG.to_vec().into()).unwrap();
 
            // native and forward components cannot find a solution
 
            assert_eq!(Err(SyncErr::Timeout), x.sync(timeout));
 
        },
 
    ]));
 
}
 

	
 
// TODO this test failed once? maybe a version error. Check on it.
 
#[test]
 
fn connector_forward_det() {
 
    // Test if a deterministic protocol and natives can pass one message
 
    /*
 
    Alice -->forward--P|A-->forward--> Bob
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    let addrs = [next_addr()];
 
    const N: usize = 5;
 
    static MSG: &[u8] = b"Hello!";
 

	
 
    assert!(run_connector_set(&[
 
        &|x| {
 
            x.configure(PDL, b"forward").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.bind_port(1, Passive(addrs[0])).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                x.put(0, MSG.to_vec().into()).unwrap();
 
                assert_eq!(Ok(0), x.sync(timeout));
 
            }
 
        },
 
        &|x| {
 
            x.configure(PDL, b"forward").unwrap();
 
            x.bind_port(0, Active(addrs[0])).unwrap();
 
            x.bind_port(1, Native).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                x.get(0).unwrap();
 
                assert_eq!(Ok(0), x.sync(timeout));
 
                assert_eq!(Ok(MSG), x.read_gotten(0));
 
            }
 
        },
 
    ]));
 
}
 

	
 
#[test]
 
fn connector_nondet_proto_det_natives() {
 
    // Test the use of a nondeterministic protocol
 
    // where Alice decides the choice and the others conform
 
    /*
 
    Alice -->sync--A|P-->sync--> Bob
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    let addrs = [next_addr()];
 
    const N: usize = 5;
 
    static MSG: &[u8] = b"Message, here!";
 
    assert!(run_connector_set(&[
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"sync").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.bind_port(1, Active(addrs[0])).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _i in 0..N {
 
                x.put(0, MSG.to_vec().into()).unwrap();
 
                assert_eq!(0, x.sync(timeout).unwrap());
 
            }
 
        },
 
        &|x| {
 
            // Bob
 
            x.configure(PDL, b"sync").unwrap();
 
            x.bind_port(0, Passive(addrs[0])).unwrap();
 
            x.bind_port(1, Native).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _i in 0..N {
 
                x.get(0).unwrap();
 
                assert_eq!(Ok(0), x.sync(timeout));
 
                assert_eq!(Ok(MSG), x.read_gotten(0));
 
            }
 
        },
 
    ]));
 
}
 

	
 
#[test]
 
fn connector_putter_determines() {
 
    // putter and getter
 
    /*
 
    Alice -->sync--A|P-->sync--> Bob
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    let addrs = [next_addr()];
 
    const N: usize = 3;
 
    static MSG: &[u8] = b"Hidey ho!";
 
    assert!(run_connector_set(&[
 
        //
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"sync").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.bind_port(1, Active(addrs[0])).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _i in 0..N {
 
                x.put(0, MSG.to_vec().into()).unwrap();
 
                assert_eq!(0, x.sync(timeout).unwrap());
 
            }
 
@@ -407,243 +377,243 @@ fn connector_getter_determines() {
 
            // Bob
 
            x.configure(PDL, b"sync").unwrap();
 
            x.bind_port(0, Passive(addrs[0])).unwrap();
 
            x.bind_port(1, Native).unwrap();
 
            x.connect(timeout).unwrap();
 

	
 
            for _i in 0..N {
 
                x.get(0).unwrap();
 
                assert_eq!(Ok(0), x.sync(timeout));
 
                assert_eq!(Ok(MSG), x.read_gotten(0));
 
            }
 
        },
 
    ]));
 
}
 

	
 
#[test]
 
fn connector_alternator_2() {
 
    // Test a deterministic system which
 
    // alternates sending Sender's messages to A or B
 
    /*                    /--|-->A
 
    Sender -->alternator_2
 
                          \--|-->B
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    let addrs = [next_addr(), next_addr()];
 
    const N: usize = 5;
 
    static MSG: &[u8] = b"message";
 
    assert!(run_connector_set(&[
 
        //
 
        &|x| {
 
            // Sender
 
            x.configure(PDL, b"alternator_2").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.bind_port(1, Passive(addrs[0])).unwrap();
 
            x.bind_port(2, Passive(addrs[1])).unwrap();
 
            x.connect(timeout).unwrap();
 

	
 
            for _ in 0..N {
 
                for _ in 0..2 {
 
                    x.put(0, MSG.to_vec().into()).unwrap();
 
                    assert_eq!(0, x.sync(timeout).unwrap());
 
                }
 
            }
 
        },
 
        &|x| {
 
            // A
 
            x.configure(PDL, b"sync").unwrap();
 
            x.bind_port(0, Active(addrs[0])).unwrap();
 
            x.bind_port(1, Native).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                // get msg round
 
                x.get(0).unwrap();
 
                assert_eq!(Ok(0), x.sync(timeout)); // GET ONE
 
                assert_eq!(Ok(MSG), x.read_gotten(0));
 

	
 
                // silent round
 
                assert_eq!(Ok(0), x.sync(timeout)); // MISS ONE
 
                assert_eq!(Err(ReadGottenErr::DidNotGet), x.read_gotten(0));
 
            }
 
        },
 
        &|x| {
 
            // B
 
            x.configure(PDL, b"sync").unwrap();
 
            x.bind_port(0, Active(addrs[1])).unwrap();
 
            x.bind_port(1, Native).unwrap();
 
            x.connect(timeout).unwrap();
 

	
 
            for _ in 0..N {
 
                // silent round
 
                assert_eq!(Ok(0), x.sync(timeout)); // MISS ONE
 
                assert_eq!(Err(ReadGottenErr::DidNotGet), x.read_gotten(0));
 

	
 
                // get msg round
 
                x.get(0).unwrap();
 
                assert_eq!(Ok(0), x.sync(timeout)); // GET ONE
 
                assert_eq!(Ok(MSG), x.read_gotten(0));
 
            }
 
        },
 
    ]));
 
}
 

	
 
#[test]
 
fn connector_composite_chain_a() {
 
    // Check if composition works. Forward messages through long chains
 
    /*
 
    Alice -->sync-->sync-->A|P-->sync--> Bob
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    let addrs = [next_addr(), next_addr()];
 
    const N: usize = 1;
 
    static MSG: &[u8] = b"SSS";
 
    assert!(run_connector_set(&[
 
        //
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"sync_2").unwrap();
 
            x.configure(PDL, b"sync").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.bind_port(1, Active(addrs[0])).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                x.put(0, MSG.to_vec().into()).unwrap();
 
                assert_eq!(0, x.sync(timeout).unwrap());
 
            }
 
        },
 
        &|x| {
 
            // Bob
 
            x.configure(PDL, b"forward").unwrap();
 
            x.bind_port(0, Passive(addrs[0])).unwrap();
 
            x.bind_port(1, Native).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                // get msg round
 
                x.get(0).unwrap();
 
                assert_eq!(Ok(0), x.sync(timeout));
 
                assert_eq!(Ok(MSG), x.read_gotten(0));
 
            }
 
        },
 
    ]));
 
}
 

	
 
#[test]
 
fn connector_composite_chain_b() {
 
    // Check if composition works. Forward messages through long chains
 
    /*
 
    Alice -->sync-->sync-->A|P-->sync-->sync--> Bob
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    let addrs = [next_addr(), next_addr()];
 
    const N: usize = 1;
 
    static MSG: &[u8] = b"SSS";
 
    assert!(run_connector_set(&[
 
        //
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"sync_2").unwrap();
 
            x.configure(PDL, b"sync").unwrap();
 
            x.bind_port(0, Native).unwrap();
 
            x.bind_port(1, Active(addrs[0])).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                x.put(0, MSG.to_vec().into()).unwrap();
 
                assert_eq!(0, x.sync(timeout).unwrap());
 
            }
 
        },
 
        &|x| {
 
            // Bob
 
            x.configure(PDL, b"sync_2").unwrap();
 
            x.configure(PDL, b"sync").unwrap();
 
            x.bind_port(0, Passive(addrs[0])).unwrap();
 
            x.bind_port(1, Native).unwrap();
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                // get msg round
 
                x.get(0).unwrap();
 
                assert_eq!(Ok(0), x.sync(timeout));
 
                assert_eq!(Ok(MSG), x.read_gotten(0));
 
            }
 
        },
 
    ]));
 
}
 

	
 
#[test]
 
fn connector_exchange() {
 
    /*
 
        /-->\      /-->P|A-->\      /-->\
 
    Alice   exchange         exchange   Bob
 
        \<--/      \<--P|A<--/      \<--/
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    let addrs = [next_addr(), next_addr()];
 
    const N: usize = 1;
 
    assert!(run_connector_set(&[
 
        //
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"exchange").unwrap();
 
            x.bind_port(0, Native).unwrap(); // native in
 
            x.bind_port(1, Native).unwrap(); // native out
 
            x.bind_port(2, Passive(addrs[0])).unwrap(); // peer out
 
            x.bind_port(3, Passive(addrs[1])).unwrap(); // peer in
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                assert_eq!(Ok(()), x.put(0, b"A->B".to_vec().into()));
 
                assert_eq!(Ok(()), x.get(1));
 
                assert_eq!(Ok(0), x.sync(timeout));
 
                assert_eq!(Ok(b"B->A" as &[u8]), x.read_gotten(1));
 
            }
 
        },
 
        &|x| {
 
            // Bob
 
            x.configure(PDL, b"exchange").unwrap();
 
            x.bind_port(0, Native).unwrap(); // native in
 
            x.bind_port(1, Native).unwrap(); // native out
 
            x.bind_port(2, Active(addrs[1])).unwrap(); // peer out
 
            x.bind_port(3, Active(addrs[0])).unwrap(); // peer in
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                assert_eq!(Ok(()), x.put(0, b"B->A".to_vec().into()));
 
                assert_eq!(Ok(()), x.get(1));
 
                assert_eq!(Ok(0), x.sync(timeout));
 
                assert_eq!(Ok(b"A->B" as &[u8]), x.read_gotten(1));
 
            }
 
        },
 
    ]));
 
}
 

	
 
#[test]
 
fn connector_both() {
 
    /* ------->   -----P|A---->   ------->
 
      / /--->\ \ / /---P|A-->\ \ / /--->\ \
 
    Alice    exchange       exchange     Bob
 
    */
 
    let timeout = Duration::from_millis(1_500);
 
    let addrs = [next_addr(), next_addr()];
 
    const N: usize = 1;
 
    assert!(run_connector_set(&[
 
        //
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"exchange").unwrap();
 
            x.bind_port(0, Native).unwrap(); // native in a
 
            x.bind_port(1, Passive(addrs[0])).unwrap(); // peer out a
 
            x.bind_port(2, Native).unwrap(); // native in b
 
            x.bind_port(3, Passive(addrs[1])).unwrap(); // peer out b
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                assert_eq!(Ok(()), x.put(0, b"one".to_vec().into()));
 
                assert_eq!(Ok(()), x.put(1, b"two".to_vec().into()));
 
                assert_eq!(Ok(0), x.sync(timeout));
 
            }
 
        },
 
        &|x| {
 
            // Alice
 
            x.configure(PDL, b"exchange").unwrap();
 
            x.bind_port(0, Active(addrs[0])).unwrap(); // peer in a
 
            x.bind_port(1, Native).unwrap(); // native out b
 
            x.bind_port(2, Active(addrs[1])).unwrap(); // peer in b
 
            x.bind_port(3, Native).unwrap(); // native out a
 
            x.connect(timeout).unwrap();
 
            for _ in 0..N {
 
                assert_eq!(Ok(()), x.get(0));
 
                assert_eq!(Ok(()), x.get(1));
 
                assert_eq!(Ok(0), x.sync(timeout));
 
                assert_eq!(Ok(b"one" as &[u8]), x.read_gotten(0));
0 comments (0 inline, 0 general)