while let Some((getter, send_payload_msg)) = rctx.getter_buffer.pop() {

                assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));

                let route = cu.port_info.routes.get(&getter);

                log!(

                    cu.logger,

                    getter,

                );

                    Some(Route::UdpEndpoint { index }) => {

                    }

                    Some(Route::NetEndpoint { index }) => {

                        let msg = Msg::CommMsg(CommMsg {

                ee.incoming_round_spec_var = None; // shouldn't be accessed before its overwritten next round; still adding for clarity.

                for (payload_predicate, payload) in ee.outgoing_payloads.drain() {

                    if payload_predicate.assigns_subset(solution_predicate) {

                        log!(

                            logger,

                            "Sent payload {:?} with pred {:?} through Udp endpoint {}",

#[derive(Debug)]

pub enum ConnectError {

    BindFailed(SocketAddr),

    PollInitFailed,

    Timeout,

    PollFailed,

}

#[derive(Debug, Clone)]

struct NetEndpointSetup {

    sock_addr: SocketAddr,

    endpoint_polarity: EndpointPolarity,

}

#[derive(Debug, Clone)]

struct UdpEndpointSetup {

    local_addr: SocketAddr,

    peer_addr: SocketAddr,

}

}

#[derive(Debug)]

struct ConnectorSetup {

    surplus_sockets: u16,

}

#[derive(Debug)]

    to_put: HashMap<PortId, Payload>,

    to_get: HashSet<PortId>,

}

enum TokenTarget {

    NetEndpoint { index: usize },

    UdpEndpoint { index: usize },

        match phased {

            ConnectorPhased::Communication(..) => Err(WrongStateError),

            ConnectorPhased::Setup(setup) => {

                let udp_index = setup.udp_endpoint_setups.len();

                let [port_nat, port_udp] =

                    [cu.id_manager.new_port_id(), cu.id_manager.new_port_id()];

                cu.port_info.routes.insert(port_udp, Route::UdpEndpoint { index: udp_index });

                cu.port_info.polarities.insert(port_nat, polarity);

                cu.port_info.polarities.insert(port_udp, !polarity);

                Ok(port_nat)

            }

        }

        match phased {

            ConnectorPhased::Communication(..) => Err(WrongStateError),

            ConnectorPhased::Setup(setup) => {

                // {polarity, route} known. {peer} unknown.

                log!(

                    cu.logger,

                    polarity,

                );

            }

        }

    }

}

fn new_endpoint_manager(

    logger: &mut dyn Logger,

    port_info: &mut PortInfo,

    deadline: &Option<Instant>,

) -> Result<EndpointManager, ConnectError> {

    struct Todo {

        todo_endpoint: TodoEndpoint,

        endpoint_setup: NetEndpointSetup,

        sent_local_port: bool,          // true <-> I've sent my local port

        recv_peer_port: Option<PortId>, // Some(..) <-> I've received my peer's port

    }

    let mut net_todos = net_endpoint_setups

        .iter()

        .enumerate()

            let token = TokenTarget::NetEndpoint { index }.into();

            let todo_endpoint = if let EndpointPolarity::Active = endpoint_setup.endpoint_polarity {

                let mut stream = TcpStream::connect(endpoint_setup.sock_addr)

            };

            Ok(Todo {

                todo_endpoint,

                sent_local_port: false,

                recv_peer_port: None,

                endpoint_setup: endpoint_setup.clone(),

    let udp_todos = udp_endpoint_setups

        .iter()

        .enumerate()

            let mut sock = UdpSocket::bind(endpoint_setup.local_addr)

                .map_err(|_| Ce::BindFailed(endpoint_setup.local_addr))?;

            poll.registry()

                .register(&mut sock, TokenTarget::UdpEndpoint { index }.into(), Interest::WRITABLE)

                .unwrap();

        })

        .collect::<Result<Vec<UdpTodo>, ConnectError>>()?;

                            continue;

                        }

                        let local_polarity =

                        if event.is_writable() && !net_todo.sent_local_port {

                            // can write and didn't send setup msg yet? Do so!

                            let msg = Msg::SetupMsg(SetupMsg::MyPortInfo(MyPortInfo {

                                polarity: local_polarity,

                            }));

                            net_endpoint

                                .send(&msg)

                                    );

                                    if peer_info.polarity == local_polarity {

                                        return Err(ConnectError::PortPeerPolarityMismatch(

                                        ));

                                    }

                                    net_todo.recv_peer_port = Some(peer_info.port);

                                    // 1. finally learned the peer of this port!

                                    // 2. learned the info of this peer port

                                    port_info.polarities.insert(peer_info.port, peer_info.polarity);

                                    if let Some(route) = port_info.routes.get(&peer_info.port) {

                                        // check just for logging purposes

                                        log!(

    let net_endpoint_exts = net_todos

        .into_iter()

        .enumerate()

            net_endpoint: match todo_endpoint {

                TodoEndpoint::NetEndpoint(mut net_endpoint) => {

                    let token = TokenTarget::NetEndpoint { index }.into();

                }

                _ => unreachable!(),

            },

        })

        .collect();

    let udp_endpoint_exts = udp_todos

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

    };

}

lazy_static::lazy_static! {

    static ref TEST_MSG: Payload = {

    };

}

fn new_net_port() {

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

    let mut c = file_logged_connector(0, test_log_path);

}

#[test]

#[test]

fn single_node_connect() {

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

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(SEC1).unwrap();

}

#[test]

fn minimal_net_connect() {

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

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            c.connect(SEC1).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            c.connect(SEC1).unwrap();

        });

    })

#[test]

fn two_natives_msg() {

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

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            c.connect(SEC1).unwrap();

            c.get(g).unwrap();

            c.sync(SEC1).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            c.connect(SEC1).unwrap();

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

            c.sync(SEC1).unwrap();

#[test]

fn connector_pair_nondet() {

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

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            c.connect(SEC1).unwrap();

            c.next_batch().unwrap();

            c.get(g).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            c.connect(SEC1).unwrap();

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

            c.sync(SEC1).unwrap();

#[test]

fn parent_timeout() {

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

    scope(|s| {

        s.spawn(|_| {

            // parent; times out

            let mut c = file_logged_connector(999, test_log_path);

            c.connect(SEC1).unwrap();

            c.sync(MS300).unwrap_err(); // timeout

        });

        s.spawn(|_| {

            // child

            let mut c = file_logged_connector(000, test_log_path);

            c.connect(SEC1).unwrap();

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

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

#[test]

fn child_timeout() {

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

    scope(|s| {

        s.spawn(|_| {

            // child; times out

            let mut c = file_logged_connector(000, test_log_path);

            c.connect(SEC1).unwrap();

            c.sync(MS300).unwrap_err(); // timeout

        });

        s.spawn(|_| {

            // parent

            let mut c = file_logged_connector(999, test_log_path);

            c.connect(SEC1).unwrap();

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

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

#[test]

fn net_self_loop() {

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

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(SEC1).unwrap();

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

    c.get(g).unwrap();

#[test]

fn nobody_connects_active() {

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

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(Some(Duration::from_secs(5))).unwrap_err();

}

#[test]

fn nobody_connects_passive() {

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

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(Some(Duration::from_secs(5))).unwrap_err();

}

    .unwrap();

}