use std::sync::{Arc, Mutex, Condvar};

use std::sync::atomic::{AtomicU32, AtomicBool, Ordering};

use std::thread;

use std::collections::VecDeque;

use crate::protocol::*;

use crate::runtime2::poll::{PollingThread, PollingThreadHandle};

use crate::runtime2::RtError;

use super::communication::Message;

use super::component::{Component, wake_up_if_sleeping, CompPDL, CompCtx};

use super::store::{ComponentStore, ComponentReservation, QueueDynMpsc, QueueDynProducer};

use super::scheduler::*;

#[derive(PartialOrd, PartialEq, Copy, Clone)]

pub enum LogLevel {

    Debug, // all logging (includes messages)

    Info, // rough logging

}

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

// Component

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

/// Key to a component. Type system somewhat ensures that there can only be one

/// of these. Only with a key one may retrieve privately-accessible memory for

/// a component. Practically just a generational index, like `CompId` is.

#[derive(Debug, Copy, Clone, PartialEq, Eq)]

pub(crate) struct CompKey(pub u32);

impl CompKey {

    pub(crate) fn downgrade(&self) -> CompId {

        return CompId(self.0);

    }

}

/// Generational ID of a component.

#[derive(Debug, Copy, Clone, PartialEq, Eq)]

pub struct CompId(pub u32);

impl CompId {

    pub(crate) fn new_invalid() -> CompId {

        return CompId(u32::MAX);

    }

    /// Upgrade component ID to component key. Unsafe because the caller needs

    /// to make sure that only one component key can exist at a time (to ensure

    /// a component can only be scheduled/executed by one thread).

    pub(crate) unsafe fn upgrade(&self) -> CompKey {

        return CompKey(self.0);

    }

}

/// Handle to a component that is being created.

pub(crate) struct CompReserved {

    reservation: ComponentReservation,

}

impl CompReserved {

    pub(crate) fn id(&self) -> CompId {

        return CompId(self.reservation.index)

    }

}

/// Representation of a runtime component. Contains the bookkeeping variables

/// for the schedulers, the publicly accessible fields, and the private fields

/// that should only be accessed by the thread running the component's routine.

pub(crate) struct RuntimeComp {

    pub public: CompPublic,

    pub component: Box<dyn Component>,

    pub ctx: CompCtx,

    pub inbox: QueueDynMpsc<Message>,

    pub exiting: bool,

}

/// Should contain everything that is accessible in a thread-safe manner

// TODO: Do something about the `num_handles` thing. This needs to be a bit more

//  "foolproof" to lighten the mental burden of using the `num_handles`

//  variable.

pub(crate) struct CompPublic {

    pub sleeping: AtomicBool,

    pub num_handles: AtomicU32, // manually modified (!)

    inbox: QueueDynProducer<Message>,

}

/// Handle to public part of a component. Would be nice if we could

/// automagically manage the `num_handles` counter. But when it reaches zero we

/// need to manually remove the handle from the runtime. So we just have debug

/// code to make sure this actually happens.

pub(crate) struct CompHandle {

    target: *const CompPublic,

    id: CompId,

    #[cfg(debug_assertions)] decremented: bool,

}

impl CompHandle {

    fn new(id: CompId, public: &CompPublic) -> CompHandle {

        let handle = CompHandle{

            target: public,

            id,

            #[cfg(debug_assertions)] decremented: false,

        };

        handle.increment_users();

        return handle;

    }

    pub(crate) fn send_message(&self, runtime: &RuntimeInner, message: Message, try_wake_up: bool) {

        self.inbox.push(message);

        if try_wake_up {

            wake_up_if_sleeping(runtime, self.id, self);

        }

    }

    #[inline]

    pub(crate) fn send_message_logged(&self, sched_ctx: &SchedulerCtx, message: Message, try_wake_up: bool) {

        sched_ctx.debug(&format!("Sending message to comp:{} ... {:?}", self.id.0, message));

        self.send_message(&sched_ctx.runtime, message, try_wake_up);

    }

    pub(crate) fn id(&self) -> CompId {

        return self.id;

    }

    fn increment_users(&self) {

        let old_count = self.num_handles.fetch_add(1, Ordering::AcqRel);

        debug_assert!(old_count > 0); // because we should never be able to retrieve a handle when the component is (being) destroyed

    }

    /// Returns the `CompKey` to the component if it should be destroyed

