#include <stdio.h>

#include <string.h>

#include "../../reowolf.h"

	char* pdl =

	"primitive forward(in i, out o) {"

	"	while(true) synchronous {"

	"		put(o, get(i));"

	"	}"

	"}"

	;

	char msg_buf[128];

	memset(msg_buf, 0, 128);

	printf("input a message to send:");

	check("fgets", fgets(msg_buf, 128-1, stdin) == NULL);

	int msg_len = strlen(msg_buf);

	msg_buf[msg_len-1] = 0;

	Connector* c = connector_new();

	check("config ", connector_configure(c, pdl, "forward"));

	check("bind 0 ", connector_bind_native(c, 0));

	check("bind 1 ", connector_bind_passive(c, 1, "127.0.0.1:7000"));

	int i;

	for (i = 0; i < 3; i++) {

		check("put ", connector_put(c, 0, msg_buf, msg_len));

	}

	connector_destroy(c);

	return 0;

}

#include <stdio.h>

#include "../../reowolf.h"

	// BOB

	Connector* c = connector_new();

	check("config ", connector_configure(c, pdl, "forward"));

	check("bind 0 ", connector_bind_active(c, 0, "127.0.0.1:7000"));

	check("bind 1 ", connector_bind_native(c, 1));

	int i;

	for (i = 0; i < 3; i++) {

		check("get ", connector_get(c, 0));

		int msg_len;

		const unsigned char * msg;

		check("read", connector_gotten(c, 0, &msg, &msg_len));

	}

	connector_destroy(c);

	return 0;

}

#!/bin/sh

LIB_PATH="../../target/release"

