#include <stdio.h>

#include <string.h>

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

#include "../utility.c"

int main(int argc, char** argv) {

	char * pdl_ptr = buffer_pdl("eg_protocols.pdl");

	size_t pdl_len = strlen(pdl_ptr);

	Arc_ProtocolDescription * pd = protocol_description_parse(pdl_ptr, pdl_len);

	char logpath[] = "./6_amy_log.txt";

	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);

	printf("Err %s\n", reowolf_error_peek(NULL));

	PortId putter, getter;

	connector_add_port_pair(c, &putter, &getter);

	connector_connect(c, -1);

	connector_print_debug(c);

	printf("Let's try to put without get\n");

	connector_put_bytes(c, putter, "hello", 5);

	// connector_get(c, getter);

	int err = connector_sync(c, 5000);

	printf("Error code %d with string `%s`\n", err, reowolf_error_peek(NULL));

	protocol_description_destroy(pd);

	connector_destroy(c);

	free(pdl_ptr);

	return 0;

}

/**

 * Given

 * - an initialized connector in setup or connecting state,

 * - a string slice for the component's identifier in the connector's configured protocol description,

 * - a set of ports (represented as a slice; duplicates are ignored) in the native component's interface,

 * the connector creates a new (internal) protocol component C, such that the set of native ports are moved to C.

 * Usable in {setup, communication} states.

 */

ErrorCode connector_add_component(Connector *connector,

                                  const uint8_t *ident_ptr,

                                  uintptr_t ident_len,

                                  const PortId *ports_ptr,

                                  uintptr_t ports_len);

/**

 * Given

 * - an initialized connector in setup or connecting state,

 * - a utf-8 encoded socket address,

 * - the logical polarity of P,

 * - the "physical" polarity in {Active, Passive} of the endpoint through which P's peer will be discovered,

 * returns P, a port newly added to the native interface.

 */

ErrorCode connector_add_net_port(Connector *connector,

                                 const uint8_t *addr_str_ptr,

                                 uintptr_t addr_str_len,

                                 Polarity port_polarity,

/**

 * Given an initialized connector in setup or connecting state,

 * - Creates a new directed port pair with logical channel putter->getter,

 * - adds the ports to the native component's interface,

 * - and returns them using the given out pointers.

 * Usable in {setup, communication} states.

 */

void connector_add_port_pair(Connector *connector, PortId *out_putter, PortId *out_getter);

/**

 * Connects this connector to the distributed system of connectors reachable through endpoints,

 * Usable in setup state, and changes the state to communication.

 */

ErrorCode connector_connect(Connector *connector, int64_t timeout_millis);

/**

 * Destroys the given a pointer to the connector on the heap, freeing its resources.

 * Usable in {setup, communication} states.

 */

void connector_destroy(Connector *connector);

ErrorCode connector_get(Connector *connector, PortId port);

                ports,

            )?;

            // swap the blocked branches back

            std::mem::swap(&mut blocked, branches);

            if branches.is_empty() {

                log!(cu.logger, "{:?} has become inconsistent!", proto_component_id);

                if let Some(parent) = comm.neighborhood.parent {

                    if already_requested_failure.replace_with_true() {

                        Self::request_failure(cu, comm, parent)?

                    } else {

                        log!(cu.logger, "Already requested failure");

                    }

                } else {

                    log!(cu.logger, "As the leader, deciding on timeout");

                    return Ok(Decision::Failure);

                }

            }

        }

        log!(cu.logger, "All proto components are blocked");

        log!(cu.logger, "Entering decision loop...");

        comm.endpoint_manager.undelay_all();

        'undecided: loop {

            // drain payloads_to_get, sending them through endpoints / feeding them to components

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

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

                    None => {

                        log!(

                            cu.logger,

                            "Delivery to getter {:?} msg {:?} failed. Physical route unmapped!",

                            getter,

                            &send_payload_msg

                        );

                    }

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

                        let msg = Msg::CommMsg(CommMsg {

                            round_index: comm.round_index,

                            contents: CommMsgContents::SendPayload(send_payload_msg),

                        });

                        comm.endpoint_manager.send_to_comms(*index, &msg)?;

                    }

                    Some(Route::LocalComponent(ComponentId::Native)) => branching_native.feed_msg(

                        cu,

                        &mut solution_storage,

                        // &mut Pay

                        getter,

                        &send_payload_msg,

                    ),

                    Some(Route::LocalComponent(ComponentId::Proto(proto_component_id))) => {

                        if let Some(branching_component) =

    ports_len: usize,

) -> ErrorCode {

    StoredError::tl_clear();

    match connector.add_component(

        &*slice_from_parts(ident_ptr, ident_len),

        &*slice_from_parts(ports_ptr, ports_len),

    ) {

        Ok(()) => 0,

        Err(err) => {

            StoredError::tl_debug_store(&err);

            -1

        }

    }

}

/// Given

/// - an initialized connector in setup or connecting state,

/// - a utf-8 encoded socket address,

/// - the logical polarity of P,

/// - the "physical" polarity in {Active, Passive} of the endpoint through which P's peer will be discovered,

/// returns P, a port newly added to the native interface.

#[no_mangle]

