mod common;

mod protocol;

mod runtime;

mod collections;

pub use common::{ConnectorId, EndpointPolarity, Payload, Polarity, PortId};

pub(crate) struct ScheduledConnector {

    pub connector: ConnectorVariant, // access by connector

    pub public: ConnectorPublic, // accessible by all schedulers and connectors

    pub router: ControlMessageHandler,

    pub shutting_down: bool,

    fn create(&mut self, connector: ConnectorVariant, initially_sleeping: bool) -> ConnectorKey {

        let mut connector = ScheduledConnector {

            connector,

            public: ConnectorPublic::new(initially_sleeping),

            router: ControlMessageHandler::new(),

            shutting_down: false,

            // No free entries, allocate new entry

            index = self.connectors.len();

            key = ConnectorKey{ index: index as u32 };

            let connector = Box::into_raw(Box::new(connector));

            self.connectors.push(connector);

            // Free spot available

            index = self.free.pop().unwrap();

            key = ConnectorKey{ index: index as u32 };

            unsafe {

                let target = self.connectors.get_mut(index);

                } else {

                    self.debug_conn(connector_id, "Running ...");

                    let scheduler_ctx = SchedulerCtx{ runtime: &*self.runtime };

                    self.debug_conn(connector_id, "Finished running");

                    // Handle all of the output from the current run: messages to

                    // Prepare for exit. Set the shutdown flag and broadcast

                    // messages to notify peers of closing channels

                    scheduled.shutting_down = true;

                        if port.state != PortState::Closed {

                            let message = scheduled.router.prepare_closing_channel(

                                port.self_id, port.peer_id,

    /// Receiving messages from the public inbox and handling them or storing

    /// them in the component's private inbox

    fn handle_inbox_messages(&mut self, scheduled: &mut ScheduledConnector) {

        while let Some(message) = scheduled.public.inbox.take_message() {

            // Check if the message has to be rerouted because we have moved the

                    match message.content {

                        ControlContent::PortPeerChanged(port_id, new_target_connector_id) => {

                            // Need to change port target

                            port.peer_connector = new_target_connector_id;

                            // Note: for simplicity we program the scheduler to always finish

                            // running a connector with an empty outbox. If this ever changes

                            // then accepting the "port peer changed" message implies we need

                            // to change the recipient of the message in the outbox.

                            // And respond with an Ack

                            let ack_message = Message::Control(ControlMessage {

                        },

                        ControlContent::CloseChannel(port_id) => {

                            // Mark the port as being closed

                            port.state = PortState::Closed;

                            // Send an Ack

                },

                _ => {

                    // All other cases have to be handled by the component

                }

            }

        }

    /// component's state that the scheduler needs to know about (e.g. a message

    /// that the component wants to send, a port that has been added).

    fn handle_changes_in_context(&mut self, scheduled: &mut ScheduledConnector) {

        // Handling any messages that were sent

            self.debug_conn(connector_id, &format!("Sending message [outbox] \n --- {:?}", message));

            let target_component_id = match &message {

                Message::Data(content) => {

                    // Data messages are always sent to a particular port, and

                    // may end up being rerouted.

                    debug_assert_eq!(port_desc.peer_id, content.data_header.target_port);

                    if port_desc.state == PortState::Closed {

            self.runtime.send_message(target_component_id, message);

        }

            match state_change {

                ComponentStateChange::CreatedComponent(component, initial_ports) => {

                    // Creating a new component. The creator needs to relinquish

                    for port_id in initial_ports {

                        // Transfer messages associated with the transferred port

                        let mut message_idx = 0;

                            if Self::get_message_target_port(message) == Some(port_id) {

                                // Need to transfer this message

                            } else {

                                message_idx += 1;

                            }

                        }

                        // Transfer the port itself

                            .position(|v| v.self_id == port_id)

                            .unwrap();

                        // Notify the peer that the port has changed

                        let reroute_message = scheduled.router.prepare_reroute(

                        );

                        self.debug_conn(connector_id, &format!("Sending message [newcon]\n --- {:?}", reroute_message));

                    self.runtime.push_work(new_key);

                },

                ComponentStateChange::CreatedPort(port) => {

                },

                ComponentStateChange::ChangedPort(port_change) => {

                    if port_change.is_acquired {

                    } else {

                            .iter()

                            .position(|v| v.self_id == port_change.port.self_id)

                            .unwrap();

                    }

                }

            }

        }

        // Finally, check if we just entered or just left a sync region

                // Just entered sync region

            } else {

                // Just left sync region. So clear inbox

            }

        }

    }

    fn try_go_to_sleep(&self, connector_key: ConnectorKey, connector: &mut ScheduledConnector) {

        debug_assert_eq!(connector.public.sleeping.load(Ordering::Acquire), false);

        // This is the running connector, and only the running connector may