bug with disabling streaming fixed

cameleon-rs · Nov 21, 2024 · 62efdb2 · 62efdb2
1 parent ad73d09
commit 62efdb2
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 8 deletions.
diff --git a/cameleon/src/gige/control_handle.rs b/cameleon/src/gige/control_handle.rs
@@ -6,14 +6,18 @@ use std::{
     convert::TryInto,
     io::Read,
     sync::{Arc, Mutex},
-    thread, time,
+    thread,
+    time::{self},
 };
 
 use async_std::{channel, future, net::UdpSocket, task};
 use futures_channel::oneshot;
 use futures_util::{select, FutureExt};
 
-use cameleon_device::gige::protocol::{ack, cmd};
+use cameleon_device::gige::{
+    protocol::{ack, cmd},
+    register_map::StreamChannelPort,
+};
 
 use crate::{
     genapi::CompressionType, utils::unzip_genxml, ControlError, ControlResult, DeviceControl,
@@ -565,8 +569,8 @@ impl DeviceControl for ControlHandleInner {
     }
 
     fn disable_streaming(&mut self) -> ControlResult<()> {
-        let port = StreamRegister::new(0).channel_port(self)?;
-        StreamRegister::new(0).set_channel_port(self, port.set_host_port(0))?;
+        StreamRegister::new(0)
+            .set_channel_port(self, StreamChannelPort::from_raw(0).set_host_port(0))?;
         Ok(())
     }
 }

diff --git a/cameleon/src/gige/stream_handle.rs b/cameleon/src/gige/stream_handle.rs
@@ -172,10 +172,6 @@ impl StreamingLoop {
         }
 
         loop {
-            if self.cancellation_rx.try_recv().is_ok() {
-                *self.completion.0.lock().unwrap() = true;
-                self.completion.1.notify_all();
-            }
             let length = self.sock.recv(&mut self.buffer).unwrap();
             let mut cursor = Cursor::new(&self.buffer[..]);
             let header = unwrap_or_continue!(PacketHeader::parse(&mut cursor));
@@ -195,6 +191,14 @@ impl StreamingLoop {
                     builder = Some(PayloadBuilder::new(header, leader, &mut payload));
                 }
                 PacketType::Trailer => {
+                    match self.cancellation_rx.try_recv() {
+                        Ok(Some(())) | Err(_) => {
+                            *self.completion.0.lock().unwrap() = true;
+                            self.completion.1.notify_all();
+                            break;
+                        }
+                        Ok(None) => {}
+                    }
                     let payload_type =
                         unwrap_or_continue!(PayloadType::parse_generic_leader(&mut cursor));
                     let Some(builder) = builder.take() else {