pub unsafe extern "C" fn connector_add_net_port(

    connector: &mut Connector,

    addr_str_ptr: *const u8,

    addr_str_len: usize,

    port_polarity: Polarity,

    endpoint_polarity: EndpointPolarity,

) -> ErrorCode {

    StoredError::tl_clear();

    let addr_bytes = &*slice_from_parts(addr_str_ptr, addr_str_len);

    let addr_str = match std::str::from_utf8(addr_bytes) {

        Ok(addr_str) => addr_str,

        Err(err) => {

            StoredError::tl_debug_store(&err);

            return -1;

        }

    };

    let sock_address: SocketAddr = match addr_str.parse() {

        Ok(addr) => addr,

        Err(err) => {

            StoredError::tl_debug_store(&err);

            return -2;

        }

    };

    match connector.new_net_port(port_polarity, sock_address, endpoint_polarity) {

        Ok(p) => {

            port.write(p);

            0

        }

        Err(err) => {

            StoredError::tl_debug_store(&err);

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

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

                .expect("mio::TcpStream connect should not fail!");

            poll.registry().register(&mut stream, token, BOTH).unwrap();

            TodoEndpoint::Endpoint(Endpoint { stream, inbox: vec![] })

        } else {

            let mut listener = TcpListener::bind(endpoint_setup.sock_addr)

                .map_err(|_| BindFailed(endpoint_setup.sock_addr))?;

            poll.registry().register(&mut listener, token, BOTH).unwrap();

            TodoEndpoint::Accepting(listener)

        };

        Ok(Todo {

            todo_endpoint,

            local_port,

            sent_local_port: false,

            recv_peer_port: None,

            endpoint_setup: endpoint_setup.clone(),

        })

    }

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

    // 1. Start to construct EndpointManager

    const WAKER_TOKEN: Token = Token(usize::MAX);

    const WAKER_PERIOD: Duration = Duration::from_millis(90);

    assert!(endpoint_setups.len() < WAKER_TOKEN.0); // using MAX usize as waker token

    let mut waker_continue_signal: Option<Arc<AtomicBool>> = None;

    let mut poll = Poll::new().map_err(|_| PollInitFailed)?;

    let mut events = Events::with_capacity(endpoint_setups.len() * 2 + 4);

    let mut polled_undrained = IndexSet::default();

    let mut delayed_messages = vec![];

    // 2. create a registered (TcpListener/Endpoint) for passive / active respectively

    let mut todos = endpoint_setups

        .iter()

        .enumerate()

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

            init_todo(Token(index), *local_port, endpoint_setup, &mut poll)

        })

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

    // 3. Using poll to drive progress:

    //    - accept an incoming connection for each TcpListener (turning them into endpoints too)

    //    - for each endpoint, send the local PortId

    //    - for each endpoint, recv the peer's PortId, and

    let mut connect_failed: HashSet<usize> = Default::default();

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

    while !setup_incomplete.is_empty() {

        let remaining = if let Some(deadline) = deadline {

            Some(deadline.checked_duration_since(Instant::now()).ok_or(Timeout)?)

                        todo.sent_local_port = true;

                    }

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

                        let maybe_msg = endpoint.try_recv(logger).map_err(|e| {

                            EndpointSetupError(endpoint.stream.local_addr().unwrap(), e)

                        })?;

                        if maybe_msg.is_some() && !endpoint.inbox.is_empty() {

                            polled_undrained.insert(index);

                        }

                        match maybe_msg {

                            None => {} // msg deserialization incomplete

                            Some(Msg::SetupMsg(SetupMsg::MyPortInfo(peer_info))) => {

                                log!(logger, "endpoint[{}] got peer info {:?}", index, peer_info);

                                if peer_info.polarity == local_polarity {

                                    return Err(ConnectError::PortPeerPolarityMismatch(

                                        todo.local_port,

                                    ));

                                }

                                todo.recv_peer_port = Some(peer_info.port);

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

                                port_info.peers.insert(todo.local_port, peer_info.port);

                                // 2. learned the info of this peer port

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

                                port_info.peers.insert(peer_info.port, todo.local_port);

                            }

                            Some(inappropriate_msg) => {

                                log!(

                                    logger,

                                    "delaying msg {:?} during channel setup phase",

                                    inappropriate_msg

                                );

                                delayed_messages.push((index, inappropriate_msg));

                            }

                        }

                    }

                    if todo.sent_local_port && todo.recv_peer_port.is_some() {

                        setup_incomplete.remove(&index);

                        log!(logger, "endpoint[{}] is finished!", index);

                    }

                }

            }

        }

        events.clear();

    }

    let endpoint_exts = todos

        .into_iter()

        .enumerate()

        .map(|(index, Todo { todo_endpoint, local_port, .. })| EndpointExt {

            c.new_net_port(Putter, sock_addrs[1], Passive).unwrap();

            c.connect(Some(Duration::from_secs(1))).unwrap();

        });

        s.spawn(|_| {

            // LEADER

            let mut c = file_logged_connector(7, test_log_path);

            c.new_net_port(Getter, sock_addrs[1], Active).unwrap();

            c.new_net_port(Putter, sock_addrs[2], Passive).unwrap();

            c.connect(Some(Duration::from_secs(1))).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(4, test_log_path);

            c.new_net_port(Getter, sock_addrs[2], Active).unwrap();

            c.new_net_port(Putter, sock_addrs[3], Passive).unwrap();

            c.connect(Some(Duration::from_secs(1))).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            c.new_net_port(Getter, sock_addrs[3], Active).unwrap();

            c.connect(Some(Duration::from_secs(1))).unwrap();

        });

    })

    .unwrap();

}