    pub(crate) fn decrement_users(&mut self) -> Option<CompKey> {

        dbg_code!(assert!(!self.decremented, "illegal to 'decrement_users' twice"));

        let old_count = self.num_handles.fetch_sub(1, Ordering::AcqRel);

        let new_count = old_count - 1;

        dbg_code!(self.decremented = true);

        if new_count == 0 {

            return Some(unsafe{ self.id.upgrade() });

        }

        return None;

    }

}

impl Clone for CompHandle {

    fn clone(&self) -> Self {

        dbg_code!(assert!(!self.decremented, "illegal to clone after 'decrement_users'"));

        self.increment_users();

        return CompHandle{

            target: self.target,

            id: self.id,

            #[cfg(debug_assertions)] decremented: false,

        };

    }

}

impl std::ops::Deref for CompHandle {

    type Target = CompPublic;

    fn deref(&self) -> &Self::Target {

        dbg_code!(assert!(!self.decremented)); // cannot access if control is relinquished

        return unsafe{ &*self.target };

    }

}

impl Drop for CompHandle {

    fn drop(&mut self) {

        dbg_code!(assert!(self.decremented, "need call to 'decrement_users' before dropping"));

    }

}

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

// Runtime

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

pub struct Runtime {

    pub(crate) inner: Arc<RuntimeInner>,

    scheduler_threads: Vec<thread::JoinHandle<()>>,

    polling_handle: PollingThreadHandle,

}

impl Runtime {

    // TODO: debug_logging should be removed at some point

    pub fn new(num_threads: u32, log_level: LogLevel, protocol_description: ProtocolDescription) -> Result<Runtime, RtError> {

        if num_threads == 0 {

            return Err(rt_error!("need at least one thread to create the runtime"));

        }

        let runtime_inner = Arc::new(RuntimeInner {

            protocol: protocol_description,

            components: ComponentStore::new(128),

            work_queue: Mutex::new(VecDeque::with_capacity(128)),

            work_condvar: Condvar::new(),

            active_elements: AtomicU32::new(1),

        });

        let (polling_handle, polling_clients) = rt_error_try!(

            PollingThread::new(runtime_inner.clone(), log_level),

            "failed to build polling thread"

        );

        let mut scheduler_threads = Vec::with_capacity(num_threads as usize);

        for thread_index in 0..num_threads {

            let mut scheduler = Scheduler::new(

                runtime_inner.clone(), polling_clients.client(),

                thread_index, log_level

            );

            let thread_handle = thread::Builder::new()

                .name(format!("scheduler:{}", thread_index))

                .spawn(move || {

                    scheduler.run();

                })

                .map_err(|reason|

                    rt_error!(

                        "failed to spawn scheduler thread {}, because: {}",

                        thread_index, reason

                    )

                )?;

            scheduler_threads.push(thread_handle);

        }

        return Ok(Runtime{

            inner: runtime_inner,

            scheduler_threads,

            polling_handle,

        });

    }

    pub fn create_component(&self, module_name: &[u8], routine_name: &[u8]) -> Result<(), ComponentCreationError> {

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

        let prompt = self.inner.protocol.new_component(

            module_name, routine_name,

            ValueGroup::new_stack(Vec::new())

        )?;

        let reserved = self.inner.start_create_pdl_component();

        let ctx = CompCtx::new(&reserved);

        let component = Box::new(CompPDL::new(prompt, 0));

        let (key, _) = self.inner.finish_create_pdl_component(reserved, component, ctx, false);

        self.inner.enqueue_work(key);

        return Ok(())

    }

}

impl Drop for Runtime {

    fn drop(&mut self) {

        self.inner.decrement_active_components();

        for handle in self.scheduler_threads.drain(..) {

            handle.join().expect("join scheduler thread");

        }

        self.polling_handle.shutdown().expect("shutdown polling thread");

    }

}

/// Memory that is maintained by "the runtime". In practice it is maintained by

/// multiple schedulers, and this serves as the common interface to that memory.

pub(crate) struct RuntimeInner {

    pub protocol: ProtocolDescription,

    components: ComponentStore<RuntimeComp>,

    work_queue: Mutex<VecDeque<CompKey>>, // TODO: should be MPMC queue

    work_condvar: Condvar,

    active_elements: AtomicU32, // active components and APIs (i.e. component creators)

}

impl RuntimeInner {

    // Scheduling and retrieving work

