diff --git a/src/receive.rs b/src/receive.rs index 15f4141..fede53b 100644 --- a/src/receive.rs +++ b/src/receive.rs @@ -197,6 +197,10 @@ pub struct SacnReceiver { /// Flag which indicates if an `UniverseTimeout` error should be thrown if it is detected that a source has timed out. announce_timeout: bool, + + /// When the last synchronisation packet was received for each synchronisation address. + /// Used to tell if synchronisation on an address has been lost, as per ANSI E1.31-2016 Section 11.1.2. + last_sync_packets: HashMap, } /// Represents an sACN source/sender on the network that has been discovered by this sACN receiver by receiving universe discovery packets. @@ -306,6 +310,7 @@ impl SacnReceiver { announce_source_discovery: ANNOUNCE_SOURCE_DISCOVERY_DEFAULT, announce_stream_termination: ANNOUNCE_STREAM_TERMINATION_DEFAULT, announce_timeout: ANNOUNCE_TIMEOUT_DEFAULT, + last_sync_packets: HashMap::new(), }; sri.listen_universes(&[E131_DISCOVERY_UNIVERSE])?; @@ -736,7 +741,19 @@ impl SacnReceiver { self.announce_timeout, )?; - if data_pkt.synchronization_address == E131_NO_SYNC_ADDR { + // As per ANSI E1.31-2016 Section 6.2.6 and 11.1.2, if the Force_Synchronization option is set and synchronisation on this + // packet's synchronisation address has been lost (or has not started), the data doesn't have to wait for a synchronisation + // packet. Without the option the data is held until synchronisation resumes. + let process_unsynchronized = data_pkt.synchronization_address != E131_NO_SYNC_ADDR + && data_pkt.force_synchronization + && !self.is_synchronized(data_pkt.synchronization_address); + + if process_unsynchronized { + // Keep listening on the synchronisation address so synchronisation can resume. + self.listen_universes(&[data_pkt.synchronization_address])?; + } + + if data_pkt.synchronization_address == E131_NO_SYNC_ADDR || process_unsynchronized { self.clear_waiting_data(data_pkt.universe); let vals: Vec = data_pkt.data.property_values.into_owned(); @@ -859,6 +876,9 @@ impl SacnReceiver { self.announce_timeout, )?; + self.last_sync_packets + .insert(sync_pkt.synchronization_address, Instant::now()); + let res = self.rtrv_waiting_data(sync_pkt.synchronization_address); if res.is_empty() { Ok(None) @@ -974,6 +994,14 @@ impl SacnReceiver { /// Goes through all the waiting data and removes any which has timed out as a sync-packet for it hasn't been received within the `E131_NETWORK_DATA_LOSS_TIMEOUT` /// period as specified by ANSI E1.31-2018 Section 11.1.2. + /// Returns true if a synchronisation packet has been received on the given synchronisation address within + /// `E131_NETWORK_DATA_LOSS_TIMEOUT`, meaning synchronisation on that address is active as per ANSI E1.31-2016 Section 11.1. + fn is_synchronized(&self, sync_addr: u16) -> bool { + self.last_sync_packets + .get(&sync_addr) + .is_some_and(|t| t.elapsed() < E131_NETWORK_DATA_LOSS_TIMEOUT) + } + fn check_waiting_data_timeouts(&mut self) { self.waiting_data .retain(|_uni, data| data.recv_timestamp.elapsed() < E131_NETWORK_DATA_LOSS_TIMEOUT); @@ -2192,6 +2220,103 @@ mod test { assert_eq!(dmx_rcv.discovered_sources[0].pages[0].universes, universes); } + fn synchronized_data_packet<'a>( + universe: u16, + sync_address: u16, + sequence_number: u8, + force_synchronization: bool, + ) -> DataPacketFramingLayer<'a> { + DataPacketFramingLayer { + synchronization_address: sync_address, + force_synchronization, + ..generate_data_packet_framing_layer_seq_num(universe, sequence_number) + } + } + + /// With `Force_Synchronization` set and no synchronisation packet ever received on the address, data isn't held. + #[test] + fn test_force_synchronization_before_sync_starts() { + const UNIVERSE: u16 = 1; + const SYNC_UNI: u16 = 2; + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); + let mut dmx_rcv = SacnReceiver::with_ip(addr, None).unwrap(); + dmx_rcv.listen_universes(&[UNIVERSE]).unwrap(); + let cid = Uuid::new_v4(); + + let res = dmx_rcv + .handle_data_packet(cid, synchronized_data_packet(UNIVERSE, SYNC_UNI, 0, true)) + .unwrap() + .expect("data with Force_Synchronization should be passed up when not synchronised"); + assert_eq!(res[0].universe, UNIVERSE); + assert!(dmx_rcv.waiting_data.is_empty()); + // The receiver still listens on the synchronisation address so synchronisation can start. + assert!(dmx_rcv.is_listening(&SYNC_UNI)); + } + + /// While synchronisation is active, data is held until the synchronisation packet even with `Force_Synchronization` set. + #[test] + fn test_force_synchronization_while_synchronized() { + const UNIVERSE: u16 = 1; + const SYNC_UNI: u16 = 2; + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); + let mut dmx_rcv = SacnReceiver::with_ip(addr, None).unwrap(); + dmx_rcv.listen_universes(&[UNIVERSE, SYNC_UNI]).unwrap(); + let cid = Uuid::new_v4(); + + // Synchronisation starts with the first synchronisation packet. + assert!(dmx_rcv + .handle_sync_packet(cid, generate_sync_packet_framing_layer_seq_num(SYNC_UNI, 0)) + .unwrap() + .is_none()); + + assert!(dmx_rcv + .handle_data_packet(cid, synchronized_data_packet(UNIVERSE, SYNC_UNI, 0, true)) + .unwrap() + .is_none()); + + let res = dmx_rcv + .handle_sync_packet(cid, generate_sync_packet_framing_layer_seq_num(SYNC_UNI, 1)) + .unwrap() + .expect("held data should be released by the synchronisation packet"); + assert_eq!(res[0].universe, UNIVERSE); + } + + /// Once synchronisation is lost, `Force_Synchronization` set means data is processed without waiting, and unset means the + /// data is held until synchronisation resumes. + #[test] + fn test_force_synchronization_after_sync_lost() { + const UNIVERSE: u16 = 1; + const SYNC_UNI: u16 = 2; + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); + let mut dmx_rcv = SacnReceiver::with_ip(addr, None).unwrap(); + dmx_rcv.listen_universes(&[UNIVERSE, SYNC_UNI]).unwrap(); + let cid = Uuid::new_v4(); + + dmx_rcv + .handle_sync_packet(cid, generate_sync_packet_framing_layer_seq_num(SYNC_UNI, 0)) + .unwrap(); + // No synchronisation packet for longer than the network data loss timeout. + dmx_rcv.last_sync_packets.insert( + SYNC_UNI, + Instant::now() - E131_NETWORK_DATA_LOSS_TIMEOUT - Duration::from_millis(100), + ); + + // Force_Synchronization unset: hold the data. + assert!(dmx_rcv + .handle_data_packet(cid, synchronized_data_packet(UNIVERSE, SYNC_UNI, 0, false)) + .unwrap() + .is_none()); + assert!(dmx_rcv.waiting_data.contains_key(&UNIVERSE)); + + // Force_Synchronization set: process the data now, replacing the held data. + let res = dmx_rcv + .handle_data_packet(cid, synchronized_data_packet(UNIVERSE, SYNC_UNI, 1, true)) + .unwrap() + .expect("data with Force_Synchronization should be passed up once synchronisation is lost"); + assert_eq!(res[0].universe, UNIVERSE); + assert!(dmx_rcv.waiting_data.is_empty()); + } + #[test] fn test_handle_multi_page_discovery_packet() { let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); diff --git a/src/source.rs b/src/source.rs index ccb346d..cd52e84 100644 --- a/src/source.rs +++ b/src/source.rs @@ -108,6 +108,11 @@ struct SacnSourceInternal { /// upon in an untested environment. preview_data: bool, + /// Flag which is included in sACN data packets that have a synchronisation address. It tells receivers what to do if + /// synchronisation is lost, as per ANSI E1.31-2016 Section 6.2.6 (`Force_Synchronization`). If false receivers hold the + /// data until synchronisation resumes, if true they may process the data without waiting for a synchronisation packet. + force_synchronization: bool, + /// The sequence numbers used for data packets, keeps a reference of the next sequence number to use for each universe. /// Sequence numbers are always in the range [0, 255]. data_sequences: RefCell>, @@ -434,6 +439,32 @@ impl SacnSource { Ok(()) } + /// Returns the value of the `Force_Synchronization` option set on data packets sent by this `SacnSource`. + /// + /// For details see (`set_force_synchronization`)[`set_force_synchronization`]. + /// + /// # Errors + /// `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 force_synchronization(&self) -> Result { + Ok(unlock_internal(&self.internal)?.force_synchronization) + } + + /// Sets the `Force_Synchronization` option on data packets sent by this `SacnSource`, as per ANSI E1.31-2016 Section 6.2.6. + /// + /// The option tells receivers what to do with synchronised data (data sent with a synchronisation address) if synchronisation is lost, + /// meaning no synchronisation packet has been received on that address within `E131_NETWORK_DATA_LOSS_TIMEOUT`: + /// - false (the default): receivers hold the data until synchronisation resumes. + /// - true: receivers may process the data straight away without waiting for a synchronisation packet. + /// + /// # Errors + /// `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 set_force_synchronization(&mut self, force_synchronization: bool) -> Result<()> { + unlock_internal_mut(&mut self.internal)?.force_synchronization = force_synchronization; + Ok(()) + } + /// Sets the `is_sending_discovery` flag to the given value. /// /// # Arguments @@ -588,6 +619,7 @@ impl SacnSourceInternal { cid, name: name.to_string(), preview_data: false, + force_synchronization: false, data_sequences: RefCell::new(HashMap::new()), sync_sequences: RefCell::new(HashMap::new()), universes: Vec::new(), @@ -833,7 +865,7 @@ impl SacnSourceInternal { sequence_number: sequence, preview_data: self.preview_data, stream_terminated: false, - force_synchronization: false, + force_synchronization: self.force_synchronization, universe, data: DataPacketDmpLayer { property_values: { diff --git a/tests/ipv4_tests.rs b/tests/ipv4_tests.rs index bf17e14..f3e92bd 100644 --- a/tests/ipv4_tests.rs +++ b/tests/ipv4_tests.rs @@ -2822,8 +2822,7 @@ fn test_send_recv_multiple_sync_universes(){ /// /// This shows that this timeout mechanism to stop universe synchronisation works. /// -/// Note that this library does not attempt to implement the force_synchronisation bit behaviour and so therefore always stops universe synchronisation if the -/// sync packet is not received within the timeout. +/// The sender leaves the `Force_Synchronization` option unset, so the receiver keeps holding data rather than processing it unsynchronised. /// #[test] #[ignore] diff --git a/tests/src_tests.rs b/tests/src_tests.rs index fe8f248..ed19781 100644 --- a/tests/src_tests.rs +++ b/tests/src_tests.rs @@ -703,6 +703,18 @@ 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, 100, ]; +/// The `Force_Synchronization` option defaults to off and can be turned on, as per ANSI E1.31-2016 Section 6.2.6. +#[test] +#[serial] +fn test_set_force_synchronization() { + let mut src = SacnSource::new_v4("Source").unwrap(); + assert!(!src.force_synchronization().unwrap()); + src.set_force_synchronization(true).unwrap(); + assert!(src.force_synchronization().unwrap()); + src.set_force_synchronization(false).unwrap(); + assert!(!src.force_synchronization().unwrap()); +} + /// A 64 byte source name fills the whole field and leaves no room for the null terminator required by /// ANSI E1.31-2016 Section 6.2.2, so it must be rejected. #[test]