WIP on events_plain_json_00

streams/src/merge.rs

@@ -1,6 +1,3 @@
-// Sets up the raw tcp connections: disk::merge::mergedfromremotes::MergedFromRemotes
-// and then sets up a disk::merge::MergedStream
-
 pub mod mergedstream;
 
 use crate::frames::eventsfromframes::EventsFromFrames;
@@ -12,23 +9,24 @@ use items::frame::make_term_frame;
 use items::sitem_data;
 use items::EventQueryJsonStringFrame;
 use items::Sitemty;
-use items_2::ChannelEvents;
 use netpod::log::*;
 use netpod::Cluster;
 use std::pin::Pin;
 use tokio::io::AsyncWriteExt;
 use tokio::net::TcpStream;
 
-pub type ChannelEventsBoxedStream = Pin<Box<dyn Stream<Item = Sitemty<ChannelEvents>> + Send>>;
+pub type BoxedStream<T> = Pin<Box<dyn Stream<Item = Sitemty<T>> + Send>>;
 
-pub async fn open_tcp_streams(
-    query: &dyn erased_serde::Serialize,
-    cluster: &Cluster,
-) -> Result<Vec<ChannelEventsBoxedStream>, Error> {
+pub async fn open_tcp_streams<Q, T>(query: Q, cluster: &Cluster) -> Result<Vec<BoxedStream<T>>, Error>
+where
+    Q: serde::Serialize,
+    // Group bounds in new trait
+    T: items::FrameTypeInnerStatic + serde::de::DeserializeOwned + Send + Unpin + 'static,
+{
     // TODO when unit tests established, change to async connect:
     let mut streams = Vec::new();
     for node in &cluster.nodes {
-        debug!("x_processed_stream_from_node  to: {}:{}", node.host, node.port_raw);
+        debug!("open_tcp_streams  to: {}:{}", node.host, node.port_raw);
         let net = TcpStream::connect(format!("{}:{}", node.host, node.port_raw)).await?;
         let qjs = serde_json::to_string(&query)?;
         let (netin, mut netout) = net.into_split();
@@ -42,8 +40,7 @@ pub async fn open_tcp_streams(
         netout.forget();
         // TODO for images, we need larger buffer capacity
         let frames = InMemoryFrameAsyncReadStream::new(netin, 1024 * 128);
-        type ITEM = ChannelEvents;
-        let stream = EventsFromFrames::<_, ITEM>::new(frames);
+        let stream = EventsFromFrames::<_, T>::new(frames);
         streams.push(Box::pin(stream) as _);
     }
     Ok(streams)

streams/src/merge/mergedstream.rs

@@ -296,7 +296,8 @@ where
 
 #[cfg(test)]
 mod test {
-    use items_2::{ChannelEvents, Empty};
+    use items_2::channelevents::ChannelEvents;
+    use items_2::Empty;
 
     #[test]
     fn merge_channel_events() {

streams/src/plaineventsjson.rs

@@ -1,16 +1,16 @@
 use crate::merge::open_tcp_streams;
 use bytes::Bytes;
 use err::Error;
-use futures_util::{future, stream, FutureExt, Stream, StreamExt};
-use items::streams::collect_plain_events_json;
-use items::{sitem_data, RangeCompletableItem, Sitemty, StreamItem};
-use items_2::ChannelEvents;
+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;
+use std::time::{Duration, Instant};
 
 pub struct BytesStream(Pin<Box<dyn Stream<Item = Sitemty<Bytes>> + Send>>);
 
@@ -22,52 +22,27 @@ impl Stream for BytesStream {
     }
 }
 
-pub async fn plain_events_json<SER>(query: SER, cluster: &Cluster) -> Result<BytesStream, Error>
+// TODO remove?
+pub async fn plain_events_json<SER>(query: SER, cluster: &Cluster) -> Result<JsonValue, Error>
 where
     SER: Serialize,
 {
-    let inps = open_tcp_streams(&query, cluster).await?;
-    let mut merged = items_2::merger_cev::ChannelEventsMerger::new(inps);
-    let timeout = Duration::from_millis(2000);
+    // 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:
+    #[cfg(NOTHING)]
+    let stream = {
+        let mut it = inps.into_iter();
+        let inp0 = it.next().unwrap();
+        let inp1 = it.next().unwrap();
+        let inp2 = it.next().unwrap();
+        let stream = inp0.chain(inp1).chain(inp2);
+        stream
+    };
+    let stream = { items_2::merger::Merger::new(inps, 1) };
+    let deadline = Instant::now() + Duration::from_millis(2000);
     let events_max = 100;
-    let do_log = false;
-    let mut coll = None;
-    while let Some(item) = merged.next().await {
-        let item = item?;
-        match item {
-            StreamItem::DataItem(item) => match item {
-                RangeCompletableItem::RangeComplete => todo!(),
-                RangeCompletableItem::Data(item) => match item {
-                    ChannelEvents::Events(mut item) => {
-                        if coll.is_none() {
-                            coll = Some(item.new_collector());
-                        }
-                        let coll = coll
-                            .as_mut()
-                            .ok_or_else(|| Error::with_msg_no_trace(format!("no collector")))?;
-                        coll.ingest(&mut item);
-                    }
-                    ChannelEvents::Status(_) => todo!(),
-                },
-            },
-            StreamItem::Log(item) => {
-                info!("log {item:?}");
-            }
-            StreamItem::Stats(item) => {
-                info!("stats {item:?}");
-            }
-        }
-    }
-    // TODO compare with
-    // streams::collect::collect_plain_events_json
-    // and remove duplicate functionality.
-    let mut coll = coll.ok_or_else(|| Error::with_msg_no_trace(format!("no collector created")))?;
-    let res = coll.result()?;
-    // TODO factor the conversion of the result out to a higher level.
-    // The output of this function should again be collectable, maybe even binnable and otherwise processable.
-    let js = serde_json::to_vec(&res)?;
-    let item = sitem_data(Bytes::from(js));
-    let stream = stream::once(future::ready(item));
-    let stream = BytesStream(Box::pin(stream));
-    Ok(stream)
+    let collected = crate::collect::collect(stream, deadline, events_max).await?;
+    let jsval = serde_json::to_value(&collected)?;
+    Ok(jsval)
 }

streams/src/test.rs

@@ -6,8 +6,9 @@ mod timebin;
 use err::Error;
 use futures_util::{stream, Stream};
 use items::{sitem_data, Sitemty};
+use items_2::channelevents::ChannelEvents;
 use items_2::eventsdim0::EventsDim0;
-use items_2::{ChannelEvents, Empty};
+use items_2::Empty;
 use netpod::timeunits::SEC;
 use std::pin::Pin;
 
@@ -46,7 +47,7 @@ fn merge_mergeable_00() -> Result<(), Error> {
     let fut = async {
         let inp0 = inmem_test_events_d0_i32_00();
         let inp1 = inmem_test_events_d0_i32_01();
-        let mut merger = items_2::merger::Merger::new(vec![inp0, inp1], 4);
+        let _merger = items_2::merger::Merger::new(vec![inp0, inp1], 4);
         Ok(())
     };
     runfut(fut)

streams/src/test/collect.rs

@@ -2,6 +2,7 @@ use crate::test::runfut;
 use err::Error;
 use futures_util::stream;
 use items::sitem_data;
+use items_2::eventsdim0::EventsDim0CollectorOutput;
 use items_2::testgen::make_some_boxed_d0_f32;
 use netpod::timeunits::SEC;
 use std::time::{Duration, Instant};
@@ -15,7 +16,14 @@ fn collect_channel_events() -> Result<(), Error> {
         let deadline = Instant::now() + Duration::from_millis(4000);
         let events_max = 10000;
         let res = crate::collect::collect(stream, deadline, events_max).await?;
-        eprintln!("collected result: {res:?}");
+        //eprintln!("collected result: {res:?}");
+        if let Some(res) = res.as_any_ref().downcast_ref::<EventsDim0CollectorOutput<f32>>() {
+            eprintln!("Great, a match");
+            eprintln!("{res:?}");
+            assert_eq!(res.len(), 40);
+        } else {
+            return Err(Error::with_msg(format!("bad type of collected result")));
+        }
         Ok(())
     };
     runfut(fut)

streams/src/test/timebin.rs

@@ -3,8 +3,8 @@ use err::Error;
 use futures_util::{stream, StreamExt};
 use items::{sitem_data, RangeCompletableItem, StreamItem};
 use items_2::binsdim0::BinsDim0;
+use items_2::channelevents::{ChannelEvents, ConnStatus, ConnStatusEvent};
 use items_2::testgen::make_some_boxed_d0_f32;
-use items_2::{ChannelEvents, ConnStatus, ConnStatusEvent};
 use netpod::timeunits::{MS, SEC};
 use std::collections::VecDeque;
 use std::time::{Duration, Instant};