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

WIP

Browse Source
This commit is contained in:
Dominik Werder
2023-04-18 11:51:48 +02:00
parent 6c9f084623
commit b3f53c60d8
8 changed files with 306 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
items_2/src/channelevents.rs
View File

@@ -532,7 +532,7 @@ impl ByteEstimate for ChannelEvents {
impl Mergeable for ChannelEvents {
fn ts_min(&self) -> Option<u64> {
match self {
ChannelEvents::Events(k) => k.ts_min(),
ChannelEvents::Events(k) => Mergeable::ts_min(k),
ChannelEvents::Status(k) => match k {
Some(k) => Some(k.ts),
None => None,
@@ -542,7 +542,7 @@ impl Mergeable for ChannelEvents {
fn ts_max(&self) -> Option<u64> {
match self {
ChannelEvents::Events(k) => k.ts_max(),
ChannelEvents::Events(k) => Mergeable::ts_max(k),
ChannelEvents::Status(k) => match k {
Some(k) => Some(k.ts),
None => None,

20
items_2/src/streams.rs
View File

@@ -3,7 +3,9 @@ use futures_util::FutureExt;
use futures_util::Stream;
use futures_util::StreamExt;
use items_0::collect_s::Collectable;
use items_0::streamitem::RangeCompletableItem;
use items_0::streamitem::Sitemty;
use items_0::streamitem::StreamItem;
use items_0::transform::CollectableStreamTrait;
use items_0::transform::EventStreamTrait;
use items_0::transform::EventTransform;
@@ -249,9 +251,23 @@ where
{
type Item = Sitemty<Box<dyn Events>>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
use Poll::*;
todo!()
match self.inp.poll_next_unpin(cx) {
Ready(Some(item)) => Ready(Some(match item {
Ok(item) => Ok(match item {
StreamItem::DataItem(item) => StreamItem::DataItem(match item {
RangeCompletableItem::RangeComplete => RangeCompletableItem::RangeComplete,
RangeCompletableItem::Data(item) => RangeCompletableItem::Data(Box::new(item)),
}),
StreamItem::Log(item) => StreamItem::Log(item),
StreamItem::Stats(item) => StreamItem::Stats(item),
}),
Err(e) => Err(e),
})),
Ready(None) => Ready(None),
Pending => Pending,
}
}
}