WIP

disk/src/binned.rs

@@ -17,12 +17,15 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::time::Duration;
 
-pub struct BinnedStreamRes {
-    pub binned_stream: BinnedStream,
+pub struct BinnedStreamRes<I> {
+    pub binned_stream: BinnedStream<I>,
     pub range: BinnedRange,
 }
 
-pub async fn binned_stream(node_config: &NodeConfigCached, query: &BinnedQuery) -> Result<BinnedStreamRes, Error> {
+pub async fn binned_stream(
+    node_config: &NodeConfigCached,
+    query: &BinnedQuery,
+) -> Result<BinnedStreamRes<Result<MinMaxAvgScalarBinBatchStreamItem, Error>>, Error> {
     if query.channel().backend != node_config.node.backend {
         let err = Error::with_msg(format!(
             "backend mismatch  node: {}  requested: {}",
@@ -96,6 +99,10 @@ pub async fn binned_bytes_for_http(
     node_config: &NodeConfigCached,
     query: &BinnedQuery,
 ) -> Result<BinnedStreamBox, Error> {
+    // TODO must decide here already which AggKind so that I can call into the generic code.
+
+    todo::todo;
+
     let res = binned_stream(node_config, query).await?;
     let ret = BinnedBytesForHttpStream::new(res.binned_stream);
     Ok(Box::pin(ret))

disk/src/binnedstream.rs

@@ -95,21 +95,20 @@ impl Stream for BinnedStreamFromPreBinnedPatches {
     }
 }
 
-pub struct BinnedStream {
-    inp: Pin<Box<dyn Stream<Item = Result<MinMaxAvgScalarBinBatchStreamItem, Error>> + Send>>,
+pub struct BinnedStream<I> {
+    inp: Pin<Box<dyn Stream<Item = I> + Send>>,
 }
 
-impl BinnedStream {
-    pub fn new(
-        inp: Pin<Box<dyn Stream<Item = Result<MinMaxAvgScalarBinBatchStreamItem, Error>> + Send>>,
-    ) -> Result<Self, Error> {
+impl<I> BinnedStream<I> {
+    // Item was: Result<MinMaxAvgScalarBinBatchStreamItem, Error>
+    pub fn new(inp: Pin<Box<dyn Stream<Item = I> + Send>>) -> Result<Self, Error> {
         Ok(Self { inp })
     }
 }
 
-impl Stream for BinnedStream {
-    // TODO make this generic over all possible things
-    type Item = Result<MinMaxAvgScalarBinBatchStreamItem, Error>;
+impl<I> Stream for BinnedStream<I> {
+    //type Item = Result<MinMaxAvgScalarBinBatchStreamItem, Error>;
+    type Item = I;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
         self.inp.poll_next_unpin(cx)