working example

Cargo.toml

@@ -8,4 +8,7 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-chain = {package = "markov", version = "1.1.0"}
+chain = {package = "markov", version = "1.1.0"}
+tokio = {version = "0.2", features=["full"]}
+warp = "0.2"
+pretty_env_logger = "0.4.0"

src/main.rs

@@ -1,47 +1,81 @@
 use chain::{
     Chain,
 };
+use warp::{
+    Filter,
+    Buf,
+};
 use std::{
-    io::{
-	BufRead,
-	self,
+    sync::Arc,
+};
+use tokio::{
+    sync::{
+	RwLock,
     },
+    stream::{Stream,StreamExt,},
+    prelude::*,
 };
 
-fn buffered_read_all_lines<T: BufRead+?Sized, F: FnMut(&str) -> io::Result<()>>(input: &mut T, mut then: F) -> io::Result<usize>
-{
-    let mut buffer = String::new();
-    let mut read;
-    let mut total=0;
-    while {read = input.read_line(&mut buffer)?; read!=0} {
-	then(&buffer[..])?;
-	buffer.clear();
-	total += read;
+const MAX_CONTENT_LENGTH: u64 = 1024 * 16;
+const MAX_GEN_SIZE: usize = 256;
+
+async fn full_body(chain: &mut Chain<String>, mut body: impl Unpin + Stream<Item = Result<impl Buf, impl std::error::Error + 'static>>) -> Result<(), Box<dyn std::error::Error>> {
+    let mut buffer = Vec::new();
+    
+    while let Some(buf) = body.next().await {
+	let mut body = buf?;
+	while body.has_remaining() {
+	    buffer.extend_from_slice(body.bytes());
+            let cnt = body.bytes().len();
+            body.advance(cnt);
+	}
     }
-    Ok(total)
+
+    let buffer = std::str::from_utf8(&buffer[..])?;
+    chain.feed_str(buffer);
+    Ok(())
 }
 
-fn main() {
-    let stdin = io::stdin();
-    let mut stdin = stdin.lock();
-    let mut chain = Chain::new();
+#[tokio::main]
+async fn main() {
+    pretty_env_logger::init();
+    
+    let chain = Arc::new(RwLock::new(Chain::new()));
+    let chain = warp::any().map(move || Arc::clone(&chain));
 
-    buffered_read_all_lines(&mut stdin, |string| {
-        chain.feed(&string.split_whitespace()
-                   .filter(|word| !word.is_empty())
-                   .map(|s| s.to_owned()).collect::<Vec<_>>());
-	
-	Ok(())
-    }).expect("Failed to read from stdin");
+    let push = warp::put()
+        .and(chain.clone())
+        .and(warp::path("put"))
+        .and(warp::body::content_length_limit(MAX_CONTENT_LENGTH))
+        .and(warp::body::stream())
+        .and_then(|chain: Arc<RwLock<Chain<String>>>, buf| {
+	    async move {
+		use std::ops::DerefMut;
+		let res = format!("{:?}", full_body(chain.write().await.deref_mut(), buf).await);
+		Ok::<String, std::convert::Infallible>(res)
+	    }
+	});
 
-    if !chain.is_empty() {
-	if let Some(num) = std::env::args().skip(1).next() {
-	    let sz: usize = num.parse().expect("Cannot parse number of tokens to generate");
-	    for string in chain.str_iter_for(sz) {
-		println!("{}", string);
+    let read = warp::get()
+        .and(chain.clone())
+        .and(warp::path("get"))
+        .and(warp::path::param().map(|opt: usize| Some(opt)).or(warp::any().map(|| Option::<usize>::None)).unify())
+        .and_then(|chain: Arc<RwLock<Chain<String>>>, num: Option<usize>| {
+	    async move {
+		let chain = chain.read().await;
+		if chain.is_empty() {
+		    Ok(String::default())
+		} else {
+		    match num {
+			Some(num) if num < MAX_GEN_SIZE => Ok(chain.str_iter_for(num).collect()),
+			_ => Ok::<String, std::convert::Infallible>(chain.generate_str()),
+		    }
+		}
 	    }
-	} else {
-	    println!("{}", chain.generate_str());
-	}
-    }
+	});
+
+    warp::serve(push
+		.or(read))
+	.bind_with_graceful_shutdown(([127,0,0,1], 8777), async { tokio::signal::ctrl_c().await.unwrap(); }).1
+	.await
 }