Limit response size in a somewhat improved way

crates/streams/src/collect.rs

@@ -44,6 +44,7 @@ macro_rules! trace4 {
 pub struct Collect {
     inp: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Collectable>>> + Send>>,
     events_max: u64,
+    bytes_max: u64,
     range: Option<SeriesRange>,
     binrange: Option<BinnedRangeEnum>,
     collector: Option<Box<dyn Collector>>,
@@ -58,6 +59,7 @@ impl Collect {
         inp: Pin<Box<dyn Stream<Item = Sitemty<Box<dyn Collectable>>> + Send>>,
         deadline: Instant,
         events_max: u64,
+        bytes_max: u64,
         range: Option<SeriesRange>,
         binrange: Option<BinnedRangeEnum>,
     ) -> Self {
@@ -65,6 +67,7 @@ impl Collect {
         Self {
             inp,
             events_max,
+            bytes_max,
             range,
             binrange,
             collector: None,
@@ -93,7 +96,12 @@ impl Collect {
                         let coll = self.collector.get_or_insert_with(|| item.new_collector());
                         coll.ingest(&mut item);
                         if coll.len() as u64 >= self.events_max {
-                            info!("reached events_max {}", self.events_max);
+                            info!("reached events_max {} / {}", coll.len(), self.events_max);
+                            coll.set_continue_at_here();
+                            self.done_input = true;
+                        }
+                        if coll.byte_estimate() >= self.bytes_max {
+                            info!("reached bytes_max {} / {}", coll.byte_estimate(), self.events_max);
                             coll.set_continue_at_here();
                             self.done_input = true;
                         }

crates/streams/src/plaineventsjson.rs

@@ -37,7 +37,15 @@ pub async fn plain_events_json(
     //let stream = TimeBinnableToCollectable::new(stream);
     let stream = Box::pin(stream);
     info!("plain_events_json  boxed stream created");
-    let collected = Collect::new(stream, deadline, evq.events_max(), Some(evq.range().clone()), None).await?;
+    let collected = Collect::new(
+        stream,
+        deadline,
+        evq.events_max(),
+        evq.bytes_max(),
+        Some(evq.range().clone()),
+        None,
+    )
+    .await?;
     info!("plain_events_json  collected");
     let jsval = serde_json::to_value(&collected)?;
     info!("plain_events_json  json serialized");

crates/streams/src/test/collect.rs

@@ -63,11 +63,12 @@ fn collect_channel_events_01() -> Result<(), Error> {
         // TODO build like in request code
         let deadline = Instant::now() + Duration::from_millis(4000);
         let events_max = 10000;
+        let bytes_max = 80 * 10000;
         let stream = PlainEventStream::new(stream);
         let stream = EventsToTimeBinnable::new(stream);
         let stream = TimeBinnableToCollectable::new(stream);
         let stream = Box::pin(stream);
-        let res = Collect::new(stream, deadline, events_max, None, None).await?;
+        let res = Collect::new(stream, deadline, events_max, bytes_max, None, None).await?;
         if let Some(res) = res.as_any_ref().downcast_ref::<EventsDim0CollectorOutput<f32>>() {
             eprintln!("Great, a match");
             eprintln!("{res:?}");
@@ -103,11 +104,12 @@ fn collect_channel_events_pulse_id_diff() -> Result<(), Error> {
         let stream = EventsToTimeBinnable::new(stream);
         let deadline = Instant::now() + Duration::from_millis(4000);
         let events_max = 10000;
+        let bytes_max = 80 * 10000;
         let stream = Box::pin(stream);
         let stream = build_time_binning_transform(&trqu, stream)?;
         let stream = TimeBinnableToCollectable::new(stream);
         let stream = Box::pin(stream);
-        let res = Collect::new(stream, deadline, events_max, None, None).await?;
+        let res = Collect::new(stream, deadline, events_max, bytes_max, None, None).await?;
         if let Some(res) = res.as_any_ref().downcast_ref::<EventsDim0CollectorOutput<i64>>() {
             eprintln!("Great, a match");
             eprintln!("{res:?}");

crates/streams/src/timebinnedjson.rs

@@ -256,10 +256,12 @@ pub async fn timebinned_json(
 ) -> Result<JsonValue, Error> {
     let deadline = Instant::now().checked_add(query.timeout_value()).unwrap();
     let binned_range = BinnedRangeEnum::covering_range(query.range().clone(), query.bin_count())?;
+    // TODO derive better values, from query
     let collect_max = 10000;
+    let bytes_max = 100 * collect_max;
     let stream = timebinned_stream(query.clone(), binned_range.clone(), ch_conf, ctx, open_bytes).await?;
     let stream = timebinned_to_collectable(stream);
-    let collected = Collect::new(stream, deadline, collect_max, None, Some(binned_range));
+    let collected = Collect::new(stream, deadline, collect_max, bytes_max, None, Some(binned_range));
     let collected: BoxFuture<_> = Box::pin(collected);
     let collected = collected.await?;
     let jsval = serde_json::to_value(&collected)?;