impl From<OutPort> for Port {

    fn from(x: OutPort) -> Self {

        x.0

    }

}

#[derive(Default, Debug)]

struct ChannelIndexStream {

    next: u32,

}

impl ChannelIndexStream {

    fn next(&mut self) -> u32 {

        self.next += 1;

        self.next - 1

    }

}

enum Connector {

    Connecting(Connecting),

    Connected(Connected),

}

#[derive(Default)]

pub struct Connecting {

    bindings: Vec<Binding>,

}

trait Binds<T> {

    fn bind(&mut self, coupling: Coupling, addr: SocketAddr) -> T;

}

impl Binds<InPort> for Connecting {

    fn bind(&mut self, coupling: Coupling, addr: SocketAddr) -> InPort {

        self.bindings.push(Binding { coupling, polarity: Polarity::Getter, addr });

        InPort(Port(self.bindings.len() - 1))

    }

}

impl Binds<OutPort> for Connecting {

    fn bind(&mut self, coupling: Coupling, addr: SocketAddr) -> OutPort {

        self.bindings.push(Binding { coupling, polarity: Polarity::Putter, addr });

        OutPort(Port(self.bindings.len() - 1))

    }

}

#[derive(Debug, Clone)]

pub enum ConnectErr {

    BindErr(SocketAddr),

    NewSocketErr(SocketAddr),

    AcceptErr(SocketAddr),

    ConnectionShutdown(SocketAddr),

    EndpointErr(Port, EndpointErr),

    PollInitFailed,

    PollingFailed,

    Timeout,

}

#[derive(Debug)]

struct Component {

    protocol: Arc<ProtocolD>,

    port_set: HashSet<Port>,

    identifier: Arc<[u8]>,

    state: ProtocolS,

}

impl From<PollDeadlineErr> for ConnectErr {

    fn from(e: PollDeadlineErr) -> Self {

        use PollDeadlineErr as P;

        match e {

            P::PollingFailed => Self::PollingFailed,

            P::Timeout => Self::Timeout,

        }

    }

}

impl From<MessengerRecvErr> for ConnectErr {

    fn from(e: MessengerRecvErr) -> Self {

        use MessengerRecvErr as M;

        match e {

            M::PollingFailed => Self::PollingFailed,

            M::EndpointErr(port, err) => Self::EndpointErr(port, err),

        }

    }

}

impl Connecting {

    fn random_controller_id() -> ControllerId {

        type Bytes8 = [u8; std::mem::size_of::<ControllerId>()];

        let mut bytes = Bytes8::default();

        getrandom::getrandom(&mut bytes).unwrap();

        unsafe {

            // safe:

            // 1. All random bytes give valid Bytes8

            // 2. Bytes8 and ControllerId have same valid representations

            std::mem::transmute::<Bytes8, ControllerId>(bytes)

        }

    }

    fn test_stream_connectivity(stream: &mut TcpStream) -> bool {

        use std::io::Write;

        stream.write(&[]).is_ok()

    }

        let mut backoff_millis = 10;

        while num_todos_remaining > 0 {

            ms.poll_events_until(deadline)?;

            for event in ms.events.iter() {

                let token = event.token();

                let index = token.0;

                let binding = &self.bindings[index];

                match todos[index].take() {

                    None => {

                        polled_undrained_later.insert(index);

                    }

                    Some(Todo::PassiveAccepting { listener, channel_id }) => {

                        let (stream, _peer_addr) =

                            listener.accept().map_err(|_| AcceptErr(binding.addr))?;

                        ms.poll.deregister(&listener).expect("wer");

                        ms.poll.register(&stream, token, ready_w, edge).expect("3y5");

                        todos[index] = Some(Todo::PassiveConnecting { stream, channel_id });

                    }

                    Some(Todo::ActiveConnecting { mut stream }) => {

                        let todo = if Self::test_stream_connectivity(&mut stream) {

                            ms.poll.reregister(&stream, token, ready_r, edge).expect("52");

                            let endpoint = Endpoint::from_fresh_stream(stream);

                            Todo::ActiveRecving { endpoint }

                        } else {

                            ms.poll.deregister(&stream).expect("wt");

                            std::thread::sleep(Duration::from_millis(backoff_millis));

                            backoff_millis = ((backoff_millis as f32) * 1.2) as u64 + 3;

                            let stream = TcpStream::connect(&binding.addr).unwrap();

                            ms.poll.register(&stream, token, ready_w, edge).expect("PAC 3");

                            Todo::ActiveConnecting { stream }

                        };

                        todos[index] = Some(todo);

                    }

                    Some(Todo::PassiveConnecting { mut stream, channel_id }) => {

                        if !Self::test_stream_connectivity(&mut stream) {

                            return Err(ConnectionShutdown(binding.addr));

                        }

                        ms.poll.reregister(&stream, token, ready_r, edge).expect("55");

                        let polarity = binding.polarity;

                        let info = EndpointInfo { polarity, channel_id };

                        let msg = Msg::SetupMsg(SetupMsg::ChannelSetup { info });

                        let mut endpoint = Endpoint::from_fresh_stream(stream);

                        endpoint.send(msg).map_err(|e| EndpointErr(Port(index), e))?;

                        let endpoint_ext = EndpointExt { endpoint, info };

                        endpoint_exts.occupy_reserved(index, endpoint_ext);

                        num_todos_remaining -= 1;

                    }

                    Some(Todo::ActiveRecving { mut endpoint }) => {

                        'recv_loop: while let Some(msg) =

                        {

                            if let Msg::SetupMsg(SetupMsg::ChannelSetup { info }) = msg {

                                if info.polarity == binding.polarity {

                                }

                                let channel_id = info.channel_id;

                                let info = EndpointInfo { polarity: binding.polarity, channel_id };

                                ms.polled_undrained.insert(ekey);

                                let endpoint_ext = EndpointExt { endpoint, info };

                                endpoint_exts.occupy_reserved(index, endpoint_ext);

                                num_todos_remaining -= 1;

                                break 'recv_loop;

                            } else {

                                ms.delayed.push(ReceivedMsg { recipient: ekey, msg });

                            }

                        }

                    }

                }

            }

        }

