fix sequencer implementation
use std::fmt;
-use std::sync::atomic::{AtomicU16, AtomicU64, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
/// Sequencer generates sequential sequence numbers for building RTP packets
pub trait Sequencer: fmt::Debug {
@@ -18,51 +17,55 @@ impl Clone for Box<dyn Sequencer + Send + Sync> {
/// NewRandomSequencer returns a new sequencer starting from a random sequence
/// number
pub fn new_random_sequencer() -> impl Sequencer {
- SequencerImpl {
- sequence_number: Arc::new(AtomicU16::new(rand::random::<u16>())),
- roll_over_count: Arc::new(AtomicU64::new(0)),
- }
+ let c = Counters {
+ sequence_number: rand::random::<u16>(),
+ roll_over_count: 0,
+ };
+ SequencerImpl(Arc::new(Mutex::new(c)))
}
/// NewFixedSequencer returns a new sequencer starting from a specific
/// sequence number
pub fn new_fixed_sequencer(s: u16) -> impl Sequencer {
- SequencerImpl {
- sequence_number: if s == 0 {
- Arc::new(AtomicU16::new(u16::MAX))
- } else {
- Arc::new(AtomicU16::new(s - 1))
- },
- roll_over_count: Arc::new(AtomicU64::new(0)),
- }
+ let sequence_number = if s == 0 { u16::MAX } else { s - 1 };
+
+ let c = Counters {
+ sequence_number,
+ roll_over_count: 0,
+ };
+
+ SequencerImpl(Arc::new(Mutex::new(c)))
}
#[derive(Debug, Clone)]
-struct SequencerImpl {
- sequence_number: Arc<AtomicU16>,
- roll_over_count: Arc<AtomicU64>,
+struct SequencerImpl(Arc<Mutex<Counters>>);
+
+#[derive(Debug)]
+struct Counters {
+ sequence_number: u16,
+ roll_over_count: u64,
}
impl Sequencer for SequencerImpl {
/// NextSequenceNumber increment and returns a new sequence number for
/// building RTP packets
fn next_sequence_number(&self) -> u16 {
- let sequence_number = self.sequence_number.load(Ordering::SeqCst);
- if sequence_number == u16::MAX {
- self.roll_over_count.fetch_add(1, Ordering::SeqCst);
- self.sequence_number.store(0, Ordering::SeqCst);
- 0
+ let mut lock = self.0.lock().unwrap();
+
+ if lock.sequence_number == u16::MAX {
+ lock.roll_over_count += 1;
+ lock.sequence_number = 0;
} else {
- self.sequence_number
- .store(sequence_number + 1, Ordering::SeqCst);
- sequence_number + 1
+ lock.sequence_number += 1;
}
+
+ lock.sequence_number
}
/// RollOverCount returns the amount of times the 16bit sequence number
/// has wrapped
fn roll_over_count(&self) -> u64 {
- self.roll_over_count.load(Ordering::SeqCst)
+ self.0.lock().unwrap().roll_over_count
}
fn clone_to(&self) -> Box<dyn Sequencer + Send + Sync> {
