Initial commit

Author: carllerche

.gitignore

@@ -0,0 +1,3 @@
+/target
+**/*.rs.bk
+Cargo.lock

Cargo.toml

@@ -0,0 +1,5 @@
+[workspace]
+members = [
+    "async-stream",
+    "async-stream-impl",
+]

async-stream-impl/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "async-stream-impl"
+version = "0.1.0"
+authors = ["Carl Lerche <[email protected]>"]
+edition = "2018"
+
+[lib]
+proc-macro = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+proc-macro-hack = "0.5.8"
+syn = { version = "0.15.43", features = ["visit-mut"]}
+quote = "0.6.13"

async-stream-impl/src/lib.rs

@@ -0,0 +1,66 @@
+extern crate proc_macro;
+
+use proc_macro::TokenStream;
+use proc_macro_hack::proc_macro_hack;
+use quote::quote;
+use syn::Token;
+use syn::parse::{Parse, ParseStream, Result};
+use syn::visit_mut::VisitMut;
+
+struct AsyncStreamImpl {
+    stmts: Vec<syn::Stmt>,
+}
+
+struct Scrub {
+    yielder: syn::Ident,
+    unit: Box<syn::Expr>,
+}
+
+impl Parse for AsyncStreamImpl {
+    fn parse(input: ParseStream) -> Result<Self> {
+        let yielder: syn::Ident = input.parse()?;
+        input.parse::<Token![,]>()?;
+
+        let mut stmts = vec![];
+        let mut scrub = Scrub {
+            yielder,
+            unit: syn::parse_quote!(()),
+        };
+
+        while !input.is_empty() {
+            let mut stmt = input.parse()?;
+            scrub.visit_stmt_mut(&mut stmt);
+            stmts.push(stmt);
+        }
+
+        Ok(AsyncStreamImpl { stmts })
+    }
+}
+
+impl VisitMut for Scrub {
+    fn visit_expr_mut(&mut self, i: &mut syn::Expr) {
+        match i {
+            syn::Expr::Yield(expr) => {
+                let value_expr = if let Some(ref e) = expr.expr {
+                    e
+                } else {
+                    &self.unit
+                };
+
+                let ident = &self.yielder;
+                *i = syn::parse_quote! { #ident.send(#value_expr).await };
+            }
+            expr => syn::visit_mut::visit_expr_mut(self, expr),
+        }
+    }
+}
+
+#[proc_macro_hack]
+pub fn async_stream_impl(input: TokenStream) -> TokenStream {
+    let AsyncStreamImpl {
+        stmts,
+    } = syn::parse_macro_input!(input as AsyncStreamImpl);
+
+    quote!(#(#stmts)*).into()
+    // quote!(#input).into()
+}

async-stream/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "async-stream"
+version = "0.1.0"
+authors = ["Carl Lerche <[email protected]>"]
+edition = "2018"
+
+[dependencies]
+async-stream-impl = { path = "../async-stream-impl" }
+futures-core-preview = "=0.3.0-alpha.17"
+proc-macro-hack = "0.5.8"
+
+[dev-dependencies]
+tokio = "=0.2.0-alpha.1"
+futures-util-preview = "=0.3.0-alpha.17"

async-stream/examples/tcp_accept.rs

@@ -0,0 +1,24 @@
+#![feature(async_await)]
+
+use async_stream::stream;
+use futures_util::pin_mut;
+use futures_util::stream::StreamExt;
+use tokio::net::TcpListener;
+
+#[tokio::main]
+async fn main() {
+    let addr = "127.0.0.1:0".parse().unwrap();
+    let mut listener = TcpListener::bind(&addr).unwrap();
+
+    let incoming = stream! {
+        loop {
+            let (socket, _) = listener.accept().await.unwrap();
+            yield socket;
+        }
+    };
+    pin_mut!(incoming);
+
+    while let Some(v) = incoming.next().await {
+        println!("handle = {:?}", v);
+    }
+}

async-stream/src/async_stream.rs

@@ -0,0 +1,27 @@
+use crate::yielder::Receiver;
+
+use futures_core::Stream;
+use std::future::Future;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+pub struct AsyncStream<T, U> {
+    rx: Receiver<T>,
+    generator: U,
+}
+
+impl<T, U> AsyncStream<T, U> {
+    pub fn new(rx: Receiver<T>, generator: U) -> AsyncStream<T, U> {
+        AsyncStream { rx, generator }
+    }
+}
+
+impl<T, U> Stream for AsyncStream<T, U>
+where U: Future<Output = ()>
+{
+    type Item = T;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
+        unimplemented!();
+    }
+}

async-stream/src/lib.rs

@@ -0,0 +1,28 @@
+#![feature(async_await)]
+
+mod async_stream;
+#[doc(hidden)]
+pub mod yielder;
+
+// Used by the macro, but not intended to be accessed publically.
+#[doc(hidden)]
+pub use crate::{
+    async_stream::AsyncStream,
+};
+
+use proc_macro_hack::proc_macro_hack;
+
+#[doc(hidden)]
+#[proc_macro_hack]
+pub use async_stream_impl::async_stream_impl;
+
+/// Asynchronous stream
+#[macro_export]
+macro_rules! stream {
+    ($($body:tt)*) => {{
+        let (mut __yield_tx, __yield_rx) = $crate::yielder::pair();
+        $crate::AsyncStream::new(__yield_rx, async move {
+            $crate::async_stream_impl!(__yield_tx, $($body)*)
+        })
+    }}
+}

async-stream/src/yielder.rs

@@ -0,0 +1,19 @@
+use std::marker::PhantomData;
+
+pub struct Sender<T> {
+    _p: PhantomData<T>,
+}
+
+pub struct Receiver<T> {
+    _p: PhantomData<T>,
+}
+
+pub fn pair<T>() -> (Sender<T>, Receiver<T>) {
+    unimplemented!();
+}
+
+impl<T> Sender<T> {
+    pub async fn send(&mut self, value: T) {
+        unimplemented!();
+    }
+}

async-stream/tests/basic.rs

@@ -0,0 +1,6 @@
+use async_stream::stream;
+
+#[test]
+fn smoke() {
+    stream!();
+}