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

WIP

Browse Source
This commit is contained in:
Dominik Werder
2024-11-26 16:28:50 +01:00
parent 760e0abed4
commit 8432601e2b
16 changed files with 278 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files

187
src/cbor_stream.rs
View File

@@ -1,3 +1,4 @@
use crate::plaineventsstream::ChannelEventsStream;
use crate::streamtimeout::StreamTimeout2;
use crate::streamtimeout::TimeoutableStream;
use bytes::Buf;
@@ -6,23 +7,21 @@ use bytes::Bytes;
use bytes::BytesMut;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::collect_s::ToCborValue;
use items_0::streamitem::sitem_err2_from_string;
use items_0::streamitem::sitem_err_from_string;
use items_0::streamitem::LogItem;
use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
use items_0::streamitem::StreamItem;
use items_0::AsAnyMut;
use items_0::AsAnyRef;
use items_0::Events;
use items_0::WithLen;
use items_2::binning::container_events::ContainerEvents;
use items_2::channelevents::ChannelEvents;
use items_2::eventsdim0::EventsDim0;
use items_2::eventsdim1::EventsDim1;
use items_2::jsonbytes::CborBytes;
use netpod::log::Level;
use netpod::log::*;
use netpod::EnumVariant;
use netpod::ScalarType;
use netpod::Shape;
use std::io::Cursor;
@@ -40,6 +39,7 @@ pub enum Error {
FromSlice(#[from] std::array::TryFromSliceError),
Msg(String),
Ciborium(#[from] ciborium::de::Error<std::io::Error>),
CiboriumValue(#[from] ciborium::value::Error),
}
struct ErrMsg<E>(E)
@@ -55,132 +55,56 @@ where
}
}
pub struct CborBytes(Bytes);
impl CborBytes {
pub fn into_inner(self) -> Bytes {
self.0
}
pub fn len(&self) -> u32 {
self.0.len() as _
}
}
impl WithLen for CborBytes {
fn len(&self) -> usize {
self.len() as usize
}
}
impl From<CborBytes> for Bytes {
fn from(value: CborBytes) -> Self {
value.0
}
}
pub type CborStream = Pin<Box<dyn Stream<Item = Result<CborBytes, Error>> + Send>>;
// TODO move this type decl because it is not specific to cbor
pub type SitemtyDynEventsStream = Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Events>>> + Send>>;
pub fn events_stream_to_cbor_stream(
stream: SitemtyDynEventsStream,
stream: ChannelEventsStream,
ivl: Duration,
timeout_provider: Box<dyn StreamTimeout2>,
) -> impl Stream<Item = Result<CborBytes, Error>> {
let ivl = Duration::from_millis(4000);
let stream = TimeoutableStream::new(ivl, timeout_provider, stream);
let stream = stream.map(|x| match x {
Some(x) => map_events(x),
None => make_keepalive(),
});
let prepend = {
let item = make_keepalive();
futures_util::stream::iter([item])
};
prepend.chain(stream)
stream
}
fn map_events(x: Sitemty<Box<dyn Events>>) -> Result<CborBytes, Error> {
fn map_events(x: Sitemty<ChannelEvents>) -> Result<CborBytes, Error> {
match x {
Ok(x) => match x {
StreamItem::DataItem(x) => match x {
RangeCompletableItem::Data(evs) => {
if false {
// TODO impl generically on EventsDim0 ?
if let Some(evs) = evs.as_any_ref().downcast_ref::<ContainerEvents<f64>>() {
let mut buf = Vec::new();
ciborium::into_writer(evs, &mut buf)
.map_err(|e| Error::Msg(e.to_string()))?;
let bytes = Bytes::from(buf);
let _item = CborBytes(bytes);
// Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))
} else {
let _item = LogItem::from_node(
0,
Level::DEBUG,
format!("cbor stream discarded item"),
);
// Ok(StreamItem::Log(item))
};
}
let mut k = evs;
let evs = if let Some(j) = k.as_any_mut().downcast_mut::<ChannelEvents>() {
match j {
ChannelEvents::Events(m) => {
if let Some(g) = m
.as_any_mut()
.downcast_mut::<ContainerEvents<EnumVariant>>()
{
trace!("consider container EnumVariant");
k
} else {
trace!(
"consider container channel events other events {}",
k.type_name()
);
k
}
}
ChannelEvents::Status(_) => {
trace!(
"consider container channel events status {}",
k.type_name()
);
k
}
}
} else {
trace!("consider container else {}", k.type_name());
k
};
let buf = evs.to_cbor_vec_u8();
use items_0::apitypes::ToUserFacingApiType;
let val = evs.to_user_facing_api_type();
let val = val.to_cbor_value()?;
let mut buf = Vec::with_capacity(64);
ciborium::into_writer(&val, &mut buf).map_err(|e| Error::Msg(e.to_string()))?;
let bytes = Bytes::from(buf);
let item = CborBytes(bytes);
let item = CborBytes::new(bytes);
Ok(item)
}
RangeCompletableItem::RangeComplete => {
use ciborium::cbor;
let item = cbor!({
let val = cbor!({
"rangeFinal" => true,
})
.map_err(|e| Error::Msg(e.to_string()))?;
let mut buf = Vec::with_capacity(64);
ciborium::into_writer(&item, &mut buf)
.map_err(|e| Error::Msg(e.to_string()))?;
ciborium::into_writer(&val, &mut buf).map_err(|e| Error::Msg(e.to_string()))?;
let bytes = Bytes::from(buf);
let item = CborBytes(bytes);
let item = CborBytes::new(bytes);
Ok(item)
}
},
StreamItem::Log(item) => {
info!("{item:?}");
let item = CborBytes(Bytes::new());
let item = CborBytes::new(Bytes::new());
Ok(item)
}
StreamItem::Stats(item) => {
info!("{item:?}");
let item = CborBytes(Bytes::new());
let item = CborBytes::new(Bytes::new());
Ok(item)
}
},
@@ -193,7 +117,58 @@ fn map_events(x: Sitemty<Box<dyn Events>>) -> Result<CborBytes, Error> {
let mut buf = Vec::with_capacity(64);
ciborium::into_writer(&item, &mut buf).map_err(|e| Error::Msg(e.to_string()))?;
let bytes = Bytes::from(buf);
let item = CborBytes(bytes);
let item = CborBytes::new(bytes);
Ok(item)
}
}
}
fn map_events_2(x: Sitemty<ChannelEvents>) -> Result<CborBytes, Error> {
match x {
Ok(x) => match x {
StreamItem::DataItem(x) => match x {
RangeCompletableItem::Data(evs) => {
let val = evs.to_cbor_value()?;
let mut buf = Vec::with_capacity(64);
ciborium::into_writer(&val, &mut buf).map_err(|e| Error::Msg(e.to_string()))?;
let bytes = Bytes::from(buf);
let item = CborBytes::new(bytes);
Ok(item)
}
RangeCompletableItem::RangeComplete => {
use ciborium::cbor;
let val = cbor!({
"rangeFinal" => true,
})
.map_err(|e| Error::Msg(e.to_string()))?;
let mut buf = Vec::with_capacity(64);
ciborium::into_writer(&val, &mut buf).map_err(|e| Error::Msg(e.to_string()))?;
let bytes = Bytes::from(buf);
let item = CborBytes::new(bytes);
Ok(item)
}
},
StreamItem::Log(item) => {
info!("{item:?}");
let item = CborBytes::new(Bytes::new());
Ok(item)
}
StreamItem::Stats(item) => {
info!("{item:?}");
let item = CborBytes::new(Bytes::new());
Ok(item)
}
},
Err(e) => {
use ciborium::cbor;
let item = cbor!({
"error" => e.to_string(),
})
.map_err(|e| Error::Msg(e.to_string()))?;
let mut buf = Vec::with_capacity(64);
ciborium::into_writer(&item, &mut buf).map_err(|e| Error::Msg(e.to_string()))?;
let bytes = Bytes::from(buf);
let item = CborBytes::new(bytes);
Ok(item)
}
}
@@ -208,18 +183,18 @@ fn make_keepalive() -> Result<CborBytes, Error> {
let mut buf = Vec::with_capacity(64);
ciborium::into_writer(&item, &mut buf).map_err(ErrMsg)?;
let bytes = Bytes::from(buf);
let item = Ok(CborBytes(bytes));
let item = Ok(CborBytes::new(bytes));
item
}
pub struct FramedBytesToSitemtyDynEventsStream<S> {
pub struct FramedBytesToChannelEventsStream<S> {
inp: S,
scalar_type: ScalarType,
shape: Shape,
buf: BytesMut,
}
impl<S> FramedBytesToSitemtyDynEventsStream<S> {
impl<S> FramedBytesToChannelEventsStream<S> {
pub fn new(inp: S, scalar_type: ScalarType, shape: Shape) -> Self {
Self {
inp,
@@ -229,7 +204,7 @@ impl<S> FramedBytesToSitemtyDynEventsStream<S> {
}
}
fn try_parse(&mut self) -> Result<Option<Sitemty<Box<dyn Events>>>, Error> {
fn try_parse(&mut self) -> Result<Option<Sitemty<ChannelEvents>>, Error> {
// debug!("try_parse {}", self.buf.len());
if self.buf.len() < FRAME_HEAD_LEN {
return Ok(None);
@@ -252,7 +227,7 @@ impl<S> FramedBytesToSitemtyDynEventsStream<S> {
let buf = &self.buf[FRAME_HEAD_LEN..frame_len];
let val: ciborium::Value =
ciborium::from_reader(std::io::Cursor::new(buf)).map_err(ErrMsg)?;
// debug!("decoded ciborium value {val:?}");
debug!("decoded ciborium value {val:?}");
let item = if let Some(map) = val.as_map() {
let keys: Vec<&str> = map
.iter()
@@ -285,6 +260,7 @@ impl<S> FramedBytesToSitemtyDynEventsStream<S> {
} else {
None
};
let item = None;
let item = if let Some(x) = item {
Some(x)
} else {
@@ -302,12 +278,12 @@ impl<S> FramedBytesToSitemtyDynEventsStream<S> {
}
}
impl<S, E> Stream for FramedBytesToSitemtyDynEventsStream<S>
impl<S, E> Stream for FramedBytesToChannelEventsStream<S>
where
S: Stream<Item = Result<Bytes, E>> + Unpin,
E: std::error::Error,
{
type Item = <SitemtyDynEventsStream as Stream>::Item;
type Item = Sitemty<ChannelEvents>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
@@ -361,7 +337,7 @@ fn decode_cbor_to_box_events(
buf: &[u8],
scalar_type: &ScalarType,
shape: &Shape,
) -> Result<Box<dyn Events>, Error> {
) -> Result<ChannelEvents, Error> {
let item: Box<dyn Events> = match shape {
Shape::Scalar => match scalar_type {
ScalarType::U8 => cbor_scalar!(u8, buf),
@@ -396,5 +372,6 @@ fn decode_cbor_to_box_events(
},
Shape::Image(_, _) => todo!(),
};
Ok(item)
// Ok(item);
todo!()
}

20
src/collect.rs
View File

@@ -45,8 +45,9 @@ pub enum CollectResult<T> {
Some(T),
}
pub struct Collect {
inp: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn CollectableDyn>>> + Send>>,
pub struct Collect<ITEM> {
// inp: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn CollectableDyn>>> + Send>>,
inp: Pin<Box<dyn Stream<Item = Sitemty<ITEM>> + Send>>,
events_max: u64,
bytes_max: u64,
range: Option<SeriesRange>,
@@ -58,9 +59,13 @@ pub struct Collect {
done_input: bool,
}
impl Collect {
impl<ITEM> Collect<ITEM>
where
ITEM: CollectableDyn,
{
pub fn new(
inp: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn CollectableDyn>>> + Send>>,
// inp: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn CollectableDyn>>> + Send>>,
inp: Pin<Box<dyn Stream<Item = Sitemty<ITEM>> + Send>>,
deadline: Instant,
events_max: u64,
bytes_max: u64,
@@ -83,7 +88,7 @@ impl Collect {
}
}
fn handle_item(&mut self, item: Sitemty<Box<dyn CollectableDyn>>) -> Result<(), Error> {
fn handle_item(&mut self, item: Sitemty<ITEM>) -> Result<(), Error> {
match item {
Ok(item) => match item {
StreamItem::DataItem(item) => match item {
@@ -164,7 +169,10 @@ impl Collect {
}
}
impl Future for Collect {
impl<ITEM> Future for Collect<ITEM>
where
ITEM: CollectableDyn,
{
type Output = Result<CollectResult<Box<dyn CollectedDyn>>, Error>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {

2
src/firsterr.rs
View File

@@ -1,8 +1,8 @@
use crate::cbor_stream::CborBytes;
use futures_util::future;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::WithLen;
use items_2::jsonbytes::CborBytes;
pub fn non_empty<S, T, E>(inp: S) -> impl Stream<Item = Result<T, E>>
where

24
src/frames.rs
View File

@@ -1,2 +1,26 @@
pub mod eventsfromframes;
pub mod inmem;
use bytes::Bytes;
use bytes::BytesMut;
use futures_util::Stream;
use futures_util::StreamExt;
use items_2::framable::Framable;
#[derive(Debug, thiserror::Error)]
#[cstm(name = "FramedStreamError")]
pub enum Error {
MakeFrame(#[from] items_2::framable::Error),
}
pub fn frameable_stream_to_bytes_stream<S, T>(stream: S) -> impl Stream<Item = Result<Bytes, Error>>
where
S: Stream<Item = T>,
T: Framable,
{
stream.map(|x| {
x.make_frame_dyn()
.map(BytesMut::freeze)
.map_err(|e| e.into())
})
}

1
src/frames/inmem.rs
View File

@@ -14,7 +14,6 @@ use netpod::ByteSize;
use std::pin::Pin;
use std::task::Context;
use std::task::Poll;
// use tokio::io::AsyncRead;
#[derive(Debug, thiserror::Error)]
#[cstm(name = "InMem")]

84
src/json_stream.rs
View File

@@ -1,8 +1,9 @@
use crate::cbor_stream::SitemtyDynEventsStream;
use crate::plaineventsstream::ChannelEventsStream;
use crate::streamtimeout::StreamTimeout2;
use crate::streamtimeout::TimeoutableStream;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::collect_s::ToJsonValue;
use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
use items_0::streamitem::StreamItem;
@@ -10,6 +11,7 @@ use items_0::Events;
use items_0::WithLen;
use items_2::binning::container_events::ContainerEvents;
use items_2::channelevents::ChannelEvents;
use items_2::jsonbytes::JsonBytes;
use netpod::log::*;
use netpod::EnumVariant;
use std::pin::Pin;
@@ -35,93 +37,31 @@ where
}
}
pub struct JsonBytes(String);
impl JsonBytes {
pub fn new<S: Into<String>>(s: S) -> Self {
Self(s.into())
}
pub fn into_inner(self) -> String {
self.0
}
pub fn len(&self) -> u32 {
self.0.len() as _
}
}
impl WithLen for JsonBytes {
fn len(&self) -> usize {
self.len() as usize
}
}
impl From<JsonBytes> for String {
fn from(value: JsonBytes) -> Self {
value.0
}
}
pub type JsonStream = Pin<Box<dyn Stream<Item = Result<JsonBytes, Error>> + Send>>;
pub fn events_stream_to_json_stream(
stream: SitemtyDynEventsStream,
stream: ChannelEventsStream,
ivl: Duration,
timeout_provider: Box<dyn StreamTimeout2>,
) -> impl Stream<Item = Result<JsonBytes, Error>> {
let ivl = Duration::from_millis(4000);
let stream = TimeoutableStream::new(ivl, timeout_provider, stream);
let stream = stream.map(|x| match x {
Some(x) => map_events(x),
None => make_keepalive(),
});
let prepend = {
let item = make_keepalive();
futures_util::stream::iter([item])
};
prepend.chain(stream)
stream
}
fn map_events(x: Sitemty<Box<dyn Events>>) -> Result<JsonBytes, Error> {
fn map_events<T>(x: Sitemty<T>) -> Result<JsonBytes, Error>
where
T: ToJsonValue,
{
match x {
Ok(x) => match x {
StreamItem::DataItem(x) => match x {
RangeCompletableItem::Data(evs) => {
let mut k = evs;
let evs = if let Some(j) = k.as_any_mut().downcast_mut::<ChannelEvents>() {
match j {
ChannelEvents::Events(m) => {
if let Some(g) = m
.as_any_mut()
.downcast_mut::<ContainerEvents<EnumVariant>>()
{
trace!("consider container EnumVariant");
let mut out = ContainerEvents::new();
for (&ts, val) in g.iter_zip() {
out.push_back(ts, val.name.to_string());
}
Box::new(ChannelEvents::Events(Box::new(out)))
} else {
trace!(
"consider container channel events other events {}",
k.type_name()
);
k
}
}
ChannelEvents::Status(_) => {
trace!(
"consider container channel events status {}",
k.type_name()
);
k
}
}
} else {
trace!("consider container else {}", k.type_name());
k
};
let s = evs.to_json_string();
let val = evs.to_json_value()?;
let s = serde_json::to_string(&val)?;
let item = JsonBytes::new(s);
Ok(item)
}

72
src/lenframe.rs Normal file
View File

@@ -0,0 +1,72 @@
use crate::log::*;
use bytes::BufMut;
use bytes::Bytes;
use bytes::BytesMut;
use futures_util::future::ready;
use futures_util::stream;
use futures_util::Stream;
use futures_util::StreamExt;
pub fn bytes_chunks_to_framed<S, T, E>(stream: S) -> impl Stream<Item = Result<Bytes, E>>
where
S: Stream<Item = Result<T, E>>,
T: Into<Bytes>,
E: std::error::Error,
{
stream
// TODO unify this map to padded bytes for both json and cbor output
.flat_map(|x| match x {
Ok(y) => {
let buf = y.into();
let adv = (buf.len() + 7) / 8 * 8;
let pad = adv - buf.len();
let mut b2 = BytesMut::with_capacity(16);
b2.put_u32_le(buf.len() as u32);
b2.put_slice(&[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
let mut b3 = BytesMut::with_capacity(16);
b3.put_slice(&[0, 0, 0, 0, 0, 0, 0, 0][..pad]);
stream::iter([Ok(b2.freeze()), Ok(buf), Ok(b3.freeze())])
}
Err(e) => {
error!("{e}");
stream::iter([Ok(Bytes::new()), Ok(Bytes::new()), Ok(Bytes::new())])
}
})
.filter(|x| {
if let Ok(x) = x {
ready(x.len() > 0)
} else {
ready(true)
}
})
}
// TODO move this, it's also used by binned.
pub fn bytes_chunks_to_len_framed_str<S, T, E>(stream: S) -> impl Stream<Item = Result<String, E>>
where
S: Stream<Item = Result<T, E>>,
T: Into<String>,
E: std::error::Error,
{
stream
.flat_map(|x| match x {
Ok(y) => {
use std::fmt::Write;
let s = y.into();
let mut b2 = String::with_capacity(16);
write!(b2, "{:15}\n", s.len()).unwrap();
stream::iter([Ok::<_, E>(b2), Ok(s), Ok(String::from("\n"))])
}
Err(e) => {
error!("{e}");
stream::iter([Ok(String::new()), Ok(String::new()), Ok(String::new())])
}
})
.filter(|x| {
if let Ok(x) = x {
ready(x.len() > 0)
} else {
ready(true)
}
})
}

1
src/lib.rs
View File

@@ -14,6 +14,7 @@ pub mod generators;
pub mod instrument;
pub mod itemclone;
pub mod json_stream;
pub mod lenframe;
pub mod lenframed;
pub mod needminbuffer;
pub mod plaineventscbor;

3
src/plaineventscbor.rs
View File

@@ -23,7 +23,8 @@ pub async fn plain_events_cbor_stream(
timeout_provider: Box<dyn StreamTimeout2>,
) -> Result<CborStream, Error> {
let stream = dyn_events_stream(evq, ch_conf, ctx, open_bytes).await?;
let stream = events_stream_to_cbor_stream(stream, timeout_provider);
let stream =
events_stream_to_cbor_stream(stream, evq.timeout_content_or_default(), timeout_provider);
let stream = non_empty(stream);
let stream = only_first_err(stream);
Ok(Box::pin(stream))

56
src/plaineventsjson.rs
View File

@@ -6,6 +6,7 @@ use crate::json_stream::events_stream_to_json_stream;
use crate::json_stream::JsonStream;
use crate::plaineventsstream::dyn_events_stream;
use crate::streamtimeout::StreamTimeout2;
use crate::streamtimeout::TimeoutableStream;
use crate::tcprawclient::OpenBoxedBytesStreamsBox;
use futures_util::StreamExt;
use items_0::collect_s::CollectableDyn;
@@ -37,61 +38,13 @@ pub async fn plain_events_json(
timeout_provider: Box<dyn StreamTimeout2>,
) -> Result<CollectResult<JsonValue>, Error> {
debug!("plain_events_json evquery {:?}", evq);
let deadline = Instant::now() + evq.timeout().unwrap_or(Duration::from_millis(4000));
let deadline = Instant::now() + evq.timeout_content_or_default();
let stream = dyn_events_stream(evq, ch_conf, ctx, open_bytes).await?;
let stream = stream.map(move |k| {
on_sitemty_data!(k, |mut k: Box<dyn items_0::Events>| {
if let Some(j) = k
.as_any_mut()
.downcast_mut::<items_2::channelevents::ChannelEvents>()
{
use items_0::AsAnyMut;
match j {
items_2::channelevents::ChannelEvents::Events(m) => {
if let Some(g) = m
.as_any_mut()
.downcast_mut::<items_2::eventsdim0::EventsDim0<netpod::EnumVariant>>()
{
trace!("consider container EnumVariant");
let mut out = items_2::eventsdim0enum::EventsDim0Enum::new();
for (&ts, val) in g.tss.iter().zip(g.values.iter()) {
out.push_back(ts, val.ix(), val.name_string());
}
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 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 CollectableDyn> = Box::new(k);
Ok(StreamItem::DataItem(RangeCompletableItem::Data(k)))
}
}
} else {
trace!("consider container else {}", k.type_name());
let k: Box<dyn CollectableDyn> = Box::new(k);
Ok(StreamItem::DataItem(RangeCompletableItem::Data(k)))
}
})
});
//let stream = PlainEventStream::new(stream);
//let stream = EventsToTimeBinnable::new(stream);
//let stream = TimeBinnableToCollectable::new(stream);
let stream = Box::pin(stream);
debug!("plain_events_json boxed stream created");
// let stream = Box::pin(stream);
let collected = Collect::new(
stream,
deadline,
@@ -122,7 +75,8 @@ pub async fn plain_events_json_stream(
) -> Result<JsonStream, Error> {
trace!("plain_events_json_stream");
let stream = dyn_events_stream(evq, ch_conf, ctx, open_bytes).await?;
let stream = events_stream_to_json_stream(stream, timeout_provider);
let stream =
events_stream_to_json_stream(stream, evq.timeout_content_or_default(), timeout_provider);
let stream = non_empty(stream);
let stream = only_first_err(stream);
Ok(Box::pin(stream))

20
src/plaineventsstream.rs
View File

@@ -22,14 +22,14 @@ pub enum Error {
TcpRawClient(#[from] crate::tcprawclient::Error),
}
pub type DynEventsStream = Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Events>>> + Send>>;
pub type ChannelEventsStream = Pin<Box<dyn Stream<Item = Sitemty<ChannelEvents>> + Send>>;
pub async fn dyn_events_stream(
evq: &PlainEventsQuery,
ch_conf: ChannelTypeConfigGen,
ctx: &ReqCtx,
open_bytes: OpenBoxedBytesStreamsBox,
) -> Result<DynEventsStream, Error> {
) -> Result<ChannelEventsStream, Error> {
trace!("dyn_events_stream {}", evq.summary_short());
use query::api4::events::EventsSubQuerySettings;
let subq = make_sub_query(
@@ -62,12 +62,6 @@ pub async fn dyn_events_stream(
evq.range().try_into()?,
evq.one_before_range(),
);
let stream = stream.map(move |k| {
on_sitemty_data!(k, |k| {
let k: Box<dyn Events> = Box::new(k);
Ok(StreamItem::DataItem(RangeCompletableItem::Data(k)))
})
});
if let Some(wasmname) = evq.test_do_wasm() {
let stream =
transform_wasm::<_, items_0::streamitem::SitemErrTy>(stream, wasmname, ctx).await?;
@@ -82,11 +76,11 @@ async fn transform_wasm<INP, ETS>(
stream: INP,
_wasmname: &str,
_ctx: &ReqCtx,
) -> Result<impl Stream<Item = Sitemty<Box<dyn Events>>> + Send, Error>
) -> Result<impl Stream<Item = Sitemty<ChannelEvents>> + Send, Error>
where
INP: Stream<Item = Sitemty<Box<dyn Events>>> + Send + 'static,
INP: Stream<Item = Sitemty<ChannelEvents>> + Send + 'static,
{
let ret: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Events>>> + Send>> = Box::pin(stream);
let ret = Box::pin(stream);
Ok(ret)
}
@@ -95,9 +89,9 @@ async fn transform_wasm<INP>(
stream: INP,
wasmname: &str,
ctx: &ReqCtx,
) -> Result<impl Stream<Item = Sitemty<Box<dyn Events>>> + Send, Error>
) -> Result<impl Stream<Item = Sitemty<ChannelEvents>> + Send, Error>
where
INP: Stream<Item = Sitemty<Box<dyn Events>>> + Send + 'static,
INP: Stream<Item = Sitemty<ChannelEvents>> + Send + 'static,
{
debug!("make wasm transform");
use httpclient::url::Url;

5
src/tcprawclient.rs
View File

@@ -186,6 +186,11 @@ where
{
let frames = InMemoryFrameStream::new(inp, bufcap);
let frames = frames.map_err(sitem_err2_from_string);
let frames = frames.inspect(|x| {
if false {
eprintln!("container_stream_from_bytes_stream see frame {:?}", x);
}
});
// TODO let EventsFromFrames accept also non-boxed input?
let frames = Box::pin(frames);
let stream = EventsFromFrames::<T>::new(frames, dbgdesc);

1
src/test.rs
View File

@@ -1,6 +1,7 @@
mod collect;
mod events;
mod events_reader;
mod framing;
mod timebin;
use futures_util::stream;

4
src/test/events.rs
View File

@@ -1,4 +1,4 @@
use crate::cbor_stream::FramedBytesToSitemtyDynEventsStream;
use crate::cbor_stream::FramedBytesToChannelEventsStream;
use crate::firsterr::only_first_err;
use crate::frames::inmem::BoxedBytesStream;
use crate::lenframed;
@@ -66,7 +66,7 @@ async fn merged_events_inner() -> Result<(), Error> {
.await
.unwrap();
let stream = lenframed::length_framed(stream);
let stream = FramedBytesToSitemtyDynEventsStream::new(
let stream = FramedBytesToChannelEventsStream::new(
stream,
ch_conf.scalar_type().clone(),
ch_conf.shape().clone(),

49
src/test/framing.rs Normal file
View File

@@ -0,0 +1,49 @@
use crate::frames::frameable_stream_to_bytes_stream;
use crate::tcprawclient::container_stream_from_bytes_stream;
use futures_util::TryStreamExt;
use items_0::streamitem::sitem_err2_from_string;
use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
use items_0::streamitem::StreamItem;
use items_2::binning::container_events::ContainerEvents;
use items_2::channelevents::ChannelEvents;
use netpod::ByteSize;
use netpod::TsNano;
async fn framing_00_inner() -> Result<(), Box<dyn std::error::Error>> {
let mut evs = ContainerEvents::<f32>::new();
evs.push_back(TsNano::from_ns(1), 1.2);
let cevs = ChannelEvents::from(evs);
let item: Sitemty<_> = Ok(StreamItem::DataItem(RangeCompletableItem::Data(cevs)));
let stream = futures_util::stream::iter([item]);
let stream = frameable_stream_to_bytes_stream(stream);
let stream = stream.map_err(sitem_err2_from_string);
let stream = stream.inspect_ok(|x| {
if false {
let a = &x[0..x.len().min(40)];
eprintln!("byte blob in stream {} {:?}", x.len(), a);
}
});
let stream = Box::pin(stream);
let bufcap = ByteSize(1024 * 1024);
let mut stream =
container_stream_from_bytes_stream::<ChannelEvents>(stream, bufcap, "test".into())?;
let mut n = 0;
while let Some(x) = stream.try_next().await? {
if false {
eprintln!("{x:?}");
}
n += 1;
}
assert_eq!(n, 1);
Ok(())
}
#[test]
fn framing_00() {
tokio::runtime::Builder::new_current_thread()
.build()
.unwrap()
.block_on(framing_00_inner())
.unwrap()
}

2
src/timebinnedjson.rs
View File

@@ -1,6 +1,5 @@
use crate::collect::Collect;
use crate::collect::CollectResult;
use crate::json_stream::JsonBytes;
use crate::json_stream::JsonStream;
use crate::rangefilter2::RangeFilter2;
use crate::streamtimeout::StreamTimeout2;
@@ -20,6 +19,7 @@ use items_0::streamitem::Sitemty;
use items_0::streamitem::StreamItem;
use items_0::Events;
use items_2::channelevents::ChannelEvents;
use items_2::jsonbytes::JsonBytes;
use items_2::merger::Merger;
use netpod::log::*;
use netpod::range::evrange::NanoRange;