WIP

streams/src/lib.rs

@@ -12,3 +12,4 @@ pub mod tcprawclient;
 #[cfg(test)]
 pub mod test;
 pub mod timebin;
+pub mod timebinnedjson;

streams/src/plaineventsjson.rs

@@ -1,28 +1,12 @@
 use crate::tcprawclient::open_tcp_streams;
-use bytes::Bytes;
 use err::Error;
-use futures_util::{Stream, StreamExt};
-use items::Sitemty;
 #[allow(unused)]
 use netpod::log::*;
 use netpod::Cluster;
 use serde::Serialize;
 use serde_json::Value as JsonValue;
-use std::pin::Pin;
-use std::task::{Context, Poll};
 use std::time::{Duration, Instant};
 
-pub struct BytesStream(Pin<Box<dyn Stream<Item = Sitemty<Bytes>> + Send>>);
-
-impl Stream for BytesStream {
-    type Item = Sitemty<Bytes>;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
-        StreamExt::poll_next_unpin(&mut self, cx)
-    }
-}
-
-// TODO remove?
 pub async fn plain_events_json<SER>(query: SER, cluster: &Cluster) -> Result<JsonValue, Error>
 where
     SER: Serialize,
@@ -41,7 +25,7 @@ where
         stream
     };
     let stream = { items_2::merger::Merger::new(inps, 1) };
-    let deadline = Instant::now() + Duration::from_millis(8000);
+    let deadline = Instant::now() + Duration::from_millis(3500);
     let events_max = 100;
     let collected = crate::collect::collect(stream, deadline, events_max).await?;
     let jsval = serde_json::to_value(&collected)?;

streams/src/timebin.rs

@@ -1,6 +1,7 @@
 use err::Error;
 use futures_util::{Future, FutureExt, Stream, StreamExt};
 use items::{sitem_data, RangeCompletableItem, Sitemty, StreamItem};
+//use items_0::{TimeBinnable, TimeBinner};
 use items_2::timebin::{TimeBinnable, TimeBinner};
 use netpod::log::*;
 use std::fmt;
@@ -11,19 +12,19 @@ use std::time::Instant;
 #[allow(unused)]
 macro_rules! trace2 {
     (D$($arg:tt)*) => ();
-    ($($arg:tt)*) => (eprintln!($($arg)*));
+    ($($arg:tt)*) => (trace!($($arg)*));
 }
 
 #[allow(unused)]
 macro_rules! trace3 {
     (D$($arg:tt)*) => ();
-    ($($arg:tt)*) => (eprintln!($($arg)*));
+    ($($arg:tt)*) => (trace!($($arg)*));
 }
 
 #[allow(unused)]
 macro_rules! trace4 {
     (D$($arg:tt)*) => ();
-    ($($arg:tt)*) => (eprintln!($($arg)*));
+    ($($arg:tt)*) => (trace!($($arg)*));
 }
 
 type MergeInp<T> = Pin<Box<dyn Stream<Item = Sitemty<T>> + Send>>;

streams/src/timebinnedjson.rs

@@ -0,0 +1,32 @@
+use crate::tcprawclient::open_tcp_streams;
+use err::Error;
+use futures_util::StreamExt;
+#[allow(unused)]
+use netpod::log::*;
+use netpod::Cluster;
+use serde::Serialize;
+use serde_json::Value as JsonValue;
+use std::time::{Duration, Instant};
+
+pub async fn timebinned_json<SER>(query: SER, cluster: &Cluster) -> Result<JsonValue, Error>
+where
+    SER: Serialize,
+{
+    // TODO should be able to ask for data-events only, instead of mixed data and status events.
+    let inps = open_tcp_streams::<_, items_2::channelevents::ChannelEvents>(&query, cluster).await?;
+    // TODO propagate also the max-buf-len for the first stage event reader:
+    let stream = { items_2::merger::Merger::new(inps, 1) };
+    let events_max = 10000;
+    let do_time_weight = true;
+    let deadline = Instant::now() + Duration::from_millis(7500);
+    let stream = Box::pin(stream);
+    let stream = crate::timebin::TimeBinnedStream::new(stream, Vec::new(), do_time_weight, deadline);
+    if false {
+        let mut stream = stream;
+        let _: Option<items::Sitemty<Box<dyn items_0::TimeBinned>>> = stream.next().await;
+        panic!()
+    }
+    let collected = crate::collect::collect(stream, deadline, events_max).await?;
+    let jsval = serde_json::to_value(&collected)?;
+    Ok(jsval)
+}