pub fn handle_new_data_message(&mut self, ticket: MessageTicket, ctx: &mut ComponentCtx) {

        // Go through all branches that are awaiting new messages and see if

        // there is one that can receive this message.

            // Message should not be handled now

            return;

        }

    }

    pub fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) -> Option<ConnectorScheduling> {

        if let Some(round_conclusion) = self.consensus.handle_new_sync_comp_message(message, ctx) {

            return Some(self.enter_non_sync_mode(round_conclusion, ctx));

        }

    Inconsistent,

}

pub(crate) enum MessageOrigin {

    Past,

    Present,

        let branch = &self.branch_annotations[branch_id.index as usize];

        // Clear out internal storage to defaults

        self.highest_connector_id = ConnectorId::new_invalid();

        self.branch_annotations.clear();

        self.branch_markers.clear();

    fn handle_received_sync_header(&mut self, sync_header: SyncHeader, ctx: &mut ComponentCtx) -> MessageOrigin {

        debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves

        let origin = self.handle_peer(&sync_header);

        if origin != MessageOrigin::Present {

            // We do not have to handle it now

            return origin;

                    // Nothing to do. But we're stil waiting for all our pending

                    // control messages to be answered.

                    self.debug_conn(connector_id, &format!("Shutting down, {} Acks remaining", scheduled.router.num_pending_acks()));

                    if scheduled.router.num_pending_acks() == 0 {

                        // We're actually done, we can safely destroy the

                        // currently running connector

                        }

                    }

                    if scheduled.router.num_pending_acks() == 0 {

                        // All ports (if any) already closed

                        self.runtime.destroy_component(connector_key);

            // If here, then we should handle the message

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

                match message.content {

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

                        // Need to change port target

                    },

                    ControlContent::Ping => {},

                }

                    scheduled.ctx.inbox.insert_new(message);

                }

            }

        }

    }

        let target_port_and_round_number = match message {

            Message::Data(msg) => Some((msg.data_header.target_port, msg.sync_header.sync_round)),

            Message::SyncComp(_) => None,

            self.runtime.send_message(port.peer_connector, Message::SyncControl(message));

        }

    }

    /// 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

            } else {

                // Just left sync region. So prepare inbox for the next sync

                // round

            }

            scheduled.ctx.changed_in_sync = false; // reset flag

/// for later reading. Later reading in this case implies that they are put back

/// for reading in the next sync round.

struct Inbox {

    messages: RawVec<Message>,

    next_delay_idx: u32,

    start_read_idx: u32,

impl Inbox {

    fn new() -> Self {

        return Inbox {

            messages: RawVec::new(),

            next_delay_idx: 0,

            start_read_idx: 0,

    fn insert_new(&mut self, message: Message) {

        assert!(self.messages.len() < u32::MAX as usize); // TODO: @Size

        self.messages.push(message);

    }

    fn get_next_message_ticket(&mut self) -> Option<MessageTicket> {

        let cur_read_idx = self.next_read_idx as usize;

        if cur_read_idx >= self.messages.len() {

            return None;

    fn read_message_using_ticket(&self, ticket: MessageTicket) -> &Message {

        debug_assert_eq!(self.generation, ticket.generation);

        return unsafe{ &*self.messages.get(ticket.index as usize) }

    }

    fn take_message_using_ticket(&mut self, ticket: MessageTicket) -> Message {

        debug_assert_eq!(self.generation, ticket.generation);

        unsafe {

            let take_idx = ticket.index as usize;

            let val = std::ptr::read(self.messages.get(take_idx));

    fn put_back_message(&mut self, message: Message) {

        // We have space in front of the array because we've taken out a message

        // before.

        debug_assert!(self.next_delay_idx < self.start_read_idx);

        unsafe {

            // Write to front of the array

    }

    fn get_read_data_messages(&self, match_port_id: PortIdLocal) -> MessagesIter {

        return MessagesIter{

            messages: self.messages.as_slice(),

            next_index: self.start_read_idx as usize,

        };

    }

        // Deallocate everything that was read

        self.destroy_range(self.start_read_idx, self.next_read_idx);

        self.generation += 1;

    }

    fn transfer_messages_for_port(&mut self, port: PortIdLocal, new_inbox: &mut Inbox) {

        let mut idx = 0;

        while idx < self.messages.len() {

            let message = unsafe{ &*self.messages.get(idx) };

                    // Transfer port

                    unsafe {

                        let message = std::ptr::read(message as *const _);

                        }

                        self.messages.len -= 1;

                        new_inbox.insert_new(message);

                    }

                }

            }

        }

    }

        }

    }

}

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

// Control messages

// pub(crate) const NUM_INSTANCES: u32 = 7;   // number of test instances constructed

// pub(crate) const NUM_LOOPS: u32 = 8;       // number of loops within a single test (not used by all tests)

fn create_runtime(pdl: &str) -> Runtime {