daqbuf/daqbuffer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP typechecks and basic run

Browse Source
This commit is contained in:
Dominik Werder
2024-10-24 16:12:29 +02:00
parent e6ece07137
commit 130dabce40
20 changed files with 418 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
crates/streams/src/collect.rs
View File

@@ -3,9 +3,9 @@ use futures_util::Future;
use futures_util::FutureExt;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::collect_s::Collectable;
use items_0::collect_s::Collected;
use items_0::collect_s::Collector;
use items_0::collect_s::CollectableDyn;
use items_0::collect_s::CollectedDyn;
use items_0::collect_s::CollectorDyn;
use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
use items_0::streamitem::StatsItem;
@@ -47,12 +47,12 @@ pub enum CollectResult<T> {
}
pub struct Collect {
inp: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Collectable>>> + Send>>,
inp: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn CollectableDyn>>> + Send>>,
events_max: u64,
bytes_max: u64,
range: Option<SeriesRange>,
binrange: Option<BinnedRangeEnum>,
collector: Option<Box<dyn Collector>>,
collector: Option<Box<dyn CollectorDyn>>,
range_final: bool,
timeout: bool,
timer: Pin<Box<dyn Future<Output = ()> + Send>>,
@@ -61,7 +61,7 @@ pub struct Collect {
impl Collect {
pub fn new(
inp: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Collectable>>> + Send>>,
inp: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn CollectableDyn>>> + Send>>,
deadline: Instant,
events_max: u64,
bytes_max: u64,
@@ -83,7 +83,7 @@ impl Collect {
}
}
fn handle_item(&mut self, item: Sitemty<Box<dyn Collectable>>) -> Result<(), Error> {
fn handle_item(&mut self, item: Sitemty<Box<dyn CollectableDyn>>) -> Result<(), Error> {
match item {
Ok(item) => match item {
StreamItem::DataItem(item) => match item {
@@ -161,7 +161,7 @@ impl Collect {
}
impl Future for Collect {
type Output = Result<CollectResult<Box<dyn Collected>>, Error>;
type Output = Result<CollectResult<Box<dyn CollectedDyn>>, Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
use Poll::*;
@@ -226,13 +226,13 @@ async fn collect_in_span<T, S>(
events_max: u64,
range: Option<SeriesRange>,
binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn Collected>, Error>
) -> Result<Box<dyn CollectedDyn>, Error>
where
S: Stream<Item = Sitemty<T>> + Unpin,
T: Collectable,
T: CollectableDyn,
{
info!("collect events_max {events_max} deadline {deadline:?}");
let mut collector: Option<Box<dyn Collector>> = None;
let mut collector: Option<Box<dyn CollectorDyn>> = None;
let mut stream = stream;
let deadline = deadline.into();
let mut range_complete = false;
@@ -329,10 +329,10 @@ pub async fn collect<T, S>(
events_max: u64,
range: Option<SeriesRange>,
binrange: Option<BinnedRangeEnum>,
) -> Result<Box<dyn Collected>, Error>
) -> Result<Box<dyn CollectedDyn>, Error>
where
S: Stream<Item = Sitemty<T>> + Unpin,
T: Collectable + WithLen + fmt::Debug,
T: CollectableDyn + WithLen + fmt::Debug,
{
let span = span!(Level::INFO, "collect");
collect_in_span(stream, deadline, events_max, range, binrange)

13
crates/streams/src/plaineventsjson.rs
View File

@@ -9,7 +9,7 @@ use crate::tcprawclient::OpenBoxedBytesStreamsBox;
use err::thiserror;
use err::ThisError;
use futures_util::StreamExt;
use items_0::collect_s::Collectable;
use items_0::collect_s::CollectableDyn;
use items_0::on_sitemty_data;
use netpod::log::*;
use netpod::ChannelTypeConfigGen;
@@ -27,6 +27,7 @@ pub enum Error {
Stream(#[from] crate::plaineventsstream::Error),
Collect(err::Error),
Json(#[from] serde_json::Error),
Err(err::Error),
}
pub async fn plain_events_json(
@@ -56,23 +57,23 @@ pub async fn plain_events_json(
for (&ts, val) in g.tss.iter().zip(g.values.iter()) {
out.push_back(ts, val.ix(), val.name_string());
}
let k: Box<dyn Collectable> = Box::new(out);
let k: Box<dyn CollectableDyn> = Box::new(out);
Ok(StreamItem::DataItem(RangeCompletableItem::Data(k)))
} else {
trace!("consider container channel events other events {}", k.type_name());
let k: Box<dyn Collectable> = Box::new(k);
let k: Box<dyn CollectableDyn> = Box::new(k);
Ok(StreamItem::DataItem(RangeCompletableItem::Data(k)))
}
}
items_2::channelevents::ChannelEvents::Status(_) => {
trace!("consider container channel events status {}", k.type_name());
let k: Box<dyn Collectable> = Box::new(k);
let k: Box<dyn CollectableDyn> = Box::new(k);
Ok(StreamItem::DataItem(RangeCompletableItem::Data(k)))
}
}
} else {
trace!("consider container else {}", k.type_name());
let k: Box<dyn Collectable> = Box::new(k);
let k: Box<dyn CollectableDyn> = Box::new(k);
Ok(StreamItem::DataItem(RangeCompletableItem::Data(k)))
}
})
@@ -95,7 +96,7 @@ pub async fn plain_events_json(
.map_err(Error::Collect)?;
debug!("plain_events_json collected");
if let CollectResult::Some(x) = collected {
let jsval = serde_json::to_value(&x)?;
let jsval = x.to_json_value().map_err(|e| Error::Err(e))?;
debug!("plain_events_json json serialized");
Ok(CollectResult::Some(jsval))
} else {

4
crates/streams/src/test/timebin.rs
View File

@@ -220,7 +220,9 @@ fn time_bin_02() -> Result<(), Error> {
} else {
let res = collect(binned_stream, deadline, 200, None, Some(binned_range)).await?;
assert_eq!(res.len(), expected_bin_count);
let d = res.to_json_result()?.to_json_bytes()?;
// use crate::json_stream::JsonBytes;
let v = res.to_json_value()?;
let d = serde_json::to_vec(&v)?;
let s = String::from_utf8_lossy(&d);
eprintln!("{s}");
let jsval: JsValue = serde_json::from_slice(&d)?;

32
crates/streams/src/timebinnedjson.rs
View File

@@ -17,7 +17,7 @@ use futures_util::Future;
use futures_util::FutureExt;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::collect_s::Collectable;
use items_0::collect_s::CollectableDyn;
use items_0::on_sitemty_data;
use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
@@ -374,7 +374,7 @@ async fn timebinned_stream(
open_bytes: OpenBoxedBytesStreamsBox,
cache_read_provider: Arc<dyn CacheReadProvider>,
events_read_provider: Arc<dyn EventsReadProvider>,
) -> Result<Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Collectable>>> + Send>>, Error> {
) -> Result<Pin<Box<dyn Stream<Item = Sitemty<Box<dyn CollectableDyn>>> + Send>>, Error> {
use netpod::query::CacheUsage;
let cache_usage = query.cache_usage().unwrap_or(CacheUsage::V0NoCache);
match cache_usage.clone() {
@@ -407,19 +407,21 @@ async fn timebinned_stream(
events_read_provider,
)
.map_err(Error::from_string)?;
// let stream = stream.map(|item| {
// on_sitemty_data!(item, |k: items_0::timebin::BinsBoxed| {
// let ret = k.to_old_time_binned();
// Ok(StreamItem::DataItem(RangeCompletableItem::Data(ret)))
// })
// });
let stream = stream.map(|item| {
on_sitemty_data!(item, |k: items_0::timebin::BinsBoxed| {
let ret = k.to_old_time_binned();
use items_0::timebin::BinningggContainerBinsDyn;
on_sitemty_data!(item, |x: Box<dyn BinningggContainerBinsDyn>| {
let g = x.new_collector();
let ret = Box::new(x) as Box<dyn CollectableDyn>;
// let ret = x as Box<dyn CollectableDyn>;
Ok(StreamItem::DataItem(RangeCompletableItem::Data(ret)))
})
});
let stream = stream.map(|item| {
on_sitemty_data!(item, |x| {
let ret = Box::new(x) as Box<dyn Collectable>;
Ok(StreamItem::DataItem(RangeCompletableItem::Data(ret)))
})
});
// let stream: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn TimeBinned>>> + Send>> = Box::pin(stream);
let stream = Box::pin(stream);
Ok(stream)
}
@@ -452,7 +454,7 @@ async fn timebinned_stream(
}
let stream = stream.map(|x| {
on_sitemty_data!(x, |x| {
let ret = Box::new(x) as Box<dyn Collectable>;
let ret = Box::new(x) as Box<dyn CollectableDyn>;
Ok(StreamItem::DataItem(RangeCompletableItem::Data(ret)))
})
});
@@ -508,14 +510,14 @@ pub async fn timebinned_json(
debug!("timebinned_json collected type_name {:?}", collres.type_name());
collres
};
let jsval = serde_json::to_value(&collres)?;
let jsval = collres.to_json_value()?;
Ok(CollectResult::Some(jsval))
}
CollectResult::Timeout => Ok(CollectResult::Timeout),
}
}
fn take_collector_result(coll: &mut Box<dyn items_0::collect_s::Collector>) -> Option<serde_json::Value> {
fn take_collector_result(coll: &mut Box<dyn items_0::collect_s::CollectorDyn>) -> Option<serde_json::Value> {
match coll.result(None, None) {
Ok(collres) => {
let collres = if let Some(bins) = collres
@@ -528,7 +530,7 @@ fn take_collector_result(coll: &mut Box<dyn items_0::collect_s::Collector>) -> O
} else {
collres
};
match serde_json::to_value(&collres) {
match collres.to_json_value() {
Ok(val) => Some(val),
Err(e) => Some(serde_json::Value::String(format!("{e}"))),
}

35
crates/streams/src/transform.rs
View File

@@ -1,7 +1,7 @@
use err::Error;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::collect_s::Collectable;
use items_0::collect_s::CollectableDyn;
use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
use items_0::streamitem::StreamItem;
@@ -109,7 +109,7 @@ impl TimeBinnableToCollectable {
}
impl Stream for TimeBinnableToCollectable {
type Item = Sitemty<Box<dyn Collectable>>;
type Item = Sitemty<Box<dyn CollectableDyn>>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
@@ -163,21 +163,24 @@ pub fn build_full_transform_collectable(
// TODO this must return a Stream!
//let evs = build_event_transform(tr, inp)?;
let trtb = tr.get_tr_time_binning();
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)),
let a: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn CollectableDyn>>> + 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 CollectableDyn> = 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)),
},
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>> =
Err(e) => Err(e),
}));
let stream: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn CollectableDyn>>> + Send>> =
Box::pin(futures_util::stream::empty());
let stream = Box::pin(futures_util::stream::empty()) as _;
match trtb {