Support the Force_Synchronization option

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<u16, Instant>,
 }
 
 /// 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<u8> = 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);

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<HashMap<u16, u8>>,
@@ -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<bool> {
+        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: {

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]

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]