WIP

disk/src/agg/binnedt.rs

@@ -1,5 +1,6 @@
 use crate::agg::streams::StreamItem;
 use crate::agg::AggregatableXdim1Bin;
+use crate::binned::RangeCompletableItem;
 use err::Error;
 use futures_core::Stream;
 use futures_util::StreamExt;
@@ -32,7 +33,7 @@ pub trait IntoBinnedT {
 
 impl<S, I> IntoBinnedT for S
 where
-    S: Stream<Item = Result<StreamItem<I>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
     I: AggregatableTdim + Unpin,
     I::Aggregator: Unpin,
 {
@@ -45,7 +46,7 @@ where
 
 pub struct IntoBinnedTDefaultStream<S, I>
 where
-    S: Stream<Item = Result<StreamItem<I>, Error>>,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
     I: AggregatableTdim,
 {
     inp: S,
@@ -56,7 +57,7 @@ where
     all_bins_emitted: bool,
     range_complete_observed: bool,
     range_complete_emitted: bool,
-    left: Option<Poll<Option<Result<StreamItem<I>, Error>>>>,
+    left: Option<Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>>>,
     errored: bool,
     completed: bool,
     tmp_agg_results: VecDeque<<I::Aggregator as AggregatorTdim>::OutputValue>,
@@ -64,7 +65,7 @@ where
 
 impl<S, I> IntoBinnedTDefaultStream<S, I>
 where
-    S: Stream<Item = Result<StreamItem<I>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
     I: AggregatableTdim,
 {
     pub fn new(inp: S, spec: BinnedRange) -> Self {
@@ -85,7 +86,7 @@ where
         }
     }
 
-    fn cur(&mut self, cx: &mut Context) -> Poll<Option<Result<StreamItem<I>, Error>>> {
+    fn cur(&mut self, cx: &mut Context) -> Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>> {
         if let Some(cur) = self.left.take() {
             cur
         } else if self.inp_completed {
@@ -113,43 +114,50 @@ where
 
     fn handle(
         &mut self,
-        cur: Poll<Option<Result<StreamItem<I>, Error>>>,
+        cur: Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>>,
-    ) -> Option<Poll<Option<Result<StreamItem<<I::Aggregator as AggregatorTdim>::OutputValue>, Error>>>> {
+    ) -> Option<
+        Poll<Option<Result<StreamItem<RangeCompletableItem<<I::Aggregator as AggregatorTdim>::OutputValue>>, Error>>>,
+    > {
         use Poll::*;
         match cur {
             Ready(Some(Ok(item))) => match item {
                 StreamItem::Log(item) => Some(Ready(Some(Ok(StreamItem::Log(item))))),
                 StreamItem::Stats(item) => Some(Ready(Some(Ok(StreamItem::Stats(item))))),
-                StreamItem::DataItem(item) => {
+                StreamItem::DataItem(item) => match item {
-                    if item.is_range_complete() {
+                    RangeCompletableItem::RangeComplete => {
                         self.range_complete_observed = true;
                         None
-                    } else if self.all_bins_emitted {
+                    }
-                        // Just drop the item because we will not emit anymore data.
+                    RangeCompletableItem::Data(item) => {
-                        // Could also at least gather some stats.
+                        if self.all_bins_emitted {
-                        None
+                            // Just drop the item because we will not emit anymore data.
-                    } else {
+                            // Could also at least gather some stats.
-                        let ag = self.aggtor.as_mut().unwrap();
-                        if ag.ends_before(&item) {
-                            None
-                        } else if ag.starts_after(&item) {
-                            self.left = Some(Ready(Some(Ok(StreamItem::DataItem(item)))));
-                            self.cycle_current_bin();
-                            // TODO cycle_current_bin enqueues the bin, can I return here instead?
                             None
                         } else {
-                            let mut item = item;
+                            let ag = self.aggtor.as_mut().unwrap();
-                            ag.ingest(&mut item);
+                            if ag.ends_before(&item) {
-                            let item = item;
+                                None
-                            if ag.ends_after(&item) {
+                            } else if ag.starts_after(&item) {
-                                self.left = Some(Ready(Some(Ok(StreamItem::DataItem(item)))));
+                                self.left =
+                                    Some(Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item))))));
                                 self.cycle_current_bin();
+                                // TODO cycle_current_bin enqueues the bin, can I return here instead?
+                                None
+                            } else {
+                                let mut item = item;
+                                ag.ingest(&mut item);
+                                let item = item;
+                                if ag.ends_after(&item) {
+                                    self.left =
+                                        Some(Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item))))));
+                                    self.cycle_current_bin();
+                                }
+                                // TODO cycle_current_bin enqueues the bin, can I return here instead?
+                                None
                             }
-                            // TODO cycle_current_bin enqueues the bin, can I return here instead?
-                            None
                         }
                     }
-                }
+                },
             },
             Ready(Some(Err(e))) => {
                 self.errored = true;
@@ -172,11 +180,11 @@ where
 
 impl<S, I> Stream for IntoBinnedTDefaultStream<S, I>
 where
-    S: Stream<Item = Result<StreamItem<I>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
     I: AggregatableTdim + Unpin,
     I::Aggregator: Unpin,
 {
-    type Item = Result<StreamItem<<I::Aggregator as AggregatorTdim>::OutputValue>, Error>;
+    type Item = Result<StreamItem<RangeCompletableItem<<I::Aggregator as AggregatorTdim>::OutputValue>>, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;
@@ -187,19 +195,14 @@ where
                 self.completed = true;
                 Ready(None)
             } else if let Some(item) = self.tmp_agg_results.pop_front() {
-                Ready(Some(Ok(StreamItem::DataItem(item))))
+                Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))))
             } else if self.range_complete_emitted {
                 self.completed = true;
                 Ready(None)
             } else if self.inp_completed && self.all_bins_emitted {
                 self.range_complete_emitted = true;
                 if self.range_complete_observed {
-                    if let Some(item) = <I::Aggregator as AggregatorTdim>::OutputValue::make_range_complete_item() {
+                    Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete))))
-                        Ready(Some(Ok(StreamItem::DataItem(item))))
-                    } else {
-                        warn!("IntoBinnedTDefaultStream  should emit RangeComplete but it doesn't have one");
-                        continue 'outer;
-                    }
                 } else {
                     continue 'outer;
                 }

disk/src/agg/binnedt2.rs

@@ -1,5 +1,7 @@
+use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
 use crate::agg::streams::StreamItem;
 use crate::agg::AggregatableXdim1Bin;
+use crate::binned::RangeCompletableItem;
 use err::Error;
 use futures_core::Stream;
 use futures_util::StreamExt;