    pub(crate) fn take_work(&self) -> Option<CompKey> {

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

        while lock.is_empty() && self.active_elements.load(Ordering::Acquire) != 0 {

            lock = self.work_condvar.wait(lock).unwrap();

        }

        // We have work, or the schedulers should exit.

        return lock.pop_front();

    }

    pub(crate) fn enqueue_work(&self, key: CompKey) {

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

        lock.push_back(key);

        self.work_condvar.notify_one();

    }

    // Creating/destroying components

    pub(crate) fn start_create_pdl_component(&self) -> CompReserved {

        self.increment_active_components();

        let reservation = self.components.reserve();

        return CompReserved{ reservation };

    }

    pub(crate) fn finish_create_pdl_component(

        &self, reserved: CompReserved,

        component: Box<dyn Component>, mut context: CompCtx, initially_sleeping: bool,

    ) -> (CompKey, &mut RuntimeComp) {

        let inbox_queue = QueueDynMpsc::new(16);

        let inbox_producer = inbox_queue.producer();

        let _id = reserved.id();

        context.id = reserved.id();

        let component = RuntimeComp {

            public: CompPublic{

                sleeping: AtomicBool::new(initially_sleeping),

                num_handles: AtomicU32::new(1), // the component itself acts like a handle

                inbox: inbox_producer,

            },

            component,

            ctx: context,

            inbox: inbox_queue,

            exiting: false,

        };

        let index = self.components.submit(reserved.reservation, component);

        debug_assert_eq!(index, _id.0);

        let component = self.components.get_mut(index);

        return (CompKey(index), component);

    }

    pub(crate) fn get_component(&self, key: CompKey) -> &mut RuntimeComp {

        let component = self.components.get_mut(key.0);

        return component;

    }

    pub(crate) fn get_component_public(&self, id: CompId) -> CompHandle {

        let component = self.components.get(id.0);

        return CompHandle::new(id, &component.public);

    }

    /// Will remove a component and its memory from the runtime. May only be

    /// called if the necessary conditions for destruction have been met.

    pub(crate) fn destroy_component(&self, key: CompKey) {

        dbg_code!({

            let component = self.get_component(key);

            debug_assert!(component.exiting);

            debug_assert_eq!(component.public.num_handles.load(Ordering::Acquire), 0);

        });

        self.decrement_active_components();

        self.components.destroy(key.0);

    }

    // Tracking number of active interfaces and the active components

    #[inline]

    fn increment_active_components(&self) {

        let _old_val = self.active_elements.fetch_add(1, Ordering::AcqRel);

        debug_assert!(_old_val > 0); // can only create a component from a API/component, so can never be 0.

    }

    fn decrement_active_components(&self) {

        let old_val = self.active_elements.fetch_sub(1, Ordering::AcqRel);

        debug_assert!(old_val > 0); // something wrong with incr/decr logic

        let new_val = old_val - 1;

        if new_val == 0 {

            // Just to be sure, in case the last thing that gets destroyed is an

            // API instead of a thread.

            let _lock = self.work_queue.lock();

            self.work_condvar.notify_all();

        }

    }

}

use std::sync::Arc;

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

use crate::runtime2::poll::PollingClient;

use super::component::*;

use super::runtime::*;

/// Data associated with a scheduler thread

pub(crate) struct Scheduler {

    runtime: Arc<RuntimeInner>,

    polling: PollingClient,

    scheduler_id: u32,

    log_level: LogLevel,

}

pub(crate) struct SchedulerCtx<'a> {

    pub runtime: &'a RuntimeInner,

    pub polling: &'a PollingClient,

    pub id: u32,

    pub comp: u32,

    pub log_level: LogLevel,

}

impl<'a> SchedulerCtx<'a> {

    pub fn new(runtime: &'a RuntimeInner, polling: &'a PollingClient, id: u32, log_level: LogLevel) -> Self {

        return Self {

            runtime,

            polling,

            id,

            comp: 0,

            log_level,

        }

    }

    pub(crate) fn debug(&self, text: &str) {

        // TODO: Probably not always use colour

        }

    }

    pub(crate) fn info(&self, text: &str) {

            println!("[s:{:02}, c:{:03}] {}", self.id, self.comp, text);

        }

    }

    pub(crate) fn error(&self, text: &str) {

        // TODO: Probably not always use colour

        println!("[s:{:02}, c:{:03}] \x1b[0;31m{}\x1b[0m", self.id, self.comp, text);

    }

}

