#![feature(
    arbitrary_self_types,
    async_await,
    await_macro,
    futures_api,
    pin
)]

extern crate futures;

use futures::{channel::mpsc, prelude::*};
use std::{cell::RefCell, collections::LinkedList};

pub use futures::{channel::mpsc::SendError, task::SpawnError};

type LocalMessage = Box<Send + 'static>;
type LocalReceiver = mpsc::Receiver<LocalMessage>;

struct LocalChannel {
    receiver: LocalReceiver,
    waiting:  LinkedList<LocalMessage>,
}

thread_local! {
    static MY_CHANNEL: RefCell<Option<LocalChannel>> = RefCell::new(None);
}

pub fn spawn<Fut>(_fut: Fut) -> impl Future<Output = Result<(), SpawnError>>
where
    Fut: Future<Output = ()> + Send + 'static,
{
    future::lazy(|_| Ok(()))
}

#[doc(hidden)]
pub async fn __receive<WantFn, Fut>(want: WantFn) -> LocalMessage
where
    Fut: Future<Output = bool>,
    WantFn: Fn(&LocalMessage) -> Fut,
{
    let mut chan = MY_CHANNEL.with(|c| c.borrow_mut().take()).unwrap();

    loop {
        let msg = await!(chan.receiver.next()).unwrap();
        if await!(want(&msg)) {
            MY_CHANNEL.with(|c| *c.borrow_mut() = Some(chan));
            return msg;
        }
        chan.waiting.push_back(msg);
    }
}

fn main() {
    futures::executor::ThreadPool::new()
        .unwrap()
        .run(spawn(
            async {
                await!(__receive(|_| async { false }));
            },
        )).unwrap();
}