Does not compile, type issues

disk/src/agg/binnedt.rs

@@ -9,7 +9,7 @@ use std::collections::VecDeque;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-pub trait AggregatorTdim {
+pub trait AggregatorTdim: Sized + Unpin {
     type InputValue;
     type OutputValue: AggregatableXdim1Bin + AggregatableTdim + Unpin;
     fn ends_before(&self, inp: &Self::InputValue) -> bool;

disk/src/binned.rs

@@ -1,9 +1,10 @@
 use crate::agg::binnedt::{AggregatableTdim, AggregatorTdim, IntoBinnedT};
 use crate::agg::scalarbinbatch::{MinMaxAvgScalarBinBatch, MinMaxAvgScalarBinBatchAggregator};
 use crate::agg::streams::{Collectable, Collected, StreamItem, ToJsonResult};
-use crate::agg::AggregatableXdim1Bin;
+use crate::agg::{AggregatableXdim1Bin, FitsInside};
 use crate::binned::scalar::{adapter_to_stream_item, binned_stream};
 use crate::binnedstream::{BinnedScalarStreamFromPreBinnedPatches, BinnedStream};
+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;
@@ -14,7 +15,9 @@ use err::Error;
 use futures_core::Stream;
 use futures_util::StreamExt;
 use netpod::log::*;
-use netpod::{AggKind, BinnedRange, NodeConfigCached, PerfOpts, PreBinnedPatchIterator, PreBinnedPatchRange};
+use netpod::{
+    AggKind, BinnedRange, NanoRange, NodeConfigCached, PerfOpts, PreBinnedPatchIterator, PreBinnedPatchRange,
+};
 use num_traits::Zero;
 use serde::{Deserialize, Serialize, Serializer};
 use std::pin::Pin;
@@ -391,10 +394,35 @@ pub async fn binned_json(node_config: &NodeConfigCached, query: &BinnedQuery) ->
     Ok(serde_json::to_value(ret)?)
 }
 
+pub trait PreBinnedItem: Unpin {
+    type BinnedStreamItem: AggregatableTdim + Unpin + Send;
+    fn into_binned_stream_item(self, fit_range: NanoRange) -> Option<Self::BinnedStreamItem>;
+}
+
+impl PreBinnedItem for PreBinnedScalarItem {
+    type BinnedStreamItem = BinnedScalarStreamItem;
+
+    fn into_binned_stream_item(self, fit_range: NanoRange) -> Option<Self::BinnedStreamItem> {
+        match self {
+            Self::RangeComplete => Some(Self::BinnedStreamItem::RangeComplete),
+            Self::Batch(item) => {
+                use super::agg::{Fits, FitsInside};
+                match item.fits_inside(fit_range) {
+                    Fits::Inside | Fits::PartlyGreater | Fits::PartlyLower | Fits::PartlyLowerAndGreater => {
+                        Some(Self::BinnedStreamItem::Values(item))
+                    }
+                    _ => None,
+                }
+            }
+        }
+    }
+}
+
 pub trait BinnedStreamKind: Clone + Unpin + Send + Sync + 'static {
     type BinnedStreamItem: MakeBytesFrame;
     type BinnedStreamType: Stream + Send + 'static;
     type Dummy: Default + Unpin + Send;
+    type PreBinnedItem: PreBinnedItem + Send;
 
     fn new_binned_from_prebinned(
         &self,
@@ -434,6 +462,7 @@ impl BinnedStreamKind for BinnedStreamKindScalar {
     type BinnedStreamItem = Result<StreamItem<BinnedScalarStreamItem>, Error>;
     type BinnedStreamType = BinnedStream<Self::BinnedStreamItem>;
     type Dummy = u32;
+    type PreBinnedItem = PreBinnedScalarItem;
 
     fn new_binned_from_prebinned(
         &self,

disk/src/binnedstream.rs

@@ -1,6 +1,6 @@
 use crate::agg::binnedt::IntoBinnedT;
 use crate::agg::streams::StreamItem;
-use crate::binned::{BinnedScalarStreamItem, BinnedStreamKind};
+use crate::binned::{BinnedScalarStreamItem, BinnedStreamKind, PreBinnedItem};
 use crate::cache::pbvfs::{PreBinnedScalarItem, PreBinnedScalarValueFetchedStream};
 use crate::cache::{CacheUsage, PreBinnedQuery};
 use err::Error;
@@ -16,8 +16,18 @@ pub struct BinnedScalarStreamFromPreBinnedPatches<BK>
 where
     BK: BinnedStreamKind,
 {
-    inp: Pin<Box<dyn Stream<Item = Result<StreamItem<BinnedScalarStreamItem>, Error>> + Send>>,
+    //inp: Pin<Box<dyn Stream<Item = Result<StreamItem<BinnedScalarStreamItem>, Error>> + Send>>,
-    _marker: BK::Dummy,
+    inp: Pin<
+        Box<
+            dyn Stream<
+                    Item = Result<
+                        StreamItem<<<BK as BinnedStreamKind>::PreBinnedItem as PreBinnedItem>::BinnedStreamItem>,
+                        Error,
+                    >,
+                > + Send,
+        >,
+    >,
+    stream_kind: BK,
 }
 
 impl<BK> BinnedScalarStreamFromPreBinnedPatches<BK>
@@ -75,34 +85,24 @@ where
                         Ok(item) => match item {
                             StreamItem::Log(item) => Some(Ok(StreamItem::Log(item))),
                             StreamItem::Stats(item) => Some(Ok(StreamItem::Stats(item))),
-                            StreamItem::DataItem(item) => match item {
+                            StreamItem::DataItem(item) => {
-                                PreBinnedScalarItem::RangeComplete => {
+                                match crate::binned::PreBinnedItem::into_binned_stream_item(item, fit_range) {
-                                    Some(Ok(StreamItem::DataItem(BinnedScalarStreamItem::RangeComplete)))
+                                    Some(item) => Some(Ok(StreamItem::DataItem(item))),
+                                    None => None,
                                 }
-                                PreBinnedScalarItem::Batch(item) => {
+                            }
-                                    use super::agg::{Fits, FitsInside};
-                                    match item.fits_inside(fit_range) {
-                                        Fits::Inside
-                                        | Fits::PartlyGreater
-                                        | Fits::PartlyLower
-                                        | Fits::PartlyLowerAndGreater => {
-                                            Some(Ok(StreamItem::DataItem(BinnedScalarStreamItem::Values(item))))
-                                        }
-                                        _ => None,
-                                    }
-                                }
-                            },
                         },
                         Err(e) => Some(Err(e)),
                     };
                     ready(g)
                 }
-            })
+            });
-            .into_binned_t(range);
+        // TODO activate the T-binning via the bin-to-bin binning trait.
-        let mm = BK::Dummy::default();
+        err::todo();
+        //let inp = IntoBinnedT::into_binned_t(inp, range);
         Ok(Self {
             inp: Box::pin(inp),
-            _marker: mm,
+            stream_kind,
         })
     }
 }
@@ -111,7 +111,7 @@ impl<BK> Stream for BinnedScalarStreamFromPreBinnedPatches<BK>
 where
     BK: BinnedStreamKind,
 {
-    type Item = Result<StreamItem<BinnedScalarStreamItem>, Error>;
+    type Item = Result<StreamItem<<<BK as BinnedStreamKind>::PreBinnedItem as PreBinnedItem>::BinnedStreamItem>, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;

disk/src/cache/pbvfs.rs

@@ -62,7 +62,7 @@ impl<BK> Stream for PreBinnedScalarValueFetchedStream<BK>
 where
     BK: BinnedStreamKind,
 {
-    type Item = Result<StreamItem<PreBinnedScalarItem>, Error>;
+    type Item = Result<StreamItem<BK::PreBinnedItem>, Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         use Poll::*;