WIP

streams/src/collect.rs

@@ -8,6 +8,7 @@ use items_0::streamitem::RangeCompletableItem;
 use items_0::streamitem::Sitemty;
 use items_0::streamitem::StatsItem;
 use items_0::streamitem::StreamItem;
+use items_0::transform::EventStream;
 use items_0::WithLen;
 use netpod::log::*;
 use netpod::range::evrange::SeriesRange;
@@ -36,6 +37,10 @@ macro_rules! trace4 {
     ($($arg:tt)*) => (eprintln!($($arg)*));
 }
 
+pub struct Collect {
+    inp: EventStream,
+}
+
 async fn collect_in_span<T, S>(
     stream: S,
     deadline: Instant,

streams/src/lib.rs

@@ -13,3 +13,4 @@ pub mod tcprawclient;
 pub mod test;
 pub mod timebin;
 pub mod timebinnedjson;
+pub mod transform;

streams/src/plaineventsjson.rs

@@ -91,7 +91,7 @@ pub async fn plain_events_json(evq: &PlainEventsQuery, chconf: &ChConf, cluster:
         info!("item after merge: {item:?}");
         item
     });
-    let stream = RangeFilter2::new(stream, evq.range().try_into()?, evquery.one_before_range());
+    //let stream = RangeFilter2::new(stream, evq.range().try_into()?, evquery.one_before_range());
     #[cfg(DISABLED)]
     let stream = stream.map(|item| {
         info!("item after rangefilter: {item:?}");

streams/src/test/timebin.rs

@@ -1,6 +1,7 @@
 use crate::collect::collect;
 use crate::generators::GenerateI32;
 use crate::test::runfut;
+use crate::transform::build_event_transform;
 use chrono::DateTime;
 use chrono::Utc;
 use err::Error;
@@ -255,6 +256,6 @@ fn transform_chain_correctness_01() -> Result<(), Error> {
     type STY = f32;
     let tq = TransformQuery::default_time_binned();
     let empty = EventsDim0::<STY>::empty();
-    tq.build_event_transform(empty.into())?;
+    build_event_transform(&tq, empty.into())?;
     Ok(())
 }

streams/src/transform.rs

@@ -0,0 +1,62 @@
+use err::Error;
+use futures_util::StreamExt;
+use items_0::transform::CollectableStream;
+use items_0::transform::EventStream;
+use items_0::transform::TransformEvent;
+use items_2::transform::make_transform_identity;
+use items_2::transform::make_transform_min_max_avg;
+use items_2::transform::make_transform_pulse_id_diff;
+use query::transform::EventTransformQuery;
+use query::transform::TimeBinningTransformQuery;
+use query::transform::TransformQuery;
+
+pub fn build_event_transform(tr: &TransformQuery, inp: EventStream) -> Result<TransformEvent, Error> {
+    let trev = tr.get_tr_event();
+    match trev {
+        EventTransformQuery::ValueFull => Ok(make_transform_identity()),
+        EventTransformQuery::MinMaxAvgDev => Ok(make_transform_min_max_avg()),
+        EventTransformQuery::ArrayPick(..) => Err(Error::with_msg_no_trace(format!(
+            "build_event_transform don't know what to do {trev:?}"
+        ))),
+        EventTransformQuery::PulseIdDiff => Ok(make_transform_pulse_id_diff()),
+        EventTransformQuery::EventBlobsVerbatim => Err(Error::with_msg_no_trace(format!(
+            "build_event_transform don't know what to do {trev:?}"
+        ))),
+        EventTransformQuery::EventBlobsUncompressed => Err(Error::with_msg_no_trace(format!(
+            "build_event_transform don't know what to do {trev:?}"
+        ))),
+    }
+}
+
+pub fn build_full_transform_collected(tr: &TransformQuery, inp: EventStream) -> Result<CollectableStream, Error> {
+    // TODO this must return a Stream!
+    //let evs = build_event_transform(tr, inp)?;
+    let trtb = tr.get_tr_time_binning();
+    use futures_util::Stream;
+    use items_0::collect_s::Collectable;
+    use items_0::streamitem::RangeCompletableItem;
+    use items_0::streamitem::Sitemty;
+    use items_0::streamitem::StreamItem;
+    use std::pin::Pin;
+    let a: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Collectable>>> + Send>> = Box::pin(inp.0.map(|item| match item {
+        Ok(item) => match item {
+            StreamItem::DataItem(item) => match item {
+                RangeCompletableItem::Data(item) => {
+                    let item: Box<dyn Collectable> = Box::new(item);
+                    Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))
+                }
+                RangeCompletableItem::RangeComplete => Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete)),
+            },
+            StreamItem::Log(item) => Ok(StreamItem::Log(item)),
+            StreamItem::Stats(item) => Ok(StreamItem::Stats(item)),
+        },
+        Err(e) => Err(e),
+    }));
+    let stream: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Collectable>>> + Send>> =
+        Box::pin(futures_util::stream::empty());
+    match trtb {
+        TimeBinningTransformQuery::None => Ok(CollectableStream(stream)),
+        TimeBinningTransformQuery::TimeWeighted => todo!(),
+        TimeBinningTransformQuery::Unweighted => todo!(),
+    }
+}