flush prev

op-rs · Sep 30, 2024 · 17b44df · 17b44df
1 parent 3ceade2
commit 17b44df
Showing 1 changed file with 2 additions and 5 deletions.
diff --git a/crates/derive/src/stages/batch_stream.rs b/crates/derive/src/stages/batch_stream.rs
@@ -107,6 +107,7 @@ where
 {
     fn flush(&mut self) {
         if self.is_active().unwrap_or(false) {
+            self.prev.flush();
             self.span = None;
             self.buffer.clear();
         }
@@ -150,9 +151,6 @@ where
                             // Flush the stage.
                             self.flush();
 
-                            // Send a signal to drop the channel that the batch was derived from.
-                            // TODO
-
                             return Err(PipelineError::Eof.temp());
                         }
                         BatchValidity::Undecided | BatchValidity::Future => {
@@ -208,8 +206,7 @@ mod test {
     use super::*;
     use crate::{
         batch::{SingleBatch, SpanBatchElement},
-        stages::test_utils::{CollectingLayer, MockBatchStreamProvider, TraceStorage},
-        traits::test_utils::TestL2ChainProvider,
+        stages::test_utils::{CollectingLayer, MockBatchStreamProvider, TraceStorage}, traits::test_utils::TestL2ChainProvider,
     };
     use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};