gcc -L $LIB_PATH -lreowolf_rs -Wl,-R$LIB_PATH bob.c -o bob

        &mut self,

        m_ctx: PolyPContext,

        protocol_description: &ProtocolD,

    ) -> Result<SyncRunResult, EndpointErr> {

        let to_run: Vec<_> = self.incomplete.drain().collect();

        self.poly_run_these_branches(m_ctx, protocol_description, to_run)

    }

    pub(crate) fn poly_run_these_branches(

        &mut self,

        mut m_ctx: PolyPContext,

        protocol_description: &ProtocolD,

        mut to_run: Vec<(Predicate, BranchP)>,

    ) -> Result<SyncRunResult, EndpointErr> {

        use SyncRunResult as Srr;

        log!(&mut m_ctx.inner.logger, "~ Running branches for PolyP {:?}!", m_ctx.my_subtree_id,);

        'to_run_loop: while let Some((mut predicate, mut branch)) = to_run.pop() {

            let mut r_ctx = BranchPContext {

                m_ctx: m_ctx.reborrow(),

                ekeys: &self.ekeys,

                predicate: &predicate,

                inbox: &branch.inbox,

            };

            use PolyBlocker as Sb;

            let blocker = branch.state.sync_run(&mut r_ctx, protocol_description);

            log!(

                &mut r_ctx.m_ctx.inner.logger,

                "~ ... ran PolyP {:?} with branch pred {:?} to blocker {:?}",

                r_ctx.m_ctx.my_subtree_id,

                &predicate,

                &blocker

            );

            match blocker {

                Sb::Inconsistent => {} // DROP

                Sb::CouldntReadMsg(ekey) => {

                    assert!(self.ekeys.contains(&ekey));

                    let channel_id =

                        r_ctx.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;

                    log!(

                        &mut r_ctx.m_ctx.inner.logger,

                        "~ ... {:?} couldnt read msg for port {:?}. has inbox {:?}",

                        r_ctx.m_ctx.my_subtree_id,

                        channel_id,

                        &branch.inbox,

                    );

                    if predicate.replace_assignment(channel_id, true) != Some(false) {

                        // don't rerun now. Rerun at next `sync_run`

                        log!(&mut m_ctx.inner.logger, "~ ... Delay {:?}", m_ctx.my_subtree_id,);

                        branch.blocking_on = Some(ekey);

                        self.incomplete.insert(predicate, branch);

                    } else {

                        log!(&mut m_ctx.inner.logger, "~ ... Drop {:?}", m_ctx.my_subtree_id,);

                    }

                    // ELSE DROP

                }

                Sb::CouldntCheckFiring(ekey) => {

                    assert!(self.ekeys.contains(&ekey));

                    let channel_id =

                        r_ctx.m_ctx.inner.endpoint_exts.get(ekey).unwrap().info.channel_id;

                    // split the branch!

                    let branch_f = branch.clone();

                    let mut predicate_f = predicate.clone();

                    if predicate_f.replace_assignment(channel_id, false).is_some() {

                        panic!("OI HANS QUERY FIRST!");

                    }

                    assert!(predicate.replace_assignment(channel_id, true).is_none());

                    to_run.push((predicate, branch));

                    to_run.push((predicate_f, branch_f));

                }

                Sb::SyncBlockEnd => {

                    let ControllerInner { logger, endpoint_exts, .. } = m_ctx.inner;

                    log!(

                        logger,

                        "~ ... ran {:?} reached SyncBlockEnd with pred {:?} ...",

                        m_ctx.my_subtree_id,

                        &predicate,

                    );

                    // come up with the predicate for this local solution

                    for ekey in self.ekeys.iter() {

                        let channel_id = endpoint_exts.get(*ekey).unwrap().info.channel_id;

                        let fired =

                            branch.inbox.contains_key(ekey) || branch.outbox.contains_key(ekey);

                        match predicate.query(channel_id) {

                            Some(true) => {

                                if !fired {

                                    // This branch should have fired but didn't!

                                    log!(

                                        logger,

                                        "~ ... ... should have fired {:?} and didn't! pruning!",

                                        channel_id,

                                    );

                                    continue 'to_run_loop;

                                }

                            }

                            None => {

                                predicate.replace_assignment(channel_id, false);

                            }

                        }

                    }

                    log!(logger, "~ ... ... and finished just fine!",);

                    m_ctx.solution_storage.submit_and_digest_subtree_solution(

                        &mut m_ctx.inner.logger,

                        m_ctx.my_subtree_id,

                        predicate.clone(),

                    );

                    self.complete.insert(predicate, branch);

                }

                Sb::PutMsg(ekey, payload) => {

                    assert!(self.ekeys.contains(&ekey));

                    let EndpointExt { info, endpoint } =

                        m_ctx.inner.endpoint_exts.get_mut(ekey).unwrap();

                    if predicate.replace_assignment(info.channel_id, true) != Some(false) {

                        branch.outbox.insert(ekey, payload.clone());

                        let msg = CommMsgContents::SendPayload {

                            payload_predicate: predicate.clone(),

                            payload,

                        }

                        .into_msg(m_ctx.inner.round_index);

                        log!(

                            &mut m_ctx.inner.logger,

                            "~ ... ... PolyP sending msg {:?} to {:?} ({:?}) now!",

                            &msg,

                            ekey,

                            (info.channel_id.controller_id, info.channel_id.channel_index),

                        );

                        endpoint.send(msg)?;

                        to_run.push((predicate, branch));

                    }

                    // ELSE DROP

                }

            }

        }

        // all in self.incomplete most recently returned Blocker::CouldntReadMsg

        Ok(if self.incomplete.is_empty() {

            if self.complete.is_empty() {

                Srr::NoBranches

            } else {

                Srr::AllBranchesComplete

            }

        } else {

            Srr::BlockingForRecv

        })

    }

    pub(crate) fn poly_recv_run(

        &mut self,

        m_ctx: PolyPContext,

        protocol_description: &ProtocolD,

        ekey: Key,

        payload_predicate: Predicate,

        payload: Payload,

    ) -> Result<SyncRunResult, EndpointErr> {

        // try exact match

        let to_run = if self.complete.contains_key(&payload_predicate) {

            // exact match with stopped machine

            log!(

                &mut m_ctx.inner.logger,

                "... poly_recv_run matched stopped machine exactly! nothing to do here",

            );

            vec![]

        } else if let Some(mut branch) = self.incomplete.remove(&payload_predicate) {

            // exact match with running machine

            log!(

                &mut m_ctx.inner.logger,

                "... poly_recv_run matched running machine exactly! pred is {:?}",

                &payload_predicate

            );

            branch.inbox.insert(ekey, payload);

            if branch.blocking_on == Some(ekey) {

                branch.blocking_on = None;

                vec![(payload_predicate, branch)]

            } else {

                vec![]

            }

        } else {

            log!(

                &mut m_ctx.inner.logger,

                "... poly_recv_run didn't have any exact matches... Let's try feed it to all branches",

            );

            let mut incomplete2 = HashMap::<_, _>::default();

            let to_run = self

                .incomplete

                .drain()

                .filter_map(|(old_predicate, mut branch)| {

                    use CommonSatResult as Csr;

                    match old_predicate.common_satisfier(&payload_predicate) {

                        Csr::FormerNotLatter | Csr::Equivalent => {

                            log!(

                                &mut m_ctx.inner.logger,

            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.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()).unwrap();

                assert_eq!(0, x.sync(timeout).unwrap());

            }

        },

        &|x| {

            // Bob

            x.configure(PDL, b"sync_2").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()));

                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()));

                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_routing_filter() {

    // Make a protocol whose behavior is a function of the contents of

    // a message. Here, the putter determines what is sent, and the proto

    // determines how it is routed

    /*

    Sender -->filter-->P|A-->sync--> Receiver

    */

    let timeout = Duration::from_millis(3_000);

    let addrs = [next_addr()];

    const N: usize = 1;

    assert!(run_connector_set(&[

        //

        &|x| {

            // Sender

            x.configure(PDL, b"filter").unwrap();

            x.bind_port(0, Native).unwrap();

            x.bind_port(1, Passive(addrs[0])).unwrap();

            x.bind_port(2, Native).unwrap(); // err channel

            x.connect(timeout).unwrap();

            for i in (0..3).cycle().take(N) {

                // messages cycle [], [4], [4,4], ...

                let msg: Payload = std::iter::repeat(4).take(i).collect();

                // batch 0: passes through filter!

                x.put(0, msg.clone()).unwrap();

                x.next_batch().unwrap();

                // batch 1: gets returned!

                x.put(0, msg.clone()).unwrap();

                x.get(1).unwrap();

                match x.sync(timeout).unwrap() {

                    0 => assert_ne!(msg.len(), 0), // ok

                    1 => assert_eq!(msg.len(), 0), // err

                    _ => unreachable!(),

                }

            }

        },

        &|x| {

            // Receiver

            x.configure(PDL, b"sync").unwrap();

            x.bind_port(0, Active(addrs[0])).unwrap();

            x.bind_port(1, Native).unwrap();