Update DetectionBatchStream to wrap batch_rx directly

Signed-off-by: declark1 <44146800+declark1@users.noreply.github.com>

src/orchestrator/types/detection_batch_stream.rs

@@ -1,14 +1,13 @@
 use futures::{stream, Stream, StreamExt};
 use tokio::sync::mpsc;
-use tokio_stream::wrappers::ReceiverStream;
 
-use super::{BoxStream, Chunk, DetectionBatcher, DetectionStream, Detections, DetectorId, InputId};
+use super::{Chunk, DetectionBatcher, DetectionStream, Detections, DetectorId, InputId};
 use crate::orchestrator::Error;
 
 /// Wraps detection streams and produces a stream
 /// of batches using a [`DetectionBatcher`].
 pub struct DetectionBatchStream<B: DetectionBatcher> {
-    inner: BoxStream<Result<B::Batch, Error>>,
+    batch_rx: mpsc::Receiver<Result<B::Batch, Error>>,
 }
 
 impl<B> DetectionBatchStream<B>
@@ -71,9 +70,7 @@ where
             }
         });
 
-        Self {
-            inner: ReceiverStream::new(batch_rx).boxed(),
-        }
+        Self { batch_rx }
     }
 }
 
@@ -84,6 +81,6 @@ impl<T: DetectionBatcher> Stream for DetectionBatchStream<T> {
         mut self: std::pin::Pin<&mut Self>,
         cx: &mut std::task::Context<'_>,
     ) -> std::task::Poll<Option<Self::Item>> {
-        self.inner.as_mut().poll_next(cx)
+        self.batch_rx.poll_recv(cx)
     }
 }