Initial commit

Author: skade

.gitignore

@@ -0,0 +1,2 @@
+/target
+**/*.rs.bk

Cargo.lock

@@ -0,0 +1,10 @@
+cargo-features = [ "edition" ]
+
+[package]
+name = "tbd-again"
+version = "0.1.0"
+authors = ["Florian Gilcher <[email protected]>"]
+edition = "2018"
+
+[dependencies]
+futures-preview = "0.3.0-alpha.2"

Cargo.toml

@@ -0,0 +1,136 @@
+#![feature(rust_2018_preview, async_await, await_macro, futures_api, pin, arbitrary_self_types)]
+
+extern crate futures;
+
+mod mini_exec;
+mod types;
+
+use std::future::FutureObj;
+use std::task::Executor;
+use futures::StreamExt;
+use futures::stream;
+use futures::future;
+
+use crate::types::*;
+
+#[derive(Debug, Clone)]
+struct Post {
+    id: u64,
+    content: String
+}
+
+#[derive(Debug, Clone)]
+struct Comment {
+    id: u64,
+    content: String,
+    post_id: u64
+}
+
+struct MemoryGateway  {
+    posts: Vec<Post>,
+    comments: Vec<Comment>
+}
+
+impl Gateway for MemoryGateway {}
+
+struct BlogRepository {
+    gateway: MemoryGateway
+}
+
+impl Repository for BlogRepository {}
+
+struct Posts;
+
+impl Relation<BlogRepository> for Posts {
+    type PrimaryKey = u64;
+    type Model = Post;
+    type Error = ();
+
+    type Stream = stream::Iter<std::vec::IntoIter<Post>>;
+    type Future = futures::future::Ready<Option<Post>>;
+
+    fn all(&self, repo: &BlogRepository) -> Self::Stream {
+        stream::iter(repo.gateway.posts.clone().into_iter())
+    }
+
+    fn one(&self, id: u64, repo: &BlogRepository) -> Self::Future {
+        future::ready(repo.gateway.posts.iter().find(|p| p.id == id).cloned())
+    }
+}
+
+
+struct Comments;
+
+impl Relation<BlogRepository> for Comments {
+    type PrimaryKey = u64;
+    type Model = Comment;
+    type Error = ();
+
+    type Stream = stream::Iter<std::vec::IntoIter<Comment>>;
+    type Future = futures::future::Ready<Option<Comment>>;
+
+    fn all(&self, repo: &BlogRepository) -> Self::Stream {
+        stream::iter(repo.gateway.comments.clone().into_iter())
+    }
+
+    fn one(&self, id: u64, repo: &BlogRepository) -> Self::Future {
+        future::ready(repo.gateway.comments.iter().find(|c| c.id == id).cloned())
+    }
+}
+
+async fn read_from_repos() {
+    let mut posts = vec![];
+
+    for id in 1..=3 {
+        posts.push(
+            Post {
+                id: id,
+                content: format!("Post number {}", id)
+            }
+        )
+    }
+
+    let mut comments = vec![];
+
+    for id in 1..=9 {
+        let post_id = id % 3;
+        comments.push(
+            Comment {
+                id: id,
+                content: format!("Comment number {} on post {}", id, post_id + 1),
+                post_id: post_id + 1
+            }
+        )
+    }
+
+    let gateway = MemoryGateway { posts, comments };
+    let repos = BlogRepository { gateway };
+
+    let f1 = Posts.all(&repos).for_each(|item|{
+        println!("{:?}", item);
+        future::ready(())
+    });
+
+    await!(f1);
+
+    let f2 = Comments.all(&repos).for_each(|item|{
+        println!("{:?}", item);
+        future::ready(())
+    });
+
+    await!(f2);
+
+    let model = await!(Comments.one(2, &repos));
+
+    println!("{:?}", model);
+}
+
+fn main() {
+    let executor = mini_exec::Executor::new();
+
+    (&executor).spawn_obj(FutureObj::new(Box::new(
+        read_from_repos()
+    ))).unwrap();
+
+    executor.run();
+}

src/main.rs

@@ -0,0 +1,79 @@
+#![feature(rust_2018_preview, async_await, await_macro, futures_api, pin, arbitrary_self_types)]
+
+use std::future::{Future, FutureObj};
+use std::mem::PinMut;
+
+use std::sync::{Arc, Mutex};
+use std::sync::mpsc::{sync_channel, SyncSender, SendError, Receiver};
+use std::task::{
+    self,
+    Executor as ExecutorTrait,
+    local_waker_from_nonlocal,
+    Poll,
+    SpawnErrorKind,
+    SpawnObjError,
+    Wake,
+};
+use std::time::Duration;
+
+static TIMEOUT: Duration = Duration::from_millis(100);
+
+pub struct Executor {
+    task_sender: SyncSender<Arc<Task>>,
+    task_receiver: Receiver<Arc<Task>>,
+}
+
+impl<'a> ExecutorTrait for &'a Executor {
+    fn spawn_obj(&mut self, future: FutureObj<'static, ()>)
+        -> Result<(), SpawnObjError>
+    {
+        let task = Arc::new(Task {
+            future: Mutex::new(Some(future)),
+            task_sender: self.task_sender.clone(),
+        });
+
+        self.task_sender.send(task).map_err(|SendError(task)| {
+            SpawnObjError {
+                kind: SpawnErrorKind::shutdown(),
+                future: task.future.lock().unwrap().take().unwrap(),
+            }
+        })
+    }
+}
+
+struct Task {
+    future: Mutex<Option<FutureObj<'static, ()>>>,
+    task_sender: SyncSender<Arc<Task>>,
+}
+
+impl Wake for Task {
+    fn wake(arc_self: &Arc<Self>) {
+        let cloned = arc_self.clone();
+        let _ = arc_self.task_sender.send(cloned);
+    }
+}
+
+impl Executor {
+    pub fn new() -> Self {
+        let (task_sender, task_receiver) = sync_channel(1000);
+        Executor { task_sender, task_receiver }
+    }
+
+    pub fn run(&self) {
+        let mut executor = &*self;
+        while let Ok(task) = self.task_receiver.recv_timeout(TIMEOUT) {
+            let mut future_slot = task.future.lock().unwrap();
+            if let Some(mut future) = future_slot.take() {
+                // Should we use the ref version here? might be nice to start
+                // w/o futures crate at first to show that it can be done,
+                // and just mention that there's a simple function to avoid
+                // the clone if anyone asks?
+                let waker = local_waker_from_nonlocal(task.clone());
+                let cx = &mut task::Context::new(&waker, &mut executor);
+                if let Poll::Pending = PinMut::new(&mut future).poll(cx) {
+                    *future_slot = Some(future);
+                }
+            }
+        }
+    }
+}