Add more int variants

disk/src/binned.rs

@@ -9,7 +9,7 @@ use crate::binned::scalar::binned_stream;
 use crate::binnedstream::{BinnedScalarStreamFromPreBinnedPatches, BoxedStream};
 use crate::cache::{BinnedQuery, MergedFromRemotes};
 use crate::decode::{Endianness, EventValues};
-use crate::frame::makeframe::FrameType;
+use crate::frame::makeframe::{FrameType, SubFrId};
 use crate::raw::EventsQuery;
 use bytes::Bytes;
 use chrono::{TimeZone, Utc};
@@ -27,7 +27,6 @@ use serde::{Deserialize, Serialize, Serializer};
 use serde_json::Map;
 use std::future::Future;
 use std::marker::PhantomData;
-use std::ops::BitXor;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::time::Duration;
@@ -540,23 +539,8 @@ impl TBinnedBins for MinMaxAvgScalarBinBatch {
     }
 }
 
-// TODO
-
-// TODO
-
-// Write a function (as demo instead of a full Stream) which couples together the flow between the
-// event decoding and the event node processing.
-// That should require me to require Input/Output combinations in the StreamKind.
-
-// TODO try to get the binning generic over the actual numeric disk-dtype.
-// That should require me to make StreamKind generic over the numeric dtype.
-// I then need a away to compose the StreamKind:
-// Instead of writing some if-else-match-monster over all possible disk-dtype and AggKind combinations,
-// I would like to decide on the disk-dtype first and get some generic intermediate type, and the
-// decide the AggKind, and maybe even other generic types.
-
-pub trait NumOps: Sized + Copy + Send + Unpin + Zero + BitXor + AsPrimitive<f32> + Bounded + PartialOrd {}
-impl<T> NumOps for T where T: Sized + Copy + Send + Unpin + Zero + BitXor + AsPrimitive<f32> + Bounded + PartialOrd {}
+pub trait NumOps: Sized + Copy + Send + Unpin + Zero + AsPrimitive<f32> + Bounded + PartialOrd + SubFrId {}
+impl<T> NumOps for T where T: Sized + Copy + Send + Unpin + Zero + AsPrimitive<f32> + Bounded + PartialOrd + SubFrId {}
 
 pub trait EventsDecoder {
     type Output;

disk/src/decode.rs

@@ -24,18 +24,34 @@ pub trait NumFromBytes<NTY, END> {
     fn convert(buf: &[u8]) -> NTY;
 }
 
-impl NumFromBytes<i32, LittleEndian> for i32 {
-    fn convert(buf: &[u8]) -> i32 {
-        i32::from_le_bytes(*arrayref::array_ref![buf, 0, 4])
-    }
+macro_rules! impl_num_from_bytes_end {
+    ($nty:ident, $nl:expr, $end:ident, $ec:ident) => {
+        impl NumFromBytes<$nty, $end> for $nty {
+            fn convert(buf: &[u8]) -> $nty {
+                $nty::$ec(*arrayref::array_ref![buf, 0, $nl])
+            }
+        }
+    };
 }
 
-impl NumFromBytes<i32, BigEndian> for i32 {
-    fn convert(buf: &[u8]) -> i32 {
-        i32::from_be_bytes(*arrayref::array_ref![buf, 0, 4])
-    }
+macro_rules! impl_num_from_bytes {
+    ($nty:ident, $nl:expr) => {
+        impl_num_from_bytes_end!($nty, $nl, LittleEndian, from_le_bytes);
+        impl_num_from_bytes_end!($nty, $nl, BigEndian, from_be_bytes);
+    };
 }
 
+impl_num_from_bytes!(u8, 1);
+impl_num_from_bytes!(u16, 2);
+impl_num_from_bytes!(u32, 4);
+impl_num_from_bytes!(u64, 8);
+impl_num_from_bytes!(i8, 1);
+impl_num_from_bytes!(i16, 2);
+impl_num_from_bytes!(i32, 4);
+impl_num_from_bytes!(i64, 8);
+impl_num_from_bytes!(f32, 4);
+impl_num_from_bytes!(f64, 8);
+
 pub trait EventValueFromBytes<NTY, END>
 where
     NTY: NumFromBytes<NTY, END>,

disk/src/frame/makeframe.rs

@@ -29,8 +29,55 @@ impl FrameType for Result<StreamItem<RangeCompletableItem<MinMaxAvgScalarEventBa
     const FRAME_TYPE_ID: u32 = 0x11;
 }
 
-impl<NTY> FrameType for Result<StreamItem<RangeCompletableItem<MinMaxAvgScalarEventBatchGen<NTY>>>, Error> {
-    const FRAME_TYPE_ID: u32 = 888888;
+impl<NTY> FrameType for Result<StreamItem<RangeCompletableItem<MinMaxAvgScalarEventBatchGen<NTY>>>, Error>
+where
+    NTY: SubFrId,
+{
+    const FRAME_TYPE_ID: u32 = 0x28c4a100 + NTY::SUB;
+}
+
+pub trait SubFrId {
+    const SUB: u32;
+}
+
+impl SubFrId for u8 {
+    const SUB: u32 = 3;
+}
+
+impl SubFrId for u16 {
+    const SUB: u32 = 5;
+}
+
+impl SubFrId for u32 {
+    const SUB: u32 = 8;
+}
+
+impl SubFrId for u64 {
+    const SUB: u32 = 10;
+}
+
+impl SubFrId for i8 {
+    const SUB: u32 = 2;
+}
+
+impl SubFrId for i16 {
+    const SUB: u32 = 4;
+}
+
+impl SubFrId for i32 {
+    const SUB: u32 = 7;
+}
+
+impl SubFrId for i64 {
+    const SUB: u32 = 9;
+}
+
+impl SubFrId for f32 {
+    const SUB: u32 = 11;
+}
+
+impl SubFrId for f64 {
+    const SUB: u32 = 12;
 }
 
 pub trait ProvidesFrameType {

disk/src/raw/conn.rs

@@ -189,8 +189,16 @@ macro_rules! pipe2 {
 macro_rules! pipe1 {
     ($nty:expr, $end:expr, $shape:expr, $agg_kind:expr, $event_blobs:expr) => {
         match $nty {
+            ScalarType::U8 => pipe2!(u8, $end, $shape, $agg_kind, $event_blobs),
+            ScalarType::U16 => pipe2!(u16, $end, $shape, $agg_kind, $event_blobs),
+            ScalarType::U32 => pipe2!(u32, $end, $shape, $agg_kind, $event_blobs),
+            ScalarType::U64 => pipe2!(u64, $end, $shape, $agg_kind, $event_blobs),
+            ScalarType::I8 => pipe2!(i8, $end, $shape, $agg_kind, $event_blobs),
+            ScalarType::I16 => pipe2!(i16, $end, $shape, $agg_kind, $event_blobs),
             ScalarType::I32 => pipe2!(i32, $end, $shape, $agg_kind, $event_blobs),
-            _ => err::todoval(),
+            ScalarType::I64 => pipe2!(i64, $end, $shape, $agg_kind, $event_blobs),
+            ScalarType::F32 => pipe2!(f32, $end, $shape, $agg_kind, $event_blobs),
+            ScalarType::F64 => pipe2!(f64, $end, $shape, $agg_kind, $event_blobs),
         }
     };
 }