images

src/main.rs

@@ -1,7 +1,7 @@
 use std::net::{UdpSocket, Ipv4Addr, SocketAddr, TcpListener};
 use byteorder::{BigEndian, WriteBytesExt, ByteOrder};
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
 use std::time::Instant;
 use std::sync::mpsc;
 use std::io::{self, Write, BufReader, Read};
@@ -66,7 +66,7 @@ fn make_vsync(frame_no: u16) -> Vec<u8> {
     ret
 }
 
-fn audio_thread(udp: UdpSocket, dest: SocketAddr, queue: Consumer<u8>) {
+fn audio_thread(udp: UdpSocket, dest: SocketAddr, queue: Consumer<u8>, fc: Arc<AtomicU8>) {
     let mut pkt = [0u8; 1008];
     // Pre-write the header on to the packet.
     for (i, b) in [0, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0x55, 0, 0, 0].iter().enumerate() {
@@ -94,6 +94,7 @@ fn audio_thread(udp: UdpSocket, dest: SocketAddr, queue: Consumer<u8>) {
             // Check if we managed to fill the buf; if not, we've xrun'd.
             if ptr != 1008 {
                 num_xruns += 1;
+                fc.store(0, Ordering::Relaxed);
                 if ptr != 16 { // reduce spam
                     eprintln!("/!\\ XRUN ptr={}smpls, t={}ns", ptr, cur);
                 }
@@ -102,10 +103,12 @@ fn audio_thread(udp: UdpSocket, dest: SocketAddr, queue: Consumer<u8>) {
                 // Reset the pointer.
                 ptr = 16;
                 num_xruns = 0;
+                fc.store(1, Ordering::Relaxed);
             }
             // Check if we're already going to miss the next deadline; if so, warn.
             if cur >= (next + INTERVAL) {
-                eprintln!("/!\\ WARNING WARNING, {}ns behind!", next - cur);
+                eprintln!("/!\\ WARNING WARNING, {}ns behind!", cur - next);
+                fc.store(2, Ordering::Relaxed);
             }
             // Advance the deadline so it's 2.8ms after the previous one.
             // We skip over if we're already going to miss it.
@@ -194,6 +197,29 @@ fn test_frame_chunker() {
     assert_eq!(test_frame as &[u8], &test as &[u8]);
 }
 
+#[test]
+fn test_frame_chunker_white() {
+    let white_framed = include!("../white.rs");
+    let mut total_framed = Vec::new();
+    let mut total_unframed = Vec::new();
+    for buf in white_framed.iter() {
+        let mut mut_arr = buf.to_vec();
+        BigEndian::write_u16(&mut mut_arr, 0);
+        total_framed.extend(&mut_arr);
+        total_unframed.extend(&mut_arr[4..]);
+    }
+    let mut chonker = FrameChunker::new(0, &total_unframed as &[u8]);
+    let mut test = Vec::new();
+    let mut expected_seq = 0;
+    while !chonker.is_empty() {
+        let buf = chonker.next_chunk();
+        test.extend(&buf);
+        assert_eq!(BigEndian::read_u16(&buf[2..]) & 0b0111_1111, expected_seq);
+        expected_seq += 1;
+    }
+    assert_eq!(&total_framed as &[u8], &test as &[u8]);
+}
+
 fn spooler_loop(bufs: mpsc::Receiver<Vec<u8>>, tx: Producer<u8>) {
     while let Ok(buf) = bufs.recv() {
         for byte in buf {
@@ -226,7 +252,9 @@ fn feeder_loop<R: Read>(mut inp: R, out: mpsc::SyncSender<Vec<u8>>) {
 
 fn main() {
     let white = include!("../white.rs");
-    let test_frame = include_bytes!("../test.mjpeg");
+    let showtime = include_bytes!("../showtime.mjpeg");
+    let fuck = include_bytes!("../behind.mjpeg");
+    let bork = include_bytes!("../bork.mjpeg");
     let noise_frame = include_bytes!("../noise.dat");
     let mut vsync = UdpSocket::bind("192.168.168.55:2067").unwrap();
     let vsync_dest: SocketAddr = "226.2.2.2:2067".parse().unwrap();
@@ -247,6 +275,7 @@ fn main() {
     wakeup.set_broadcast(true).unwrap();
     println!("spawning wakeup thread");
     let mut init = Arc::new(AtomicBool::new(false));
+    let mut fucked = Arc::new(AtomicU8::new(0));
     let ic = init.clone();
     std::thread::spawn(move || {
         wakeup_loop(wakeup, wakeup_dest, ic);
@@ -255,19 +284,25 @@ fn main() {
     init.store(true, Ordering::Relaxed);
     println!("spawning audio thread");
     let (tx, rx) = make::<u8>(BUFFER_SIZE);
+    let fc = fucked.clone();
     std::thread::spawn(move || {
-        audio_thread(sound, sound_dest, rx);
+        audio_thread(sound, sound_dest, rx, fc);
     });
     println!("going brr");
     println!("{} seqs", white.len()); 
     std::thread::spawn(move || {
         loop {
+            let image = match fucked.load(Ordering::Relaxed) {
+                0 => bork as &[u8],
+                2 => fuck as &[u8],
+                _ => showtime as &[u8]
+            };
+            chonker = FrameChunker::new(frame_no, image);
             let vsync_pkt = make_vsync(frame_no);
             vsync.send_to(&vsync_pkt, &vsync_dest).unwrap();
-            for arr in white.iter() {
-                let mut mut_arr = arr.to_vec();
-                BigEndian::write_u16(&mut mut_arr, frame_no);
-                mjpeg.send_to(&mut_arr, mjpeg_dest).unwrap();
+            while !chonker.is_empty() {
+                let buf = chonker.next_chunk();
+                mjpeg.send_to(&buf, mjpeg_dest).unwrap();
             }
             frame_no += 1;
             ::std::thread::sleep_ms(33);