msg: Payload,

}

#[derive(Debug)]

    poll: Poll,

    events: Events,

    undrained_endpoints: IndexSet<usize>,

    delayed_messages: Vec<(usize, Msg)>,

    undelayed_messages: Vec<(usize, Msg)>,

}

#[derive(Debug)]

pub struct Connector {

        surplus_sockets: u16,

    },

    Communication {

        neighborhood: Neighborhood,

        mem_inbox: Vec<MemInMsg>,

    },

    BrokenEndpoint(usize),

}

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

        use TryRecyAnyError::*;

        // 1. try messages already buffered

        if let Some(x) = self.undelayed_messages.pop() {

        }

        // 2. try read from sockets nonblocking

        while let Some(index) = self.undrained_endpoints.pop() {

                .endpoint

                .try_recv()

                .map_err(|error| EndpointRecvErr { error, index })?

            self.poll.poll(&mut self.events, Some(remaining)).map_err(|_| PollFailed)?;

            for event in self.events.iter() {

                let Token(index) = event.token();

                    .endpoint

                    .try_recv()

                    .map_err(|error| EndpointRecvErr { error, index })?

                log!(self.logger, "Call to connecting in setup state. Timeout {:?}", timeout);

                let deadline = Instant::now() + timeout;

                // connect all endpoints in parallel; send and receive peer ids through ports

                    &mut *self.logger,

                    endpoint_setups,

                    &mut self.inp_to_route,

                    deadline,

                )?;

                // leader election and tree construction

                let neighborhood = init_neighborhood(

                    self.id_manager.controller_id,

                    &mut *self.logger,

                    deadline,

                )?;

                log!(self.logger, "Successfully created neighborhood {:?}", &neighborhood);

                // TODO session optimization goes here

                self.phased = ConnectorPhased::Communication {

                    neighborhood,

                    mem_inbox: Default::default(),

                };

    endpoint_setups: &[(PortId, EndpointSetup)],

    inp_to_route: &mut HashMap<PortId, InpRoute>,

    deadline: Instant,

    const BOTH: Interest = Interest::READABLE.add(Interest::WRITABLE);

    struct Todo {

        todo_endpoint: TodoEndpoint,

    };

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

        poll: Poll::new().map_err(drop)?,

        events: Events::with_capacity(64),

        undrained_endpoints: Default::default(),

        delayed_messages: Default::default(),

        undelayed_messages: Default::default(),

    };

    let mut todos = endpoint_setups

        .iter()

        .enumerate()

        .map(|(index, (local_port, endpoint_setup))| {

        })

        .collect::<Result<Vec<Todo>, _>>()?;

    let mut unfinished: HashSet<usize> = (0..todos.len()).collect();

    while !unfinished.is_empty() {

        let remaining = deadline.checked_duration_since(Instant::now()).ok_or(())?;

            let token = event.token();

            let Token(index) = token;

            let todo: &mut Todo = &mut todos[index];

            if let TodoEndpoint::Listener(listener) = &mut todo.todo_endpoint {

                let (mut stream, peer_addr) = listener.accept().map_err(drop)?;

                log!(logger, "Endpoint({}) accepted a connection from {:?}", index, peer_addr);

                let endpoint = Endpoint { stream, inbox: vec![] };

                todo.todo_endpoint = TodoEndpoint::Endpoint(endpoint);

                        *sent_local_port = true;

                    }

                    if event.is_readable() && recv_peer_port.is_none() {

                        if let Some(peer_port_info) =

                            endpoint.try_recv::<MyPortInfo>().map_err(drop)?

                        {

                Todo { todo_endpoint: TodoEndpoint::Listener(_), .. } => unreachable!(),

            }

        }

    }

            endpoint: match todo_endpoint {

                TodoEndpoint::Endpoint(endpoint) => endpoint,

                TodoEndpoint::Listener(..) => unreachable!(),

            },

            inp_for_emerging_msgs: recv_peer_port.unwrap(),

}

fn init_neighborhood(

    controller_id: ControllerId,

    logger: &mut dyn Logger,

    deadline: Instant,

) -> Result<Neighborhood, ()> {

    log!(logger, "beginning neighborhood construction");

    // 1. broadcast my ID as the first echo. await reply from all neighbors

    let echo = S(LeaderEcho { maybe_leader: controller_id });

        log!(logger, "{:?}'s initial echo to {:?}, {:?}", controller_id, index, &echo);

        ee.endpoint.send(&echo)?;

        awaiting.insert(index);

    //    adopt it as leader, sender as parent, and reset the await set.

    let mut parent: Option<usize> = None;

    let mut my_leader = controller_id;

    'echo_loop: while !awaiting.is_empty() || parent.is_some() {

        log!(logger, "GOT from index {:?} msg {:?}", &index, &msg);

        match msg {

            S(LeaderAnnounce { leader }) => {

                        if awaiting.is_empty() {

                            if let Some(p) = parent {

                                // return the echo to my parent

                            } else {

                                // DECIDE!

                                break 'echo_loop;

                        my_leader = maybe_leader;

                        let echo = S(LeaderEcho { maybe_leader: my_leader });

                        awaiting.clear();

                            // immediately reply to parent

                            log!(logger, "replying echo to parent {:?} immediately", index);

                        } else {

                                if index2 == index {

                                    continue;

                                }

                    }

                }

            }

        }

    }

    match parent {

    //    in this loop, every node sends 1 message to each neighbor

    // await 1 message from all non-parents

    let msg_for_non_parents = S(LeaderAnnounce { leader: my_leader });

        let msg = if Some(index) == parent {

            &S(YouAreMyParent)

        } else {

        ee.endpoint.send(msg)?;

    }

    let mut children = Vec::default();

    while !awaiting.is_empty() {

        match msg {

            S(YouAreMyParent) => {

                assert!(awaiting.remove(&index));

                assert!(Some(index) != parent);

                // they wouldn't send me this if they considered me their parent

            }

        }

    }

    children.sort();