@@ -9,7 +11,7 @@ use std::collections::VecDeque;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-pub trait AggregatorTdim2: Sized + Unpin {
+pub trait AggregatorTdim2: Sized + Send + Unpin {
     type InputValue;
     fn ends_before(&self, inp: &Self::InputValue) -> bool;
     fn ends_after(&self, inp: &Self::InputValue) -> bool;
@@ -21,8 +23,6 @@ pub trait AggregatorTdim2: Sized + Unpin {
 pub trait AggregatableTdim2: Sized {
     type Aggregator: AggregatorTdim2<InputValue = Self>;
     fn aggregator_new_static(ts1: u64, ts2: u64) -> Self::Aggregator;
-    fn is_range_complete(&self) -> bool;
-    fn make_range_complete_item() -> Option<Self>;
 }
 
 pub trait IntoBinnedT {
@@ -32,7 +32,7 @@ pub trait IntoBinnedT {
 
 impl<S, I> IntoBinnedT for S
 where
-    S: Stream<Item = Result<StreamItem<I>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
     I: AggregatableTdim2 + Unpin,
     I::Aggregator: Unpin,
 {
@@ -45,7 +45,7 @@ where
 
 pub struct IntoBinnedTDefaultStream<S, I>
 where
-    S: Stream<Item = Result<StreamItem<I>, Error>>,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>>,
     I: AggregatableTdim2,
 {
     inp: S,
@@ -56,7 +56,7 @@ where
     all_bins_emitted: bool,
     range_complete_observed: bool,
     range_complete_emitted: bool,
-    left: Option<Poll<Option<Result<StreamItem<I>, Error>>>>,
+    left: Option<Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>>>,
     errored: bool,
     completed: bool,
     tmp_agg_results: VecDeque<I>,
@@ -64,7 +64,7 @@ where
 
 impl<S, I> IntoBinnedTDefaultStream<S, I>
 where
-    S: Stream<Item = Result<StreamItem<I>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
     I: AggregatableTdim2,
 {
     pub fn new(inp: S, spec: BinnedRange) -> Self {
@@ -85,7 +85,7 @@ where
         }
     }
 
-    fn cur(&mut self, cx: &mut Context) -> Poll<Option<Result<StreamItem<I>, Error>>> {
+    fn cur(&mut self, cx: &mut Context) -> Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>> {
         if let Some(cur) = self.left.take() {
             cur
         } else if self.inp_completed {
@@ -113,43 +113,47 @@ where
 
     fn handle(
         &mut self,
-        cur: Poll<Option<Result<StreamItem<I>, Error>>>,
+        cur: Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>>,
-    ) -> Option<Poll<Option<Result<StreamItem<I>, Error>>>> {
+    ) -> Option<Poll<Option<Result<StreamItem<RangeCompletableItem<I>>, Error>>>> {
         use Poll::*;
         match cur {
             Ready(Some(Ok(item))) => match item {
                 StreamItem::Log(item) => Some(Ready(Some(Ok(StreamItem::Log(item))))),
                 StreamItem::Stats(item) => Some(Ready(Some(Ok(StreamItem::Stats(item))))),
-                StreamItem::DataItem(item) => {
+                StreamItem::DataItem(item) => match item {
-                    if item.is_range_complete() {
+                    RangeCompletableItem::RangeComplete => Some(Ready(Some(Ok(StreamItem::DataItem(
-                        self.range_complete_observed = true;
+                        RangeCompletableItem::RangeComplete,
-                        None
+                    ))))),
-                    } else if self.all_bins_emitted {
+                    RangeCompletableItem::Data(item) => {
-                        // Just drop the item because we will not emit anymore data.
+                        if self.all_bins_emitted {
-                        // Could also at least gather some stats.
+                            // Just drop the item because we will not emit anymore data.
-                        None
+                            // Could also at least gather some stats.
-                    } else {
-                        let ag = self.aggtor.as_mut().unwrap();
-                        if ag.ends_before(&item) {
-                            None
-                        } else if ag.starts_after(&item) {
-                            self.left = Some(Ready(Some(Ok(StreamItem::DataItem(item)))));
-                            self.cycle_current_bin();
-                            // TODO cycle_current_bin enqueues the bin, can I return here instead?
                             None
                         } else {
-                            let mut item = item;
+                            let ag = self.aggtor.as_mut().unwrap();
-                            ag.ingest(&mut item);
+                            if ag.ends_before(&item) {
-                            let item = item;
+                                None
-                            if ag.ends_after(&item) {
+                            } else if ag.starts_after(&item) {
-                                self.left = Some(Ready(Some(Ok(StreamItem::DataItem(item)))));
+                                self.left =
+                                    Some(Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item))))));
                                 self.cycle_current_bin();
+                                // TODO cycle_current_bin enqueues the bin, can I return here instead?
+                                None
+                            } else {
+                                let mut item = item;
+                                ag.ingest(&mut item);
+                                let item = item;
+                                if ag.ends_after(&item) {
+                                    self.left =
+                                        Some(Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item))))));
+                                    self.cycle_current_bin();
+                                }
+                                // TODO cycle_current_bin enqueues the bin, can I return here instead?
+                                None
                             }
-                            // TODO cycle_current_bin enqueues the bin, can I return here instead?
-                            None
                         }
                     }
-                }
+                },
             },
             Ready(Some(Err(e))) => {
                 self.errored = true;
@@ -172,11 +176,11 @@ where
 
 impl<S, I> Stream for IntoBinnedTDefaultStream<S, I>
 where
-    S: Stream<Item = Result<StreamItem<I>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
     I: AggregatableTdim2 + Unpin,
     I::Aggregator: Unpin,
 {
-    type Item = Result<StreamItem<I>, Error>;
+    type Item = Result<StreamItem<RangeCompletableItem<I>>, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;
@@ -187,19 +191,14 @@ where
                 self.completed = true;
                 Ready(None)
             } else if let Some(item) = self.tmp_agg_results.pop_front() {
-                Ready(Some(Ok(StreamItem::DataItem(item))))
+                Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))))
             } else if self.range_complete_emitted {
                 self.completed = true;
                 Ready(None)
             } else if self.inp_completed && self.all_bins_emitted {
                 self.range_complete_emitted = true;
                 if self.range_complete_observed {
-                    if let Some(item) = I::make_range_complete_item() {
+                    Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete))))
-                        Ready(Some(Ok(StreamItem::DataItem(item))))
-                    } else {
-                        warn!("IntoBinnedTDefaultStream  should emit RangeComplete but it doesn't have one");
-                        continue 'outer;
-                    }
                 } else {
                     continue 'outer;
                 }
@@ -213,3 +212,37 @@ where
         }
     }
 }
+
+pub struct MinMaxAvgScalarBinBatchAgg {}
+
+impl AggregatorTdim2 for MinMaxAvgScalarBinBatchAgg {
+    type InputValue = MinMaxAvgScalarBinBatch;
+
+    fn ends_before(&self, inp: &Self::InputValue) -> bool {
+        todo!()
+    }
+
+    fn ends_after(&self, inp: &Self::InputValue) -> bool {
+        todo!()
+    }
+
+    fn starts_after(&self, inp: &Self::InputValue) -> bool {
+        todo!()
+    }
+
+    fn ingest(&mut self, inp: &mut Self::InputValue) {
+        todo!()
+    }
+
+    fn result(self) -> Vec<Self::InputValue> {
+        todo!()
+    }
+}
+
+impl AggregatableTdim2 for MinMaxAvgScalarBinBatch {
+    type Aggregator = MinMaxAvgScalarBinBatchAgg;
+
+    fn aggregator_new_static(ts1: u64, ts2: u64) -> Self::Aggregator {
+        todo!()
+    }
+}

disk/src/agg/binnedx.rs

@@ -1,5 +1,6 @@
 use crate::agg::streams::StreamItem;
 use crate::agg::AggregatableXdim1Bin;
+use crate::binned::RangeCompletableItem;
 use err::Error;
 use futures_core::Stream;
 use futures_util::StreamExt;
@@ -8,20 +9,19 @@ use std::task::{Context, Poll};
 
 pub trait IntoBinnedXBins1<I>
 where
+    Self: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
     I: AggregatableXdim1Bin,
 {
     type StreamOut;
-    fn into_binned_x_bins_1(self) -> Self::StreamOut
+    fn into_binned_x_bins_1(self) -> Self::StreamOut;
-    where
-        Self: Stream<Item = Result<StreamItem<I>, Error>>;
 }
 
-impl<T, I> IntoBinnedXBins1<I> for T
+impl<S, I> IntoBinnedXBins1<I> for S
 where
-    T: Stream<Item = Result<StreamItem<I>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
     I: AggregatableXdim1Bin,
 {
-    type StreamOut = IntoBinnedXBins1DefaultStream<T, I>;
+    type StreamOut = IntoBinnedXBins1DefaultStream<S, I>;
 
     fn into_binned_x_bins_1(self) -> Self::StreamOut {
         IntoBinnedXBins1DefaultStream { inp: self }
@@ -30,7 +30,7 @@ where
 
 pub struct IntoBinnedXBins1DefaultStream<S, I>
 where
-    S: Stream<Item = Result<StreamItem<I>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
     I: AggregatableXdim1Bin,
 {
     inp: S,
@@ -38,10 +38,10 @@ where
 
 impl<S, I> Stream for IntoBinnedXBins1DefaultStream<S, I>
 where
-    S: Stream<Item = Result<StreamItem<I>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<I>>, Error>> + Unpin,
     I: AggregatableXdim1Bin,
 {
-    type Item = Result<StreamItem<I::Output>, Error>;
+    type Item = Result<StreamItem<RangeCompletableItem<I::Output>>, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;
@@ -49,7 +49,14 @@ where
             Ready(Some(Ok(k))) => match k {
                 StreamItem::Log(item) => Ready(Some(Ok(StreamItem::Log(item)))),
                 StreamItem::Stats(item) => Ready(Some(Ok(StreamItem::Stats(item)))),
-                StreamItem::DataItem(item) => Ready(Some(Ok(StreamItem::DataItem(item.into_agg())))),
+                StreamItem::DataItem(item) => match item {
+                    RangeCompletableItem::RangeComplete => {
+                        Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete))))
+                    }
+                    RangeCompletableItem::Data(item) => Ready(Some(Ok(StreamItem::DataItem(
+                        RangeCompletableItem::Data(item.into_agg()),
+                    )))),
+                },
             },
             Ready(Some(Err(e))) => Ready(Some(Err(e))),
             Ready(None) => Ready(None),