impl Scheduler {

    // public interface to thread

    pub fn new(runtime: Arc<RuntimeInner>, polling: PollingClient, scheduler_id: u32, log_level: LogLevel) -> Self {

        return Scheduler{ runtime, polling, scheduler_id, log_level }

    }

    pub fn run(&mut self) {

        let mut scheduler_ctx = SchedulerCtx::new(&*self.runtime, &self.polling, self.scheduler_id, self.log_level);

        'run_loop: loop {

            // Wait until we have something to do (or need to quit)

            let comp_key = self.runtime.take_work();

            if comp_key.is_none() {

                break 'run_loop;

            }

            let comp_key = comp_key.unwrap();

            let component = self.runtime.get_component(comp_key);

            scheduler_ctx.comp = comp_key.0;

            // Run the component until it no longer indicates that it needs to

            // be re-executed immediately.

            let mut new_scheduling = CompScheduling::Immediate;

            while let CompScheduling::Immediate = new_scheduling {

                while let Some(message) = component.inbox.pop() {

                    component.component.handle_message(&mut scheduler_ctx, &mut component.ctx, message);

                }

                new_scheduling = component.component.run(&mut scheduler_ctx, &mut component.ctx);

            }

            // Handle the new scheduling

            match new_scheduling {

                CompScheduling::Immediate => unreachable!(),

                CompScheduling::Requeue => { self.runtime.enqueue_work(comp_key); },

                CompScheduling::Sleep => { self.mark_component_as_sleeping(comp_key, component); },

                CompScheduling::Exit => {

                    component.component.on_shutdown(&scheduler_ctx);

                    self.mark_component_as_exiting(&scheduler_ctx, component);

                }

            }

        }

    }

    // local utilities

    /// Marks component as sleeping, if after marking itself as sleeping the

    /// inbox contains messages then the component will be immediately

    /// rescheduled. After calling this function the component should not be

    /// executed anymore.

    fn mark_component_as_sleeping(&self, key: CompKey, component: &mut RuntimeComp) {

        debug_assert_eq!(key.downgrade(), component.ctx.id); // make sure component matches key

        debug_assert_eq!(component.public.sleeping.load(Ordering::Acquire), false); // we're executing it, so it cannot be sleeping

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

        if component.inbox.can_pop() {

            let should_reschedule = component.public.sleeping

                .compare_exchange(true, false, Ordering::AcqRel, Ordering::Relaxed)

                .is_ok();

            if should_reschedule {

                self.runtime.enqueue_work(key);

            }

        }

    }

    /// Marks the component as exiting by removing the reference it holds to

    /// itself. Afterward the component will enter "normal" sleeping mode (if it

    /// has not yet been destroyed)

    fn mark_component_as_exiting(&self, sched_ctx: &SchedulerCtx, component: &mut RuntimeComp) {

        // If we didn't yet decrement our reference count, do so now

        let comp_key = unsafe{ component.ctx.id.upgrade() };

        if !component.exiting {

            component.exiting = true;

            let old_count = component.public.num_handles.fetch_sub(1, Ordering::AcqRel);

            let new_count = old_count - 1;

            if new_count == 0 {

                sched_ctx.runtime.destroy_component(comp_key);

                return;

            }

        }

        // Enter "regular" sleeping mode

        self.mark_component_as_sleeping(comp_key, component);

    }

}

use crate::protocol::*;

use crate::protocol::eval::*;

use crate::runtime2::runtime::*;

use crate::runtime2::component::{CompCtx, CompPDL};

mod error_handling;

const NUM_THREADS: u32 = 4;

pub(crate) fn compile_and_create_component(source: &str, routine_name: &str, args: ValueGroup) {

    let protocol = ProtocolDescription::parse(source.as_bytes())

        .expect("successful compilation");

        .expect("successful runtime startup");

    create_component(&runtime, "", routine_name, args);

}

pub(crate) fn create_component(rt: &Runtime, module_name: &str, routine_name: &str, args: ValueGroup) {

    let prompt = rt.inner.protocol.new_component(

        module_name.as_bytes(), routine_name.as_bytes(), args

    ).expect("create prompt");

    let reserved = rt.inner.start_create_pdl_component();

    let ctx = CompCtx::new(&reserved);

    let component = Box::new(CompPDL::new(prompt, 0));

    let (key, _) = rt.inner.finish_create_pdl_component(reserved, component, ctx, false);

    rt.inner.enqueue_work(key);

}

