use crate::protocol::eval::{Prompt, Value, ValueGroup};

use crate::protocol::{RunContext, RunResult};

use super::branch::{BranchId, ExecTree, QueueKind, SpeculativeState};

use super::consensus::{Consensus, Consistency, find_ports_in_value_group};

use super::inbox::{DataMessage, DataContent, Message, SyncMessage, PublicInbox};

use super::native::Connector;

use super::port::{PortKind, PortIdLocal};

use super::scheduler::{ComponentCtx, SchedulerCtx};

pub(crate) struct ConnectorPublic {

    pub inbox: PublicInbox,

    pub sleeping: AtomicBool,

}

impl ConnectorPublic {

    pub fn new(initialize_as_sleeping: bool) -> Self {

        ConnectorPublic{

            inbox: PublicInbox::new(),

            sleeping: AtomicBool::new(initialize_as_sleeping),

        }

    }

}

pub(crate) enum ConnectorScheduling {

    Immediate,      // Run again, immediately

    Later,          // Schedule for running, at some later point in time

    NotNow,         // Do not reschedule for running

    Exit,           // Connector has exited

}

pub(crate) struct ConnectorPDL {

    tree: ExecTree,

    consensus: Consensus,

    last_finished_handled: Option<BranchId>,

}

struct ConnectorRunContext<'a> {

    branch_id: BranchId,

    consensus: &'a Consensus,

    received: &'a HashMap<PortIdLocal, ValueGroup>,

    scheduler: SchedulerCtx<'a>,

    prepared_channel: Option<(Value, Value)>,

}

impl<'a> RunContext for ConnectorRunContext<'a>{

    fn did_put(&mut self, port: PortId) -> bool {

        let port_id = PortIdLocal::new(port.0.u32_suffix);

                    self.runtime.push_work(new_key);

                },

                ComponentStateChange::CreatedPort(port) => {

                    scheduled.ctx.ports.push(port);

                },

                ComponentStateChange::ChangedPort(port_change) => {

                    if port_change.is_acquired {

                        scheduled.ctx.ports.push(port_change.port);

                    } else {

                        let index = scheduled.ctx.ports

                            .iter()

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

                            .unwrap();

                        scheduled.ctx.ports.remove(index);

                    }

                }

            }

        }

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

        if scheduled.ctx.changed_in_sync {

            if scheduled.ctx.is_in_sync {

                // Just entered sync region

            } else {

                scheduled.ctx.inbox_len_read = 0;

            }

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

        }

    }

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

        debug_assert_eq!(connector_key.index, connector.ctx.id.0);

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

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

        // decide it wants to sleep again.

        connector.public.sleeping.store(true, Ordering::Release);

        // But due to reordering we might have received messages from peers who

        // did not consider us sleeping. If so, then we wake ourselves again.

        if !connector.public.inbox.is_empty() {

            // Try to wake ourselves up (needed because someone might be trying

            // the exact same atomic compare-and-swap at this point in time)

            let should_wake_up_again = connector.public.sleeping

                .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)

                .is_ok();

// ComponentCtx

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

enum ComponentStateChange {

    CreatedComponent(ConnectorPDL, Vec<PortIdLocal>),

    CreatedPort(Port),

    ChangedPort(ComponentPortChange),

}

#[derive(Clone)]

pub(crate) struct ComponentPortChange {

    pub is_acquired: bool, // otherwise: released

    pub port: Port,

}

/// The component context (better name may be invented). This was created

/// because part of the component's state is managed by the scheduler, and part

/// of it by the component itself. When the component starts a sync block or

/// exits a sync block the partially managed state by both component and

/// scheduler need to be exchanged.

pub(crate) struct ComponentCtx {

    // Mostly managed by the scheduler

    pub(crate) id: ConnectorId,

    ports: Vec<Port>,

    inbox_len_read: usize,

    // Submitted by the component

    is_in_sync: bool,

    changed_in_sync: bool,

    outbox: VecDeque<Message>,

    state_changes: VecDeque<ComponentStateChange>,

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

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

    // TODO: Move to scheduler ctx, this is the wrong place

    pub workspace_ports: Vec<PortIdLocal>,

    pub workspace_branches: Vec<BranchId>,

}

impl ComponentCtx {

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

        return Self{

            id: ConnectorId::new_invalid(),

            ports: Vec::new(),

            inbox_messages: Vec::new(),

            inbox_len_read: 0,

            is_in_sync: false,

            changed_in_sync: false,

            outbox: VecDeque::new(),

            state_changes: VecDeque::new(),

    let req_chan = api.create_channel().unwrap();

    let resp_chan = api.create_channel().unwrap();

    api.create_connector("", "handler", ValueGroup::new_stack(vec![

        Value::Input(PortId::new(req_chan.getter_id.index)),

        Value::Output(PortId::new(resp_chan.putter_id.index)),

        Value::UInt32(NUM_LOOPS),

    ])).unwrap();

    for loop_idx in 0..NUM_LOOPS {

        api.perform_sync_round(vec![

            ApplicationSyncAction::Put(req_chan.putter_id, ValueGroup::new_stack(vec![Value::UInt32(loop_idx)])),

            ApplicationSyncAction::Get(resp_chan.getter_id)

        ]).expect("start sync round");

        let result = api.wait().expect("finish sync round");

        assert!(result.len() == 1);

        if let Value::UInt32(gotten) = result[0].values[0] {

            assert_eq!(gotten, loop_idx * 2);

        } else {

            assert!(false);

        }

    }

}