From e216465a023cd4a861a9833fbecbee0d434cdf17 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 12 Nov 2026 09:32:07 +0000 Subject: [PATCH] Add terminate_stream_to for receivers sent data by unicast terminate_stream only sends the termination packets to the universe's multicast group, so a receiver that was sent the data by unicast and hasn't joined the group never sees them and only drops the source after the 2.5 s data loss timeout. terminate_stream_to sends the termination packets to each given unicast destination, then deregisters the universe. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KzaQVPkrkczgX2g7wGkip5 --- src/source.rs | 71 ++++++++++++++++++++++++++++++++++++++++++++++ tests/src_tests.rs | 31 ++++++++++++++++++++ 2 files changed, 102 insertions(+) diff --git a/src/source.rs b/src/source.rs index f701751..c5d26c4 100644 --- a/src/source.rs +++ b/src/source.rs @@ -364,6 +364,42 @@ impl SacnSource { unlock_internal_mut(&mut self.internal)?.terminate_stream(universe, start_code) } + /// Terminates sending on the given universe, sending the termination packets to the given unicast destinations. + /// + /// `terminate_stream` sends the termination packets only to the universe's multicast group, so a receiver that was + /// sent the universe's data by unicast and hasn't joined that group never sees them. It then only notices the + /// source has gone after the network data loss timeout. Pass the same destinations that were given to `send` here + /// so those receivers are told straight away. + /// + /// The `E131_TERMINATE_STREAM_PACKET_COUNT` termination packets are sent to each destination in turn, then the + /// universe is deregistered. If `dst_ips` is empty this behaves exactly like `terminate_stream` (multicast). + /// + /// # Arguments + /// universe: The universe that is being terminated. + /// + /// `start_code`: used for the first byte of the otherwise empty data payload to indicate the `start_code` of the data. + /// + /// `dst_ips`: The unicast destinations to send the termination packets to. + /// + /// # Errors: + /// `IllegalUniverse`: Returned if the universe is outwith the allowed range of sACN universes as defined in ANSI E1.31-2018 Section 6.2.7. + /// + /// `UniverseNotRegistered`: Returned if the universe is not registered on this source. + /// + /// Io: Returned if the termination packets fail to be sent on the socket. The packets are still sent to the + /// remaining destinations and the universe is still deregistered; the first error is returned. + /// + /// `SourceCorrupt`: Returned if the Mutex used to control access to the internal sender is poisoned by a thread encountering + /// a panic while accessing causing the source to be left in a potentially inconsistent state. + pub fn terminate_stream_to( + &mut self, + universe: u16, + start_code: u8, + dst_ips: &[SocketAddr], + ) -> Result<()> { + unlock_internal_mut(&mut self.internal)?.terminate_stream_to(universe, start_code, dst_ips) + } + /// Returns the ACN CID device identifier of the `SacnSourceInternal`. /// /// # Errors @@ -1026,6 +1062,41 @@ impl SacnSourceInternal { Ok(()) } + /// Terminates a universe stream, sending the termination packets to each of the given unicast destinations. + /// + /// See (`SacnSource::terminate_stream_to`)[`SacnSource::terminate_stream_to`]. + fn terminate_stream_to( + &mut self, + universe: u16, + start_code: u8, + dst_ips: &[SocketAddr], + ) -> Result<()> { + if dst_ips.is_empty() { + return self.terminate_stream(universe, start_code); + } + + // Check before sending anything so that an unregistered universe sends nothing. + self.universe_allowed(&universe)?; + + // Keep going after a failed send so one unreachable destination doesn't stop the others being told. + let mut first_err: Option = None; + for dst in dst_ips { + for _ in 0..E131_TERMINATE_STREAM_PACKET_COUNT { + if let Err(e) = self.send_terminate_stream_pkt(universe, Some(*dst), start_code) { + first_err.get_or_insert(e); + break; + } + } + } + + self.deregister_universe(universe)?; + + match first_err { + Some(e) => Err(e), + None => Ok(()), + } + } + /// Terminates the DMX source. /// /// This includes terminating each registered universe with the `start_code` given. diff --git a/tests/src_tests.rs b/tests/src_tests.rs index ee9605f..e711e63 100644 --- a/tests/src_tests.rs +++ b/tests/src_tests.rs @@ -702,3 +702,34 @@ const TEST_DATA_MULTIPLE_UNIVERSE: [u8; 712] = [ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 100, ]; + +/// A receiver that is sent a universe by unicast, and hasn't joined that universe's multicast group, must still be +/// told when the source terminates the universe. +#[test] +#[serial] +fn test_terminate_stream_to_reaches_unicast_receiver() { + use sacn::receive::SacnReceiver; + use std::time::Duration; + + let rcv_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); + let mut rcv = SacnReceiver::with_ip(rcv_addr, None).unwrap(); + // Unicast only: don't join the multicast group for universe 1. + rcv.set_is_multicast_enabled(false).unwrap(); + rcv.listen_universes(&[1]).unwrap(); + rcv.set_announce_stream_termination(true); + + let src_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0); + let mut src = SacnSource::with_ip("Unicast source", src_addr).unwrap(); + src.register_universe(1).unwrap(); + src.send(&[1], &[0, 1, 2], None, Some(rcv_addr), None).unwrap(); + assert_eq!(rcv.recv(Some(Duration::from_secs(2))).unwrap()[0].universe, 1); + + src.terminate_stream_to(1, 0, &[rcv_addr]).unwrap(); + match rcv.recv(Some(Duration::from_secs(2))) { + Err(SacnError::UniverseTerminated(cid, 1)) => assert_eq!(cid, src.cid().unwrap()), + other => panic!("expected UniverseTerminated for universe 1, got {other:?}"), + } + + // The universe is deregistered afterwards, as with terminate_stream. + assert!(!src.universes().unwrap().contains(&1)); +}