pub(crate) fn no_args() -> ValueGroup { ValueGroup::new_stack(Vec::new()) }

#[test]

fn test_component_creation() {

    let pd = ProtocolDescription::parse(b"

    primitive nothing_at_all() {

        s32 a = 5;

        auto b = 5 + a;

    }

    ").expect("compilation");

    for _i in 0..20 {

        create_component(&rt, "", "nothing_at_all", no_args());

    }

}

#[test]

fn test_component_communication() {

    let pd = ProtocolDescription::parse(b"

    primitive sender(out<u32> o, u32 outside_loops, u32 inside_loops) {

        u32 outside_index = 0;

        while (outside_index < outside_loops) {

            u32 inside_index = 0;

            sync while (inside_index < inside_loops) {

                put(o, inside_index);

                inside_index += 1;

            }

            outside_index += 1;

        }

    }

    primitive receiver(in<u32> i, u32 outside_loops, u32 inside_loops) {

        u32 outside_index = 0;

        while (outside_index < outside_loops) {

            u32 inside_index = 0;

            sync while (inside_index < inside_loops) {

                auto val = get(i);

                while (val != inside_index) {} // infinite loop if incorrect value is received

                inside_index += 1;

            }

            outside_index += 1;

        }

    }

    composite constructor() {

        channel o_orom -> i_orom;

        channel o_mrom -> i_mrom;

        channel o_ormm -> i_ormm;

        channel o_mrmm -> i_mrmm;

        // one round, one message per round

        new sender(o_orom, 1, 1);

        new receiver(i_orom, 1, 1);

        // multiple rounds, one message per round

        new sender(o_mrom, 5, 1);

        new receiver(i_mrom, 5, 1);

        // one round, multiple messages per round

        new sender(o_ormm, 1, 5);

        new receiver(i_ormm, 1, 5);

        // multiple rounds, multiple messages per round

        new sender(o_mrmm, 5, 5);

        new receiver(i_mrmm, 5, 5);

    }").expect("compilation");

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_intermediate_messenger() {

    let pd = ProtocolDescription::parse(b"

    primitive receiver<T>(in<T> rx, u32 num) {

        auto index = 0;

        while (index < num) {

            sync { auto v = get(rx); }

            index += 1;

        }

    }

    primitive middleman<T>(in<T> rx, out<T> tx, u32 num) {

        auto index = 0;

        while (index < num) {

            sync { put(tx, get(rx)); }

            index += 1;

        }

    }

    primitive sender<T>(out<T> tx, u32 num) {

        auto index = 0;

        while (index < num) {

            sync put(tx, 1337);

            index += 1;

        }

    }

    composite constructor_template<T>() {

        auto num = 0;

        channel<T> tx_a -> rx_a;

        channel tx_b -> rx_b;

        new sender(tx_a, 3);

        new middleman(rx_a, tx_b, 3);

        new receiver(rx_b, 3);

    }

    composite constructor() {

        new constructor_template<u16>();

        new constructor_template<u32>();

        new constructor_template<u64>();

        new constructor_template<s16>();

        new constructor_template<s32>();

        new constructor_template<s64>();

    }

    ").expect("compilation");

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_simple_select() {

    let pd = ProtocolDescription::parse(b"

    func infinite_assert<T>(T val, T expected) -> () {

        while (val != expected) { print(\"nope!\"); }

        return ();

    }

    primitive receiver(in<u32> in_a, in<u32> in_b, u32 num_sends) {

        auto num_from_a = 0;

        auto num_from_b = 0;

        while (num_from_a + num_from_b < 2 * num_sends) {

            sync select {

                auto v = get(in_a) -> {

                    print(\"got something from A\");

                    auto _ = infinite_assert(v, num_from_a);

                    num_from_a += 1;

                }

                auto v = get(in_b) -> {

                    print(\"got something from B\");

                    auto _ = infinite_assert(v, num_from_b);

                    num_from_b += 1;

                }

            }

        }

    }

    primitive sender(out<u32> tx, u32 num_sends) {

        auto index = 0;

        while (index < num_sends) {

            sync {

                put(tx, index);

                index += 1;

            }

        }

    }

    composite constructor() {

        auto num_sends = 1;

        channel tx_a -> rx_a;

        channel tx_b -> rx_b;

        new sender(tx_a, num_sends);

        new receiver(rx_a, rx_b, num_sends);

        new sender(tx_b, num_sends);

    }

    ").expect("compilation");

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_unguarded_select() {

    let pd = ProtocolDescription::parse(b"

    primitive constructor_outside_select() {

        u32 index = 0;

        while (index < 5) {

            sync select { auto v = () -> print(\"hello\"); }

            index += 1;

        }

    }

    primitive constructor_inside_select() {

        u32 index = 0;

        while (index < 5) {

            sync select { auto v = () -> index += 1; }

        }

    }

    ").expect("compilation");

    create_component(&rt, "", "constructor_outside_select", no_args());

    create_component(&rt, "", "constructor_inside_select", no_args());

}

#[test]

fn test_empty_select() {

    let pd = ProtocolDescription::parse(b"

    primitive constructor() {

        u32 index = 0;

        while (index < 5) {

            sync select {}

            index += 1;

        }

    }

    ").expect("compilation");

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_random_u32_temporary_thingo() {

    let pd = ProtocolDescription::parse(b"

    import std.random::random_u32;

    primitive random_taker(in<u32> generator, u32 num_values) {

        auto i = 0;

        while (i < num_values) {

            sync {

                auto a = get(generator);

            }

            i += 1;

        }

    }

    composite constructor() {

        channel tx -> rx;

        auto num_values = 25;

        new random_u32(tx, 1, 100, num_values);

        new random_taker(rx, num_values);

    }

    ").expect("compilation");

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_tcp_socket_http_request() {

    let _pd = ProtocolDescription::parse(b"

    import std.internet::*;

    primitive requester(out<Cmd> cmd_tx, in<u8[]> data_rx) {

        print(\"*** TCPSocket: Sending request\");

        sync {

            put(cmd_tx, Cmd::Send(b\"GET / HTTP/1.1\\r\\n\\r\\n\"));

        }

        print(\"*** TCPSocket: Receiving response\");

        auto buffer = {};

        auto done_receiving = false;

        sync while (!done_receiving) {

            put(cmd_tx, Cmd::Receive);

            auto data = get(data_rx);

            buffer @= data;

            // Completely crap detection of end-of-document. But here we go, we

            // try to detect the trailing </html>. Proper way would be to parse

            // for 'content-length' or 'content-encoding'

            s32 index = 0;

            s32 partial_length = cast(length(data) - 7);

            while (index < partial_length) {

                // No string conversion yet, so check byte buffer one byte at

                // a time.

                auto c1 = data[index];

                if (c1 == cast('<')) {

                    auto c2 = data[index + 1];

                    auto c3 = data[index + 2];

                    auto c4 = data[index + 3];

                    auto c5 = data[index + 4];

                    auto c6 = data[index + 5];

                    auto c7 = data[index + 6];

                    if ( // i.e. if (data[index..] == '</html>'

                        c2 == cast('/') && c3 == cast('h') && c4 == cast('t') &&

                        c5 == cast('m') && c6 == cast('l') && c7 == cast('>')

                    ) {

                        print(\"*** TCPSocket: Detected </html>\");

                        put(cmd_tx, Cmd::Finish);

                        done_receiving = true;

                    }

                }

                index += 1;

            }

        }

        print(\"*** TCPSocket: Requesting shutdown\");

        sync {

            put(cmd_tx, Cmd::Shutdown);

        }

    }

    composite main() {

        channel cmd_tx -> cmd_rx;

        channel data_tx -> data_rx;

        new tcp_client({142, 250, 179, 163}, 80, cmd_rx, data_tx); // port 80 of google

        new requester(cmd_tx, data_rx);

    }

    ").expect("compilation");

    // This test is disabled because it performs a HTTP request to google.

    // let rt = Runtime::new(1, true, pd).unwrap();

    // create_component(&rt, "", "main", no_args());

}

#[test]

fn test_sending_receiving_union() {

    let pd = ProtocolDescription::parse(b"

    union Cmd {

        Set(u8[]),

        Get,

        Shutdown,

    }

    primitive database(in<Cmd> rx, out<u8[]> tx) {

        auto stored = {};

        auto done = false;

        while (!done) {

            sync {

                auto command = get(rx);

                if (let Cmd::Set(bytes) = command) {

                    print(\"database: storing value\");

                    stored = bytes;

                } else if (let Cmd::Get = command) {

                    print(\"database: returning value\");