disk/src/agg/eventbatch.rs

@@ -1,7 +1,11 @@
 use crate::agg::binnedt::{AggregatableTdim, AggregatorTdim};
 use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
+use crate::agg::streams::StreamItem;
 use crate::agg::AggregatableXdim1Bin;
+use crate::binned::{MakeBytesFrame, RangeCompletableItem};
+use crate::frame::makeframe::make_frame;
 use bytes::{BufMut, Bytes, BytesMut};
+use err::Error;
 use netpod::log::*;
 use serde::{Deserialize, Serialize};
 use std::mem::size_of;
@@ -230,3 +234,9 @@ impl AggregatorTdim for MinMaxAvgScalarEventBatchAggregator {
         vec![v]
     }
 }
+
+impl MakeBytesFrame for Result<StreamItem<RangeCompletableItem<MinMaxAvgScalarEventBatch>>, Error> {
+    fn make_bytes_frame(&self) -> Result<Bytes, Error> {
+        Ok(make_frame(self)?.freeze())
+    }
+}

disk/src/agg/scalarbinbatch.rs

@@ -1,5 +1,5 @@
 use crate::agg::binnedt::{AggregatableTdim, AggregatorTdim};
-use crate::agg::streams::Bins;
+use crate::agg::streams::{Bins, StreamItem};
 use crate::agg::{AggregatableXdim1Bin, Fits, FitsInside};
 use crate::binned::{MakeBytesFrame, RangeCompletableItem};
 use crate::frame::makeframe::make_frame;
@@ -302,7 +302,7 @@ impl AggregatorTdim for MinMaxAvgScalarBinBatchAggregator {
     }
 }
 
