Do simple reads from disk

disk/src/lib.rs

@@ -1,3 +1,5 @@
+#[allow(unused_imports)]
+use tracing::{error, warn, info, debug, trace};
 use err::Error;
 use std::task::{Context, Poll};
 use std::pin::Pin;
@@ -11,28 +13,41 @@ pub async fn read_test_1() -> Result<netpod::BodyStream, Error> {
         .read(true)
         .open(path)
         .await?;
+    let meta = fin.metadata().await;
+    debug!("file meta {:?}", meta);
     let stream = netpod::BodyStream {
-        inner: Box::new(FileReader { file: fin }),
+        inner: Box::new(FileReader {
+            file: fin,
+            nreads: 0,
+        }),
     };
     Ok(stream)
 }
 
 struct FileReader {
     file: tokio::fs::File,
+    nreads: u32,
 }
 
 impl futures_core::Stream for FileReader {
     type Item = Result<bytes::Bytes, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let mut buf2 = bytes::BytesMut::with_capacity(13);
+        if self.nreads >= 10 {
-        if buf2.as_mut().len() != 13 {
+            return Poll::Ready(None);
+        }
+        let blen = 13;
+        let mut buf2 = bytes::BytesMut::with_capacity(blen);
+        buf2.resize(buf2.capacity(), 0);
+        if buf2.as_mut().len() != blen {
             panic!("todo prepare slice");
         }
         let mut buf = tokio::io::ReadBuf::new(buf2.as_mut());
-        let g = Pin::new(&mut self.file).poll_read(cx, &mut buf);
+        match Pin::new(&mut self.file).poll_read(cx, &mut buf) {
-        match g {
             Poll::Ready(Ok(_)) => {
+                info!("read from disk: {}   nreads {}", buf.filled().len(), self.nreads);
+                info!("buf2 len: {}", buf2.len());
+                self.nreads += 1;
                 Poll::Ready(Some(Ok(buf2.freeze())))
             }
             Poll::Ready(Err(e)) => {

err/src/lib.rs

@@ -3,6 +3,22 @@ pub struct Error {
     msg: String,
 }
 
+impl Error {
+    pub fn with_msg<S: Into<String>>(s: S) -> Self {
+        Self {
+            msg: s.into(),
+        }
+    }
+}
+
+impl From<String> for Error {
+    fn from(k: String) -> Self {
+        Self {
+            msg: k,
+        }
+    }
+}
+
 impl std::fmt::Display for Error {
     fn fmt(&self, fmt: &mut std::fmt::Formatter) -> std::fmt::Result {
         write!(fmt, "Error")

netpod/src/lib.rs

@@ -4,7 +4,7 @@ use err::Error;
 
 #[derive(Serialize, Deserialize)]
 pub struct AggQuerySingleChannel {
-    channel: String,
+    pub channel: String,
 }
 
 pub struct BodyStream {

retrieval/Cargo.toml

@@ -20,3 +20,5 @@ serde_json = "1.0"
 chrono = "0.4"
 clap = "3.0.0-beta.2"
 err = { path = "../err" }
+netpod = { path = "../netpod" }
+httpret = { path = "../httpret" }

retrieval/src/bin/retrieval.rs

@@ -1,6 +1,22 @@
+#[allow(unused_imports)]
+use tracing::{error, warn, info, debug, trace};
 use err::Error;
+use http::Method;
+use hyper::Body;
 
 pub fn main() {
+    match run(go()) {
+        Ok(k) => {
+            info!("{:?}", k);
+        }
+        Err(k) => {
+            error!("{:?}", k);
+        }
+    }
+}
+
+fn run<T, F: std::future::Future<Output=Result<T, Error>>>(f: F) -> Result<T, Error> {
+    tracing_init();
     tokio::runtime::Builder::new_multi_thread()
         .worker_threads(12)
         .max_blocking_threads(256)
@@ -8,36 +24,66 @@ pub fn main() {
         .build()
         .unwrap()
         .block_on(async {
-            tracing_subscriber::fmt()
+            f.await
-                .with_env_filter(tracing_subscriber::EnvFilter::new("daqbuffer=trace,tokio_postgres=info"))
-                .init();
-            go().await;
         })
 }
 
-async fn go() {
+async fn go() -> Result<(), Error> {
-    match inner().await {
-        Ok(_) => (),
-        Err(_) => ()
-    }
-}
-
-async fn inner() -> Result<(), Error> {
     use clap::Clap;
-    use retrieval::cli::Opts;
+    use retrieval::cli::{Opts, SubCmd};
-    let _opts = Opts::parse();
+    let opts = Opts::parse();
+    match opts.subcmd {
+        SubCmd::Retrieval(_subcmd) => {
+            trace!("testout");
+            info!("testout");
+            error!("testout");
+        }
+    }
     Ok(())
 }
 
 pub fn tracing_init() {
     tracing_subscriber::fmt()
-        .with_env_filter(tracing_subscriber::EnvFilter::new("retrieval=trace,tokio_postgres=info"))
+        //.with_timer(tracing_subscriber::fmt::time::uptime())
+        .with_target(true)
+        .with_thread_names(true)
+        //.with_max_level(tracing::Level::INFO)
+        .with_env_filter(tracing_subscriber::EnvFilter::new("info,retrieval=trace,tokio_postgres=info"))
         .init();
 }
 
 #[test]
-fn t1() {
+fn simple_fetch() {
-    tracing_init();
+    run(async {
-    tracing::error!("parsed");
+        let query = netpod::AggQuerySingleChannel {
-    //panic!();
+            channel: "S10CB01-RIQM-STA:DACI".into(),
+        };
+        let query_string = serde_json::to_string(&query).unwrap();
+        let host = tokio::spawn(httpret::host(8360));
+        let req = hyper::Request::builder()
+            .method(Method::POST)
+            .uri("http://localhost:8360/api/1/parsed_raw")
+            .body(query_string.into()).unwrap();
+        let client = hyper::Client::new();
+        let res = client.request(req).await?;
+        info!("client response {:?}", res);
+        let mut res_body = res.into_body();
+        use hyper::body::HttpBody;
+        loop {
+            match res_body.data().await {
+                Some(Ok(k)) => {
+                    info!("packet..  len {}", k.len());
+                }
+                Some(Err(e)) => {
+                    error!("{:?}", e);
+                }
+                None => {
+                    info!("response stream exhausted");
+                    break;
+                }
+            }
+        }
+        //Err::<(), _>(format!("test error").into())
+        Ok(())
+    }).unwrap();
 }

retrieval/src/cli.rs

@@ -11,7 +11,6 @@ pub struct Opts {
 
 #[derive(Debug, Clap)]
 pub enum SubCmd {
-    Version,
     Retrieval(Retrieval),
 }
 

retrieval/src/lib.rs

@@ -1,3 +1 @@
 pub mod cli;
-
-pub fn test() {}