}

impl ConnectorPDL {

        Self{

            tree: ExecTree::new(initial),

            consensus: Consensus::new(),

            let receiving_branch = &mut self.tree[receiving_branch_id];

            receiving_branch.insert_message(message.data_header.target_port, message.content.clone());

            // And prepare the branch for running

            self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);

                        definition_id, monomorph_idx, arguments

                    ),

                };

                return ConnectorScheduling::Later;

            },

        let port_info = ctx.get_port_by_id(source_port_id).unwrap();

        let data_header = DataHeader{

            expected_mapping: branch.port_mapping.clone(),

            target_port: port_info.peer_id,

            new_mapping: branch_id

        };

use super::ConnectorId;

use crate::protocol::eval::ValueGroup;

use super::connector::BranchId;

use super::port::PortIdLocal;

// TODO: @Optimize, lazy concurrency. Probably ringbuffer with read/write heads.

//  Should behave as a MPSC queue.

pub struct PublicInbox {

}

impl PublicInbox {

        }

    }

        let mut lock = self.messages.lock().unwrap();

        lock.push_back(message);

    }

        let mut lock = self.messages.lock().unwrap();

        return lock.pop_front();

    }

use crate::runtime2::ConnectorId;

use crate::runtime2::port::PortIdLocal;

pub(crate) struct PortAnnotation {

    pub port_id: PortIdLocal,

    pub registered_id: Option<BranchId>,

}

/// The header added by the synchronization algorithm to all.

pub(crate) struct SyncHeader {

    pub sending_component_id: ConnectorId,

    pub highest_component_id: ConnectorId,

}

/// The header added to data messages

pub(crate) struct DataHeader {

    pub expected_mapping: Vec<PortAnnotation>,

    pub target_port: PortIdLocal,

    pub new_mapping: BranchId,

}

/// A data message is a message that is intended for the receiver's PDL code,

/// but will also be handled by the consensus algrorithm

pub(crate) struct DataMessageFancy {

    pub sync_header: SyncHeader,

    pub data_header: DataHeader,

    pub content: ValueGroup,

}

pub(crate) enum SyncContent {

}

/// A sync message is a message that is intended only for the consensus

/// algorithm.

pub(crate) struct SyncMessageFancy {

    pub sync_header: SyncHeader,

    pub content: SyncContent,

/// A control message is a message intended for the scheduler that is executing

/// a component.

pub(crate) struct ControlMessageFancy {

    pub id: u32, // generic identifier, used to match request to response

    pub content: ControlContent,

}

pub(crate) enum ControlContent {

    PortPeerChanged(PortIdLocal, ConnectorId),

    CloseChannel(PortIdLocal),

}

/// Combination of data message and control messages.

pub(crate) enum MessageFancy {

    Data(DataMessageFancy),

    Sync(SyncMessageFancy),

use crate::ProtocolDescription;

use inbox::Message;

use scheduler::{Scheduler, ControlMessageHandler};

use native::{Connector, ConnectorApplication, ApplicationInterface};

use crate::runtime2::port::{Port, PortState};

use crate::runtime2::scheduler::{ComponentCtxFancy, SchedulerCtx};

    /// Sends a message to a particular connector. If the connector happened to

    /// be sleeping then it will be scheduled for execution.

        let target = self.get_component_public(target_id);

        target.inbox.insert_message(message);

use std::collections::VecDeque;

use std::sync::Arc;

use std::sync::atomic::Ordering;

use super::{ScheduledConnector, RuntimeInner, ConnectorId, ConnectorKey, ConnectorVariant};

use super::port::{Port, PortState, PortIdLocal};

use super::native::Connector;

// Because it contains pointers we're going to do a copy by value on this one

#[derive(Clone, Copy)]

                                connector_id

                            );

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

                        }

                    }

            // Handle special messages here, messages for the component

            // will be added to the inbox.

            self.debug_conn(connector_id, " ... Handling message myself");

                            // Need to change port target

                            let port = scheduled.ctx_fancy.get_port_mut_by_id(port_id).unwrap();

                            port.peer_connector = new_target_connector_id;

                            debug_assert!(scheduled.ctx_fancy.outbox.is_empty());

                            // And respond with an Ack

                                id: content.id,

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

                        },

                            // Mark the port as being closed

                            let port = scheduled.ctx_fancy.get_port_mut_by_id(port_id).unwrap();

                            port.state = PortState::Closed;

                            // Send an Ack

                                id: content.id,

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

                        },

                            scheduled.router.handle_ack(content.id);

                        },

                },

                _ => {

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

    /// Handles changes to the context that were made by the component. This is

    /// the way (due to Rust's borrowing rules) that we bubble up changes in the

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

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

        let connector_id = scheduled.ctx_fancy.id;

        // Handling any messages that were sent

        while let Some(mut message) = scheduled.ctx_fancy.outbox.pop_front() {

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

                    }

                },

                }

            };

        }

        while let Some(state_change) = scheduled.ctx_fancy.state_changes.pop_front() {

            match state_change {

                    // Add the new connector to the global registry

                    let new_key = self.runtime.create_pdl_component(component, false);

                    let new_connector = self.runtime.get_component_private(&new_key);

                    // Transfer ports

                        // Transfer messages associated with the transferred port

                        let mut message_idx = 0;

                        while message_idx < scheduled.ctx_fancy.inbox_messages.len() {

                        );

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

                    }

                    // Schedule new connector to run

// -----------------------------------------------------------------------------

enum ComponentStateChange {

    CreatedPort(Port),

    ChangedPort(ComponentPortChange),

}

    is_in_sync: bool,

    changed_in_sync: bool,

    outbox: VecDeque<MessageFancy>,

    // Workspaces that may be used by components to (generally) prevent

    // allocations. Be a good scout and leave it empty after you've used it.

    pub workspace_ports: Vec<PortIdLocal>,

    pub workspace_branches: Vec<BranchId>,

}

impl ComponentCtxFancy {

    pub(crate) fn new_empty() -> Self {

        return Self{

            changed_in_sync: false,

            outbox: VecDeque::new(),

            state_changes: VecDeque::new(),

        };

    }

    /// Notify the runtime that the component has created a new component. May

    /// only be called outside of a sync block.

        debug_assert!(!self.is_in_sync);

    }

    /// Notify the runtime that the component has created a new port. May only

        if !self.is_in_sync { return None; }

        if self.inbox_len_read == self.inbox_messages.len() { return None; }

        let message = &self.inbox_messages[self.inbox_len_read];

                self.inbox_len_read += 1;

                let message = self.inbox_messages.remove(self.inbox_len_read);

        }

    }

}

    pub fn prepare_closing_channel(

        &mut self, self_port_id: PortIdLocal, peer_port_id: PortIdLocal,

        self_connector_id: ConnectorId

        let id = self.take_id();

        self.active.push(ControlEntry{

            }),

        });

            id,

        };

    }

        port_id: PortIdLocal, peer_port_id: PortIdLocal,

        self_connector_id: ConnectorId, peer_connector_id: ConnectorId,

        new_owner_connector_id: ConnectorId

        let id = self.take_id();

        self.active.push(ControlEntry{

            }),

        });

            id,

        };

    }