        assert_eq!(None, endpoint_exts.iter_reserved().next());

        drop(todos);

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

        // 1. construct `family', i.e. perform the sink tree setup procedure

        use {Msg::SetupMsg as S, SetupMsg::*};

        let mut messenger = (&mut ms, &mut endpoint_exts);

        impl Messengerlike for (&mut MessengerState, &mut VecStorage<EndpointExt>) {

            fn get_state_mut(&mut self) -> &mut MessengerState {

                self.0

            }

            fn get_endpoint_mut(&mut self, ekey: Key) -> &mut Endpoint {

                &mut self

                    .1

                    .get_occupied_mut(ekey.to_raw() as usize)

                    .expect("OUT OF BOUNDS")

                    .endpoint

            }

        }

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

        let neighbors = (0..self.bindings.len()).map(Port);

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

        let mut awaiting = IndexSet::<Port>::with_capacity(neighbors.len());

        for n in neighbors.clone() {

            messenger.send(n, echo.clone()).map_err(|e| EndpointErr(n, e))?;

            awaiting.insert(n);

        }

        // 2. Receive incoming replies. whenever a higher-id echo arrives,

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

        let mut parent: Option<Port> = None;

    inbufptr: *mut u8,

    opslen: usize,

    opsptr: *mut PortOp,

    subsetslen: usize,

    subsetsptr: *const *const usize,

) -> i32 {

    let buf: &mut [u8] = as_mut_slice(inbuflen, inbufptr);

    let ops: &mut [PortOp] = as_mut_slice(opslen, opsptr);

    let subsets: &[*const usize] = as_const_slice(subsetslen, subsetsptr);

    let subsetlen = usizes_for_bits(opslen);

    // don't yet know subsetptr; which subset fires unknown!

    let (subset_index, wrote) = sync_inner(connected, buf);

    let subsetptr: *const usize = subsets[subset_index];

    let subset: &[usize] = as_const_slice(subsetlen, subsetptr);

    for index in BitChunkIter::new(subset.iter().copied()) {

        let op = &mut ops[index as usize];

        if let Some(range) = wrote.get(&op.port) {

            op.msgptr = inbufptr.add(range.start);

            op.msglen = range.end - range.start;

        }

    }

    subset_index as i32

}

// dummy fn for the actual synchronous round

fn sync_inner<'c, 'b>(

    _connected: &'c mut Connected,

    _buf: &'b mut [u8],

) -> (usize, &'b HashMap<Port, Range<usize>>) {

    todo!()

}

unsafe fn as_mut_slice<'a, T>(len: usize, ptr: *mut T) -> &'a mut [T] {

    std::slice::from_raw_parts_mut(ptr, len)

}

unsafe fn as_const_slice<'a, T>(len: usize, ptr: *const T) -> &'a [T] {

    std::slice::from_raw_parts(ptr, len)

}

#[test]

fn api_connecting() {

    let addrs: [SocketAddr; 3] = [

        "127.0.0.1:8888".parse().unwrap(),

        "127.0.0.1:8889".parse().unwrap(),

        "127.0.0.1:8890".parse().unwrap(),

    ];

    let handles = vec![

        std::thread::spawn(move || {

            let mut connecting = Connecting::default();

            println!("A: {:#?}", connected);

        }),

        std::thread::spawn(move || {

            let mut connecting = Connecting::default();

            println!("B: {:#?}", connected);

        }),

    ];

    for h in handles {

        h.join().unwrap();

    }

}