-impl MakeBytesFrame for Result<RangeCompletableItem<MinMaxAvgScalarBinBatch>, Error> {
+impl MakeBytesFrame for Result<StreamItem<RangeCompletableItem<MinMaxAvgScalarBinBatch>>, Error> {
     fn make_bytes_frame(&self) -> Result<Bytes, Error> {
         Ok(make_frame(self)?.freeze())
     }

disk/src/agg/streams.rs

@@ -28,7 +28,7 @@ pub trait Collected {
 
 pub trait Collectable {
     type Collected: Collected;
-    fn append_to(&mut self, collected: &mut Self::Collected);
+    fn append_to(&self, collected: &mut Self::Collected);
 }
 
 pub trait ToJsonResult {

disk/src/aggtest.rs

@@ -67,14 +67,9 @@ async fn agg_x_dim_0_inner() {
     );
     let fut1 = IntoDim1F32Stream::into_dim_1_f32_stream(fut1);
     let fut1 = IntoBinnedXBins1::into_binned_x_bins_1(fut1);
+    let fut1 = IntoBinnedT::into_binned_t(fut1, BinnedRange::covering_range(range, bin_count).unwrap().unwrap());
     let fut1 = fut1
-        .into_binned_t(BinnedRange::covering_range(range, bin_count).unwrap().unwrap())
+        //.into_binned_t(BinnedRange::covering_range(range, bin_count).unwrap().unwrap())
-        .map(|k| {
-            if false {
-                trace!("after T binning  {:?}", k.as_ref().unwrap());
-            }
-            k
-        })
         .for_each(|_k| ready(()));
     fut1.await;
 }

disk/src/binned.rs

@@ -1,11 +1,11 @@
 use crate::agg::binnedt::{AggregatableTdim, AggregatorTdim, IntoBinnedT};
+use crate::agg::binnedt2::AggregatableTdim2;
 use crate::agg::eventbatch::MinMaxAvgScalarEventBatch;
 use crate::agg::scalarbinbatch::{MinMaxAvgScalarBinBatch, MinMaxAvgScalarBinBatchAggregator};
 use crate::agg::streams::{Collectable, Collected, StreamItem, ToJsonResult};
 use crate::agg::{AggregatableXdim1Bin, Fits, FitsInside};
 use crate::binned::scalar::binned_stream;
-use crate::binnedstream::{BinnedScalarStreamFromPreBinnedPatches, BinnedStream};
+use crate::binnedstream::{BinnedScalarStreamFromPreBinnedPatches, BoxedStream};
-use crate::cache::pbvfs::PreBinnedScalarItem;
 use crate::cache::{BinnedQuery, MergedFromRemotes};
 use crate::channelconfig::{extract_matching_config_entry, read_local_config};
 use crate::frame::makeframe::make_frame;
@@ -32,7 +32,7 @@ use tokio::io::{AsyncRead, ReadBuf};
 pub mod scalar;
 
 pub struct BinnedStreamRes<I> {
-    pub binned_stream: BinnedStream<I>,
+    pub binned_stream: BoxedStream<Result<StreamItem<RangeCompletableItem<I>>, Error>>,
     pub range: BinnedRange,
 }
 
@@ -117,18 +117,29 @@ impl ToJsonResult for MinMaxAvgScalarBinBatchCollected {
     }
 }
 
-impl MakeBytesFrame for Result<StreamItem<RangeCompletableItem<MinMaxAvgScalarBinBatch>>, Error> {
+impl ToJsonResult for MinMaxAvgScalarBinBatch {
-    fn make_bytes_frame(&self) -> Result<Bytes, Error> {
+    type Output = MinMaxAvgScalarBinBatch;
-        Ok(make_frame(self)?.freeze())
+
+    fn to_json_result(&self) -> Result<Self::Output, Error> {
+        err::todo();
+        let ret = MinMaxAvgScalarBinBatch {
+            ts1s: self.ts1s.clone(),
+            ts2s: self.ts2s.clone(),
+            counts: self.counts.clone(),
+            mins: self.mins.clone(),
+            maxs: self.maxs.clone(),
+            avgs: self.avgs.clone(),
+        };
+        Ok(ret)
     }
 }
 
-type BinnedStreamBox = Pin<Box<dyn Stream<Item = Result<Bytes, Error>> + Send>>;
+type BinnedBytesStreamBox = Pin<Box<dyn Stream<Item = Result<Bytes, Error>> + Send>>;
 
 pub async fn binned_bytes_for_http(
     node_config: &NodeConfigCached,
     query: &BinnedQuery,
-) -> Result<BinnedStreamBox, Error> {
+) -> Result<BinnedBytesStreamBox, Error> {
     let channel_config = read_local_config(&query.channel(), &node_config.node).await?;
     let entry = extract_matching_config_entry(query.range(), &channel_config)?;
     info!("binned_bytes_for_http  found config entry {:?}", entry);
@@ -139,6 +150,8 @@ pub async fn binned_bytes_for_http(
             Ok(Box::pin(ret))
         }
         AggKind::DimXBinsN(_) => {
+            // TODO pass a different stream kind here:
+            err::todo();
             let res = binned_stream(node_config, query, BinnedStreamKindScalar::new()).await?;
             let ret = BinnedBytesForHttpStream::new(res.binned_stream);
             Ok(Box::pin(ret))
@@ -220,7 +233,7 @@ impl Serialize for IsoDateTime {
 }
 
 pub async fn collect_all<T>(
-    stream: impl Stream<Item = Result<StreamItem<T>, Error>> + Unpin,
+    stream: impl Stream<Item = Result<StreamItem<RangeCompletableItem<T>>, Error>> + Unpin,
     bin_count_exp: u32,
 ) -> Result<<T as Collectable>::Collected, Error>
 where
@@ -249,10 +262,13 @@ where
                     Ok(item) => match item {
                         StreamItem::Log(_) => {}
                         StreamItem::Stats(_) => {}
-                        StreamItem::DataItem(mut item) => {
+                        StreamItem::DataItem(item) => match item {
-                            item.append_to(&mut main_item);
+                            RangeCompletableItem::RangeComplete => {}
-                            i1 += 1;
+                            RangeCompletableItem::Data(mut item) => {
-                        }
+                                item.append_to(&mut main_item);
+                                i1 += 1;
+                            }
+                        },
                     },
                     Err(e) => {
                         // TODO  Need to use some flags to get good enough error message for remote user.
@@ -286,23 +302,24 @@ pub async fn binned_json(node_config: &NodeConfigCached, query: &BinnedQuery) ->
     // TODO create the matching stream based on AggKind and ConfigEntry.
 
     let t = binned_stream(node_config, query, BinnedStreamKindScalar::new()).await?;
+    // TODO need to collect also timeout, number of missing expected bins, ...
     let collected = collect_all(t.binned_stream, t.range.count as u32).await?;
-    let ret = collected.to_json_result()?;
+    let ret = ToJsonResult::to_json_result(&collected)?;
     Ok(serde_json::to_value(ret)?)
 }
 
-pub struct ReadPbv<PBI>
+pub struct ReadPbv<T>
 where
-    PBI: PreBinnedItem,
+    T: ReadableFromFile,
 {
     buf: Vec<u8>,
     file: Option<File>,
-    _mark: std::marker::PhantomData<PBI>,
+    _mark: std::marker::PhantomData<T>,
 }
 
-impl<PBI> ReadPbv<PBI>
+impl<T> ReadPbv<T>
 where
-    PBI: PreBinnedItem,
+    T: ReadableFromFile,
 {
     fn new(file: File) -> Self {
         Self {
@@ -313,11 +330,11 @@ where
     }
 }
 
-impl<PBI> Future for ReadPbv<PBI>
+impl<T> Future for ReadPbv<T>
 where
-    PBI: PreBinnedItem,
+    T: ReadableFromFile,
 {
-    type Output = Result<StreamItem<PBI>, Error>;
+    type Output = Result<StreamItem<RangeCompletableItem<T>>, Error>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
         use Poll::*;
@@ -332,8 +349,8 @@ where
                         self.buf.extend_from_slice(&mut buf);
                         Pending
                     } else {
-                        match PBI::from_buf(&mut self.buf) {
+                        match T::from_buf(&mut self.buf) {
-                            Ok(item) => Ready(Ok(StreamItem::DataItem(item))),
+                            Ok(item) => Ready(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))),
                             Err(e) => Ready(Err(e)),
                         }
                     }
@@ -345,93 +362,61 @@ where
     }
 }
 
-pub trait PreBinnedItem: Send + Serialize + DeserializeOwned + Unpin {
+pub trait ReadableFromFile: Sized {
-    type BinnedStreamItem: AggregatableTdim + Unpin + Send;
+    fn read_from_file(file: File) -> Result<ReadPbv<Self>, Error>;
-    fn into_binned_stream_item(self, fit_range: NanoRange) -> Option<Self::BinnedStreamItem>;
+    // TODO should not need this:
-    fn make_range_complete() -> Self;
+    fn from_buf(buf: &[u8]) -> Result<Self, Error>;
-    fn read_pbv(file: File) -> Result<ReadPbv<Self>, Error>
-    where
-        Self: Sized;
-    fn from_buf(buf: &[u8]) -> Result<Self, Error>
-    where
-        Self: Sized;
 }
 
-impl PreBinnedItem for PreBinnedScalarItem {
+impl ReadableFromFile for MinMaxAvgScalarBinBatch {
-    type BinnedStreamItem = BinnedScalarStreamItem;
+    fn read_from_file(file: File) -> Result<ReadPbv<Self>, Error> {
-
-    fn into_binned_stream_item(self, fit_range: NanoRange) -> Option<Self::BinnedStreamItem> {
-        match self {
-            Self::RangeComplete => Some(Self::BinnedStreamItem::RangeComplete),
-            Self::Batch(item) => match item.fits_inside(fit_range) {
-                Fits::Inside | Fits::PartlyGreater | Fits::PartlyLower | Fits::PartlyLowerAndGreater => {
-                    Some(Self::BinnedStreamItem::Values(item))
-                }
-                _ => None,
-            },
-        }
-    }
-
-    fn make_range_complete() -> Self {
-        Self::RangeComplete
-    }
-
-    fn read_pbv(file: File) -> Result<ReadPbv<Self>, Error> {
         Ok(ReadPbv::new(file))
     }
-
     fn from_buf(buf: &[u8]) -> Result<Self, Error> {
         let dec: MinMaxAvgScalarBinBatch = serde_cbor::from_slice(&buf)?;
-        Ok(Self::Batch(dec))
+        Ok(dec)
     }
 }
 
-pub trait XBinnedEventsStreamItem:
+pub trait FilterFittingInside: Sized {
-    Send + Serialize + DeserializeOwned + Unpin + Collectable + Collected + AggregatableTdim
+    fn filter_fitting_inside(self, fit_range: NanoRange) -> Option<Self>;
-{
-    fn make_range_complete() -> Self;
 }
 
-impl Collected for MinMaxAvgScalarEventBatchStreamItem {
+impl FilterFittingInside for MinMaxAvgScalarBinBatch {
+    fn filter_fitting_inside(self, fit_range: NanoRange) -> Option<Self> {
+        match self.fits_inside(fit_range) {
+            Fits::Inside | Fits::PartlyGreater | Fits::PartlyLower | Fits::PartlyLowerAndGreater => Some(self),
+            _ => None,
+        }
+    }
+}
+
+impl Collected for MinMaxAvgScalarEventBatch {
     // TODO for this case we don't have an expected number of events. Factor out into another trait?
     fn new(bin_count_exp: u32) -> Self {
         // TODO factor out the concept of RangeComplete into another trait layer:
-        Self::Values(MinMaxAvgScalarEventBatch::empty())
+        MinMaxAvgScalarEventBatch::empty()
     }
+
     fn timed_out(&mut self, k: bool) {}
 }
 
-impl Collectable for MinMaxAvgScalarEventBatchStreamItem {
+impl Collectable for MinMaxAvgScalarEventBatch {
-    type Collected = MinMaxAvgScalarEventBatchStreamItem;
+    type Collected = MinMaxAvgScalarEventBatch;
-    fn append_to(&mut self, collected: &mut Self::Collected) {
-        match self {
-            Self::RangeComplete => {
-                // TODO would be more nice to insert another type layer for RangeComplete concept.
-                panic!()
-            }
-            Self::Values(this) => match collected {
-                Self::RangeComplete => {}
-                Self::Values(coll) => {
-                    coll.tss.append(&mut coll.tss);
-                    coll.mins.append(&mut coll.mins);
-                    coll.maxs.append(&mut coll.maxs);
-                    coll.avgs.append(&mut coll.avgs);
-                }
-            },
-        }
-    }
-}
 
-impl XBinnedEventsStreamItem for MinMaxAvgScalarEventBatchStreamItem {
+    fn append_to(&self, collected: &mut Self::Collected) {
-    fn make_range_complete() -> Self {
+        // TODO create separate traits for different concerns:
-        Self::RangeComplete
+        // Some occasion I want to just append.
+        // In other case, I need to collect also timeout flag, missing bin count and such.
+        collected.tss.extend_from_slice(&self.tss);
+        collected.mins.extend_from_slice(&self.mins);
+        collected.maxs.extend_from_slice(&self.maxs);
+        collected.avgs.extend_from_slice(&self.avgs);
     }
 }
 
 pub trait TBinned: Send + Serialize + DeserializeOwned + Unpin + Collectable + AggregatableTdim<Output = Self> {}
 
-impl TBinned for MinMaxAvgScalarBinBatchStreamItem {}
-
 impl Collected for MinMaxAvgScalarBinBatch {
     fn new(bin_count_exp: u32) -> Self {
         MinMaxAvgScalarBinBatch::empty()
@@ -441,21 +426,45 @@ impl Collected for MinMaxAvgScalarBinBatch {
 
 impl Collectable for MinMaxAvgScalarBinBatch {
     type Collected = MinMaxAvgScalarBinBatch;
-    fn append_to(&mut self, collected: &mut Self::Collected) {
+    fn append_to(&self, collected: &mut Self::Collected) {
-        collected.ts1s.append(&mut self.ts1s);
+        collected.ts1s.extend_from_slice(&self.ts1s);
-        collected.ts2s.append(&mut self.ts2s);
+        collected.ts2s.extend_from_slice(&self.ts2s);
-        collected.counts.append(&mut self.counts);
+        collected.counts.extend_from_slice(&self.counts);
-        collected.mins.append(&mut self.mins);
+        collected.mins.extend_from_slice(&self.mins);
-        collected.maxs.append(&mut self.maxs);
+        collected.maxs.extend_from_slice(&self.maxs);
-        collected.avgs.append(&mut self.avgs);
+        collected.avgs.extend_from_slice(&self.avgs);
     }
 }
 
+pub trait XBinnedEvents:
+    Sized + Unpin + Send + Serialize + DeserializeOwned + Collectable + Collected + AggregatableTdim
+{
+}
+
+impl XBinnedEvents for MinMaxAvgScalarEventBatch {}
+
+impl TBinnedBins for MinMaxAvgScalarBinBatch {}
+
+pub trait TBinnedBins:
+    Sized
+    + Unpin
+    + Send
+    + Serialize
+    + DeserializeOwned
+    + Collectable
+    + Collected
+    + ReadableFromFile
+    + FilterFittingInside
+    + AggregatableTdim2
+{
+}
+
 pub trait BinnedStreamKind: Clone + Unpin + Send + Sync + 'static {
-    type BinnedStreamItem: MakeBytesFrame;
+    type TBinnedStreamType: Stream<Item = Result<StreamItem<RangeCompletableItem<Self::TBinnedBins>>, Error>>
-    type BinnedStreamType: Stream + Send + 'static;
+        + Send
-    type PreBinnedItem: PreBinnedItem;
+        + 'static;
-    type XBinnedEvents;
+    type XBinnedEvents: XBinnedEvents;
+    type TBinnedBins: TBinnedBins;
 
     fn new_binned_from_prebinned(
         &self,
@@ -463,7 +472,7 @@ pub trait BinnedStreamKind: Clone + Unpin + Send + Sync + 'static {
         range: BinnedRange,
         pre_range: PreBinnedPatchRange,
         node_config: &NodeConfigCached,
-    ) -> Result<Self::BinnedStreamType, Error>;
+    ) -> Result<Self::TBinnedStreamType, Error>;
 
     fn new_binned_from_merged(
         &self,
@@ -471,14 +480,13 @@ pub trait BinnedStreamKind: Clone + Unpin + Send + Sync + 'static {
         perf_opts: PerfOpts,
         range: BinnedRange,
         node_config: &NodeConfigCached,
-    ) -> Result<Self::BinnedStreamType, Error>;
+    ) -> Result<Self::TBinnedStreamType, Error>;
-
-    fn pbv_handle_fut2_item(item: StreamItem<Self::PreBinnedItem>) -> Option<StreamItem<Self::PreBinnedItem>>;
 }
 
 #[derive(Clone)]
 pub struct BinnedStreamKindScalar {}
 
+#[derive(Clone)]
 pub struct BinnedStreamKindWave {}
 
 impl BinnedStreamKindScalar {
@@ -493,16 +501,17 @@ impl BinnedStreamKindWave {
     }
 }
 
+#[derive(Debug, Serialize, Deserialize)]
 pub enum RangeCompletableItem<T> {
     RangeComplete,
     Data(T),
 }
 
 impl BinnedStreamKind for BinnedStreamKindScalar {
-    type BinnedStreamItem = Result<StreamItem<BinnedScalarStreamItem>, Error>;
+    // TODO is this really needed?
-    type BinnedStreamType = BinnedStream<Self::BinnedStreamItem>;
+    type TBinnedStreamType = BoxedStream<Result<StreamItem<RangeCompletableItem<Self::TBinnedBins>>, Error>>;
-    type PreBinnedItem = PreBinnedScalarItem;
     type XBinnedEvents = MinMaxAvgScalarEventBatch;
+    type TBinnedBins = MinMaxAvgScalarBinBatch;
 
     fn new_binned_from_prebinned(
         &self,
@@ -510,7 +519,7 @@ impl BinnedStreamKind for BinnedStreamKindScalar {
         range: BinnedRange,
         pre_range: PreBinnedPatchRange,
         node_config: &NodeConfigCached,
-    ) -> Result<Self::BinnedStreamType, Error> {
+    ) -> Result<Self::TBinnedStreamType, Error> {
         let s = BinnedScalarStreamFromPreBinnedPatches::new(
             PreBinnedPatchIterator::from_range(pre_range),
             query.channel().clone(),
@@ -521,7 +530,7 @@ impl BinnedStreamKind for BinnedStreamKindScalar {
             query.disk_stats_every().clone(),
             self.clone(),
         )?;
-        Ok(BinnedStream::new(Box::pin(s))?)
+        Ok(BoxedStream::new(Box::pin(s))?)
     }
 
     fn new_binned_from_merged(
@@ -530,37 +539,33 @@ impl BinnedStreamKind for BinnedStreamKindScalar {
         perf_opts: PerfOpts,
         range: BinnedRange,
         node_config: &NodeConfigCached,
-    ) -> Result<Self::BinnedStreamType, Error> {
+    ) -> Result<Self::TBinnedStreamType, Error> {
         let s = MergedFromRemotes::new(evq, perf_opts, node_config.node_config.cluster.clone(), self.clone());
         // TODO use the binned2 instead
         let s = crate::agg::binnedt::IntoBinnedT::into_binned_t(s, range);
-        let s = s.map(adapter_to_stream_item);
+        Ok(BoxedStream::new(Box::pin(s))?)
-        Ok(BinnedStream::new(Box::pin(s))?)
-    }
-
-    fn pbv_handle_fut2_item(item: StreamItem<Self::PreBinnedItem>) -> Option<StreamItem<Self::PreBinnedItem>> {
-        // TODO make this code work in this context:
-        // Do I need more parameters here?
-        /*Ok(item) => match item {
-            StreamItem::Log(item) => Ready(Some(Ok(StreamItem::Log(item)))),
-            StreamItem::Stats(item) => Ready(Some(Ok(StreamItem::Stats(item)))),
-            StreamItem::DataItem(item) => match item {
-                PreBinnedScalarItem::RangeComplete => {
-                    self.range_complete_observed = true;
-                    None
-                }
-                PreBinnedScalarItem::Batch(batch) => {
-                    self.values.ts1s.extend(batch.ts1s.iter());
-                    self.values.ts2s.extend(batch.ts2s.iter());
-                    self.values.counts.extend(batch.counts.iter());
-                    self.values.mins.extend(batch.mins.iter());
-                    self.values.maxs.extend(batch.maxs.iter());
-                    self.values.avgs.extend(batch.avgs.iter());
-                    StreamItem::DataItem(PreBinnedScalarItem::Batch(batch))
-                }
-            },
-        },*/
-        err::todo();
-        None
     }
 }
+
+// TODO this code is needed somewhere:
+fn pbv_handle_fut2_item(
+    item: StreamItem<RangeCompletableItem<MinMaxAvgScalarBinBatch>>,
+) -> Option<StreamItem<RangeCompletableItem<MinMaxAvgScalarBinBatch>>> {
+    // TODO make this code work in this context:
+    // Do I need more parameters here?
+    /*Ok(item) => match item {
+        StreamItem::DataItem(item) => match item {
+            PreBinnedScalarItem::Batch(batch) => {
+                self.values.ts1s.extend(batch.ts1s.iter());
+                self.values.ts2s.extend(batch.ts2s.iter());
+                self.values.counts.extend(batch.counts.iter());
+                self.values.mins.extend(batch.mins.iter());
+                self.values.maxs.extend(batch.maxs.iter());
+                self.values.avgs.extend(batch.avgs.iter());
+                StreamItem::DataItem(PreBinnedScalarItem::Batch(batch))
+            }
+        },
+    },*/
+    err::todo();
+    None
+}

disk/src/binned/scalar.rs

@@ -1,7 +1,7 @@
 use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
 use crate::agg::streams::StreamItem;
 use crate::binned::{BinnedStreamKind, BinnedStreamRes, RangeCompletableItem};
-use crate::binnedstream::BinnedStream;
+use crate::binnedstream::BoxedStream;
 use crate::cache::BinnedQuery;
 use crate::raw::EventsQuery;
 use err::Error;
@@ -9,13 +9,13 @@ use futures_core::Stream;
 use netpod::log::*;
 use netpod::{BinnedRange, NodeConfigCached, PerfOpts, PreBinnedPatchRange};
 
-pub async fn binned_stream<BK>(
+pub async fn binned_stream<SK>(
     node_config: &NodeConfigCached,
     query: &BinnedQuery,
-    stream_kind: BK,
+    stream_kind: SK,
-) -> Result<BinnedStreamRes<<BK::BinnedStreamType as Stream>::Item>, Error>
+) -> Result<BinnedStreamRes<SK::TBinnedBins>, Error>
 where
-    BK: BinnedStreamKind,
+    SK: BinnedStreamKind,
 {
     if query.channel().backend != node_config.node.backend {
         let err = Error::with_msg(format!(
@@ -40,8 +40,8 @@ where
                 );
                 return Err(Error::with_msg(msg));
             }
-            let s = BK::new_binned_from_prebinned(&stream_kind, query, range.clone(), pre_range, node_config)?;
+            let s = SK::new_binned_from_prebinned(&stream_kind, query, range.clone(), pre_range, node_config)?;
-            let s = BinnedStream::new(Box::pin(s))?;
+            let s = BoxedStream::new(Box::pin(s))?;
             let ret = BinnedStreamRes {
                 binned_stream: s,
                 range,
@@ -59,8 +59,8 @@ where
                 agg_kind: query.agg_kind().clone(),
             };
             // TODO do I need to set up more transformations or binning to deliver the requested data?
-            let s = BK::new_binned_from_merged(&stream_kind, evq, perf_opts, range.clone(), node_config)?;
+            let s = SK::new_binned_from_merged(&stream_kind, evq, perf_opts, range.clone(), node_config)?;
-            let s = BinnedStream::new(Box::pin(s))?;
+            let s = BoxedStream::new(Box::pin(s))?;
             let ret = BinnedStreamRes {
                 binned_stream: s,
                 range,

disk/src/binnedstream.rs

@@ -1,5 +1,5 @@
 use crate::agg::streams::StreamItem;
-use crate::binned::{BinnedStreamKind, PreBinnedItem};
+use crate::binned::{BinnedStreamKind, RangeCompletableItem};
 use crate::cache::pbvfs::PreBinnedScalarValueFetchedStream;
 use crate::cache::{CacheUsage, PreBinnedQuery};
 use crate::frame::makeframe::FrameType;
@@ -16,15 +16,10 @@ pub struct BinnedScalarStreamFromPreBinnedPatches<BK>
 where
     BK: BinnedStreamKind,
 {
-    //inp: Pin<Box<dyn Stream<Item = Result<StreamItem<BinnedScalarStreamItem>, Error>> + Send>>,
     inp: Pin<
         Box<
-            dyn Stream<
+            dyn Stream<Item = Result<StreamItem<RangeCompletableItem<<BK as BinnedStreamKind>::TBinnedBins>>, Error>>
-                    Item = Result<
+                + Send,
-                        StreamItem<<<BK as BinnedStreamKind>::PreBinnedItem as PreBinnedItem>::BinnedStreamItem>,
-                        Error,
-                    >,
-                > + Send,
         >,
     >,
     stream_kind: BK,
@@ -33,7 +28,7 @@ where
 impl<BK> BinnedScalarStreamFromPreBinnedPatches<BK>
 where
     BK: BinnedStreamKind,
-    Result<StreamItem<<BK as BinnedStreamKind>::PreBinnedItem>, err::Error>: FrameType,
+    Result<StreamItem<RangeCompletableItem<<BK as BinnedStreamKind>::TBinnedBins>>, Error>: FrameType,
 {
     pub fn new(
         patch_it: PreBinnedPatchIterator,
@@ -87,12 +82,17 @@ where
                         Ok(item) => match item {
                             StreamItem::Log(item) => Some(Ok(StreamItem::Log(item))),
                             StreamItem::Stats(item) => Some(Ok(StreamItem::Stats(item))),
-                            StreamItem::DataItem(item) => {
+                            StreamItem::DataItem(item) => match item {
-                                match crate::binned::PreBinnedItem::into_binned_stream_item(item, fit_range) {
+                                RangeCompletableItem::RangeComplete => {
-                                    Some(item) => Some(Ok(StreamItem::DataItem(item))),
+                                    Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete)))
-                                    None => None,
                                 }
-                            }
+                                RangeCompletableItem::Data(item) => {
+                                    match crate::binned::FilterFittingInside::filter_fitting_inside(item, fit_range) {
+                                        Some(item) => Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))),
+                                        None => None,
+                                    }
+                                }
+                            },
                         },
                         Err(e) => Some(Err(e)),
                     };
@@ -100,8 +100,8 @@ where
                 }
             });
         // TODO activate the T-binning via the bin-to-bin binning trait.
-        err::todo();
+        //err::todo();
-        //let inp = IntoBinnedT::into_binned_t(inp, range);
+        let inp = crate::agg::binnedt2::IntoBinnedT::into_binned_t(inp, range);
         Ok(Self {
             inp: Box::pin(inp),
             stream_kind,
@@ -109,33 +109,33 @@ where
     }
 }
 
-impl<BK> Stream for BinnedScalarStreamFromPreBinnedPatches<BK>
+// TODO change name, type is generic now:
+// Can I remove the whole type or keep for static check?
+impl<SK> Stream for BinnedScalarStreamFromPreBinnedPatches<SK>
 where
-    BK: BinnedStreamKind,
+    SK: BinnedStreamKind,
 {
-    type Item = Result<StreamItem<<<BK as BinnedStreamKind>::PreBinnedItem as PreBinnedItem>::BinnedStreamItem>, Error>;
+    type Item = Result<StreamItem<RangeCompletableItem<SK::TBinnedBins>>, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
-        use Poll::*;
+        self.inp.poll_next_unpin(cx)
-        match self.inp.poll_next_unpin(cx) {
-            Ready(Some(item)) => Ready(Some(item)),
-            Ready(None) => Ready(None),
-            Pending => Pending,
-        }
     }
 }
 
-pub struct BinnedStream<I> {
+pub struct BoxedStream<I> {
     inp: Pin<Box<dyn Stream<Item = I> + Send>>,
 }
 
-impl<I> BinnedStream<I> {
+impl<I> BoxedStream<I> {
-    pub fn new(inp: Pin<Box<dyn Stream<Item = I> + Send>>) -> Result<Self, Error> {
+    pub fn new<T>(inp: T) -> Result<Self, Error>
-        Ok(Self { inp })
+    where
+        T: Stream<Item = I> + Send + 'static,
+    {
+        Ok(Self { inp: Box::pin(inp) })
     }
 }
 
-impl<I> Stream for BinnedStream<I> {
+impl<I> Stream for BoxedStream<I> {
     type Item = I;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {

disk/src/cache.rs

@@ -230,14 +230,14 @@ fn channel_from_params(params: &BTreeMap<String, String>) -> Result<Channel, Err
 // NOTE  This answers a request for a single valid pre-binned patch.
 // A user must first make sure that the grid spec is valid, and that this node is responsible for it.
 // Otherwise it is an error.
-pub fn pre_binned_bytes_for_http<'a, BK>(
+pub fn pre_binned_bytes_for_http<SK>(
     node_config: &NodeConfigCached,
     query: &PreBinnedQuery,
-    stream_kind: BK,
+    stream_kind: SK,
-) -> Result<PreBinnedValueByteStream<BK>, Error>
+) -> Result<PreBinnedValueByteStream<SK>, Error>
 where
-    BK: BinnedStreamKind,
+    SK: BinnedStreamKind,
-    Result<StreamItem<<BK as BinnedStreamKind>::PreBinnedItem>, err::Error>: FrameType,
+    Result<StreamItem<RangeCompletableItem<SK::TBinnedBins>>, err::Error>: FrameType,
 {
     if query.channel.backend != node_config.node.backend {
         let err = Error::with_msg(format!(
@@ -508,13 +508,16 @@ impl CacheFileDesc {
     }
 }
 
-pub async fn write_pb_cache_min_max_avg_scalar(
+pub async fn write_pb_cache_min_max_avg_scalar<T>(
-    values: MinMaxAvgScalarBinBatch,
+    values: T,
     patch: PreBinnedPatchCoord,
     agg_kind: AggKind,
     channel: Channel,
     node_config: NodeConfigCached,
-) -> Result<(), Error> {
+) -> Result<(), Error>
+where
+    T: Serialize,
+{
     let cfd = CacheFileDesc {
         channel: channel.clone(),
         patch: patch.clone(),

disk/src/cache/pbv.rs

@@ -1,8 +1,9 @@
 use crate::agg::binnedt::IntoBinnedT;
 use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
-use crate::agg::streams::StreamItem;
+use crate::agg::streams::{Collectable, Collected, StreamItem};
-use crate::binned::{BinnedStreamKind, PreBinnedItem, RangeCompletableItem};
+use crate::binned::RangeCompletableItem::RangeComplete;
-use crate::cache::pbvfs::{PreBinnedScalarItem, PreBinnedScalarValueFetchedStream};
+use crate::binned::{BinnedStreamKind, RangeCompletableItem};
+use crate::cache::pbvfs::PreBinnedScalarValueFetchedStream;
 use crate::cache::{CacheFileDesc, MergedFromRemotes, PreBinnedQuery};
 use crate::frame::makeframe::{make_frame, FrameType};
 use crate::raw::EventsQuery;
@@ -35,8 +36,8 @@ pub fn pre_binned_value_byte_stream_new<SK>(
     stream_kind: SK,
 ) -> PreBinnedValueByteStream<SK>
 where
-    SK: BinnedStreamKind + Unpin,
+    SK: BinnedStreamKind,
-    Result<StreamItem<<SK as BinnedStreamKind>::PreBinnedItem>, err::Error>: FrameType,
+    Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::TBinnedBins>>, err::Error>: FrameType,
 {
     let s1 = PreBinnedValueStream::new(query.clone(), node_config, stream_kind);
     let s2 = PreBinnedValueByteStreamInner { inp: s1 };
@@ -45,9 +46,8 @@ where
 
 impl<SK> Stream for PreBinnedValueByteStreamInner<SK>
 where
-    SK: BinnedStreamKind + Unpin,
+    SK: BinnedStreamKind,
-    Result<StreamItem<<SK as BinnedStreamKind>::PreBinnedItem>, err::Error>: FrameType,
+    Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::TBinnedBins>>, err::Error>: FrameType,
-    PreBinnedValueStream<SK>: Unpin,
 {
     type Item = Result<Bytes, Error>;
 
@@ -55,7 +55,10 @@ where
         use Poll::*;
         match self.inp.poll_next_unpin(cx) {
             Ready(Some(item)) => {
-                match make_frame::<Result<StreamItem<<SK as BinnedStreamKind>::PreBinnedItem>, Error>>(&item) {
+                match make_frame::<
+                    Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::TBinnedBins>>, err::Error>,
+                >(&item)
+                {
                     Ok(buf) => Ready(Some(Ok(buf.freeze()))),
                     Err(e) => Ready(Some(Err(e.into()))),
                 }
@@ -73,8 +76,18 @@ where
     query: PreBinnedQuery,
     node_config: NodeConfigCached,
     open_check_local_file: Option<Pin<Box<dyn Future<Output = Result<File, std::io::Error>> + Send>>>,
-    fut2:
+    fut2: Option<
-        Option<Pin<Box<dyn Stream<Item = Result<StreamItem<<SK as BinnedStreamKind>::PreBinnedItem>, Error>> + Send>>>,
+        Pin<
+            Box<
+                dyn Stream<
+                        Item = Result<
+                            StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::TBinnedBins>>,
+                            err::Error,
+                        >,
+                    > + Send,
+            >,
+        >,
+    >,
     read_from_cache: bool,
     cache_written: bool,
     data_complete: bool,
@@ -83,10 +96,19 @@ where
     errored: bool,
     completed: bool,
     streamlog: Streamlog,
-    values: MinMaxAvgScalarBinBatch,
+    values: <SK as BinnedStreamKind>::TBinnedBins,
     write_fut: Option<Pin<Box<dyn Future<Output = Result<(), Error>> + Send>>>,
     read_cache_fut: Option<
-        Pin<Box<dyn Future<Output = Result<StreamItem<<SK as BinnedStreamKind>::PreBinnedItem>, Error>> + Send>>,
+        Pin<
+            Box<
+                dyn Future<
+                        Output = Result<
+                            StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::TBinnedBins>>,
+                            err::Error,
+                        >,
+                    > + Send,
+            >,
+        >,
     >,
     stream_kind: SK,
 }
@@ -94,7 +116,7 @@ where
 impl<SK> PreBinnedValueStream<SK>
 where
     SK: BinnedStreamKind,
-    Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::PreBinnedItem>>, err::Error>: FrameType,
+    Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::TBinnedBins>>, err::Error>: FrameType,
 {
     pub fn new(query: PreBinnedQuery, node_config: &NodeConfigCached, stream_kind: SK) -> Self {
         Self {
@@ -110,7 +132,8 @@ where
             errored: false,
             completed: false,
             streamlog: Streamlog::new(node_config.ix as u32),
-            values: MinMaxAvgScalarBinBatch::empty(),
+            // TODO refactor usage of parameter
+            values: <<SK as BinnedStreamKind>::TBinnedBins as Collected>::new(0),
             write_fut: None,
             read_cache_fut: None,
             stream_kind,
@@ -244,9 +267,9 @@ where
 impl<SK> Stream for PreBinnedValueStream<SK>
 where
     SK: BinnedStreamKind + Unpin,
-    Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::PreBinnedItem>>, err::Error>: FrameType,
+    Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::TBinnedBins>>, err::Error>: FrameType,
 {
-    type Item = Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::PreBinnedItem>>, Error>;
+    type Item = Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::TBinnedBins>>, err::Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;
@@ -301,7 +324,7 @@ where
                 if self.cache_written {
                     if self.range_complete_observed {
                         self.range_complete_emitted = true;
-                        let item = <<SK as BinnedStreamKind>::PreBinnedItem as PreBinnedItem>::make_range_complete();
+                        let item = RangeCompletableItem::RangeComplete;
                         Ready(Some(Ok(StreamItem::DataItem(item))))
                     } else {
                         self.completed = true;
@@ -316,10 +339,16 @@ where
                             let msg = format!(
                                 "write cache file  query: {:?}  bin count: {}",
                                 self.query.patch,
-                                self.values.ts1s.len()
+                                //self.values.ts1s.len()
+                                // TODO create trait to extract number of bins from item:
+                                0
                             );
                             self.streamlog.append(Level::INFO, msg);
-                            let values = std::mem::replace(&mut self.values, MinMaxAvgScalarBinBatch::empty());
+                            let values = std::mem::replace(
+                                &mut self.values,
+                                // Do not use expectation on the number of bins here:
+                                <<SK as BinnedStreamKind>::TBinnedBins as Collected>::new(0),
+                            );
                             let fut = super::write_pb_cache_min_max_avg_scalar(
                                 values,
                                 self.query.patch.clone(),
@@ -339,9 +368,22 @@ where
             } else if let Some(fut) = self.fut2.as_mut() {
                 match fut.poll_next_unpin(cx) {
                     Ready(Some(k)) => match k {
-                        Ok(item) => match SK::pbv_handle_fut2_item(item) {
+                        Ok(item) => match item {
-                            None => continue 'outer,
+                            StreamItem::Log(item) => Ready(Some(Ok(StreamItem::Log(item)))),
-                            Some(item) => Ready(Some(Ok(item))),
+                            StreamItem::Stats(item) => Ready(Some(Ok(StreamItem::Stats(item)))),
+                            StreamItem::DataItem(item) => match item {
+                                RangeCompletableItem::RangeComplete => {
+                                    self.range_complete_observed = true;
+                                    continue 'outer;
+                                }
+                                RangeCompletableItem::Data(item) => {
+                                    // TODO need trait Appendable which simply appends to the same type, so that I can
+                                    // write later the whole batch of numbers in one go.
+                                    err::todo();
+                                    //item.append_to(&mut self.values);
+                                    Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::Data(item)))))
+                                }
+                            },
                         },
                         Err(e) => {
                             self.errored = true;
@@ -361,7 +403,8 @@ where
                         match item {
                             Ok(file) => {
                                 self.read_from_cache = true;
-                                let fut = <SK as BinnedStreamKind>::PreBinnedItem::read_pbv(file)?;
+                                use crate::binned::ReadableFromFile;
+                                let fut = <<SK as BinnedStreamKind>::TBinnedBins as crate::binned::ReadableFromFile>::read_from_file(file)?;
                                 self.read_cache_fut = Some(Box::pin(fut));
                                 continue 'outer;
                             }

disk/src/cache/pbvfs.rs

@@ -1,6 +1,6 @@
 use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
 use crate::agg::streams::StreamItem;
-use crate::binned::BinnedStreamKind;
+use crate::binned::{BinnedStreamKind, RangeCompletableItem};
 use crate::cache::{node_ix_for_patch, HttpBodyAsAsyncRead, PreBinnedQuery};
 use crate::frame::inmem::InMemoryFrameAsyncReadStream;
 use crate::frame::makeframe::{decode_frame, FrameType};
@@ -52,18 +52,13 @@ where
     }
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+// TODO change name, is now generic:
-pub enum PreBinnedScalarItem {
+impl<SK> Stream for PreBinnedScalarValueFetchedStream<SK>
-    Batch(MinMaxAvgScalarBinBatch),
-    RangeComplete,
-}
-
-impl<BK> Stream for PreBinnedScalarValueFetchedStream<BK>
 where
-    BK: BinnedStreamKind,
+    SK: BinnedStreamKind,
-    Result<StreamItem<<BK as BinnedStreamKind>::PreBinnedItem>, err::Error>: FrameType,
+    Result<StreamItem<RangeCompletableItem<SK::TBinnedBins>>, err::Error>: FrameType,
 {
-    type Item = Result<StreamItem<BK::PreBinnedItem>, Error>;
+    type Item = Result<StreamItem<RangeCompletableItem<SK::TBinnedBins>>, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;
@@ -80,7 +75,7 @@ where
                         StreamItem::Log(item) => Ready(Some(Ok(StreamItem::Log(item)))),
                         StreamItem::Stats(item) => Ready(Some(Ok(StreamItem::Stats(item)))),
                         StreamItem::DataItem(item) => {
-                            match decode_frame::<Result<StreamItem<<BK as BinnedStreamKind>::PreBinnedItem>, Error>>(
+                            match decode_frame::<Result<StreamItem<RangeCompletableItem<SK::TBinnedBins>>, Error>>(
                                 &item,
                             ) {
                                 Ok(Ok(item)) => Ready(Some(Ok(item))),

disk/src/eventblobs.rs

@@ -1,6 +1,7 @@
 use crate::agg::streams::StreamItem;
+use crate::binned::RangeCompletableItem;
 use crate::dataopen::{open_files, OpenedFile};
-use crate::eventchunker::{EventChunker, EventChunkerConf};
+use crate::eventchunker::{EventChunker, EventChunkerConf, EventFull};
 use crate::file_content_stream;
 use crate::streamlog::LogItem;
 use err::Error;
@@ -56,7 +57,7 @@ impl EventBlobsComplete {
 }
 
 impl Stream for EventBlobsComplete {
-    type Item = Result<StreamItem<EventChunkerItem>, Error>;
+    type Item = Result<StreamItem<RangeCompletableItem<EventFull>>, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;

disk/src/frame/makeframe.rs

@@ -1,6 +1,7 @@
 use crate::agg::eventbatch::MinMaxAvgScalarEventBatch;
+use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
 use crate::agg::streams::StreamItem;
-use crate::cache::pbvfs::PreBinnedScalarItem;
+use crate::binned::RangeCompletableItem;
 use crate::frame::inmem::InMemoryFrame;
 use crate::raw::EventQueryJsonStringFrame;
 use bytes::{BufMut, BytesMut};
@@ -19,20 +20,12 @@ impl FrameType for EventQueryJsonStringFrame {
     const FRAME_TYPE_ID: u32 = 0x03;
 }
 
-impl FrameType for Result<StreamItem<BinnedScalarStreamItem>, Error> {
+impl FrameType for Result<StreamItem<RangeCompletableItem<MinMaxAvgScalarBinBatch>>, Error> {
-    const FRAME_TYPE_ID: u32 = 0x06;
+    const FRAME_TYPE_ID: u32 = 0x10;
 }
 
-impl FrameType for Result<MinMaxAvgScalarBinBatchStreamItem, Error> {
+impl FrameType for Result<StreamItem<RangeCompletableItem<MinMaxAvgScalarEventBatch>>, Error> {
-    const FRAME_TYPE_ID: u32 = 0x07;
+    const FRAME_TYPE_ID: u32 = 0x11;
-}
-
-impl FrameType for Result<StreamItem<PreBinnedScalarItem>, Error> {
-    const FRAME_TYPE_ID: u32 = 0x08;
-}
-
-impl FrameType for Result<StreamItem<MinMaxAvgScalarEventBatchStreamItem>, Error> {
-    const FRAME_TYPE_ID: u32 = 0x09;
 }
 
 pub fn make_frame<FT>(item: &FT) -> Result<BytesMut, Error>

disk/src/merge.rs

@@ -1,7 +1,7 @@
 use crate::agg::binnedt::AggregatableTdim;
 use crate::agg::eventbatch::MinMaxAvgScalarEventBatch;
 use crate::agg::streams::{Collectable, Collected, StatsItem, StreamItem};
-use crate::binned::{BinnedStreamKind, XBinnedEventsStreamItem};
+use crate::binned::{BinnedStreamKind, RangeCompletableItem};
 use crate::streamlog::LogItem;
 use err::Error;
 use futures_core::Stream;
@@ -14,7 +14,7 @@ use std::task::{Context, Poll};
 
 pub struct MergedMinMaxAvgScalarStream<S, SK>
 where
-    S: Stream<Item = Result<StreamItem<<SK as BinnedStreamKind>::XBinnedEvents>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<SK::XBinnedEvents>>, Error>> + Unpin,
     SK: BinnedStreamKind,
 {
     inps: Vec<S>,
@@ -35,7 +35,7 @@ where
 
 impl<S, SK> MergedMinMaxAvgScalarStream<S, SK>
 where
-    S: Stream<Item = Result<StreamItem<<SK as BinnedStreamKind>::XBinnedEvents>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<SK::XBinnedEvents>>, Error>> + Unpin,
     SK: BinnedStreamKind,
 {
     pub fn new(inps: Vec<S>) -> Self {
@@ -80,9 +80,8 @@ where
                                     }
                                     continue 'l1;
                                 }
-                                StreamItem::DataItem(item) => {
+                                StreamItem::DataItem(item) => match item {
-                                    // TODO factor out the concept of RangeComplete into another trait layer.
+                                    RangeCompletableItem::RangeComplete => {
-                                    if item.is_range_complete() {
                                         self.range_complete_observed[i1] = true;
                                         let d = self.range_complete_observed.iter().filter(|&&k| k).count();
                                         if d == self.range_complete_observed.len() {
@@ -92,11 +91,12 @@ where
                                             trace!("MergedStream  range_complete  d  {}", d);
                                         }
                                         continue 'l1;
-                                    } else {
+                                    }
+                                    RangeCompletableItem::Data(item) => {
                                         self.ixs[i1] = 0;
                                         self.current[i1] = MergedCurVal::Val(item);
                                     }
-                                }
+                                },
                             },
                             Ready(Some(Err(e))) => {
                                 // TODO emit this error, consider this stream as done, anything more to do here?
@@ -124,12 +124,13 @@ where
     }
 }
 
+// TODO change name, it is generic now:
 impl<S, SK> Stream for MergedMinMaxAvgScalarStream<S, SK>
 where
-    S: Stream<Item = Result<StreamItem<<SK as BinnedStreamKind>::XBinnedEvents>, Error>> + Unpin,
+    S: Stream<Item = Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::XBinnedEvents>>, Error>> + Unpin,
     SK: BinnedStreamKind,
 {
-    type Item = Result<StreamItem<<SK as BinnedStreamKind>::XBinnedEvents>, Error>;
+    type Item = Result<StreamItem<RangeCompletableItem<<SK as BinnedStreamKind>::XBinnedEvents>>, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;
@@ -150,9 +151,7 @@ where
                         Ready(None)
                     } else {
                         self.range_complete_observed_all_emitted = true;
-                        Ready(Some(Ok(StreamItem::DataItem(
+                        Ready(Some(Ok(StreamItem::DataItem(RangeCompletableItem::RangeComplete))))
-                            <<SK as BinnedStreamKind>::XBinnedEvents as XBinnedEventsStreamItem>::make_range_complete(),
-                        ))))
                     }
                 } else {
                     self.completed = true;

disk/src/raw/conn.rs

@@ -1,5 +1,6 @@
 use crate::agg::binnedx::IntoBinnedXBins1;
 use crate::agg::eventbatch::MinMaxAvgScalarEventBatch;
+use crate::agg::scalarbinbatch::MinMaxAvgScalarBinBatch;
 use crate::agg::streams::StreamItem;
 use crate::agg::IntoDim1F32Stream;
 use crate::binned::{BinnedStreamKind, BinnedStreamKindScalar, RangeCompletableItem};
@@ -56,7 +57,8 @@ async fn events_conn_handler_inner(
         Ok(_) => (),
         Err(mut ce) => {
             // TODO is it guaranteed to be compatible to serialize this way?
-            let buf = make_frame::<Result<StreamItem<MinMaxAvgScalarEventBatchStreamItem>, Error>>(&Err(ce.err))?;
+            let buf =
+                make_frame::<Result<StreamItem<RangeCompletableItem<MinMaxAvgScalarEventBatch>>, Error>>(&Err(ce.err))?;
             match ce.netout.write_all(&buf).await {
                 Ok(_) => (),
                 Err(e) => return Err(e)?,