video buffer telemetry

Cargo.lock

@@ -37,13 +37,14 @@ checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa"
 
 [[package]]
 name = "balloon_tx_monolith"
-version = "0.3.0"
+version = "0.3.1"
 dependencies = [
  "anyhow",
  "aprs-parser",
  "base64",
  "bitvec",
  "crc",
+ "crossbeam-channel",
  "gstreamer",
  "gstreamer-app",
  "gstreamer-video",

Cargo.toml

 [package]
 name = "balloon_tx_monolith"
-version = "0.3.0"
+version = "0.3.1"
 edition = "2021"
 license = "MIT"
 
@@ -32,3 +32,4 @@ toml = "0.7.3"
 gstreamer = "0.21.0"
 gstreamer-app = "0.21.0"
 gstreamer-video = "0.21.0"
+crossbeam-channel = "0.5.8"

src/control.rs

@@ -31,7 +31,7 @@ impl Controller {
 
     pub fn run_forever(self) {
         let (img_tx, img_rx) = mpsc::sync_channel(IMAGE_PACKET_QUEUE_LENGTH);
-        let (vid_tx, vid_rx) = mpsc::sync_channel(VIDEO_PACKET_QUEUE_LENGTH);
+        let (vid_tx, vid_rx) = crossbeam_channel::bounded(VIDEO_PACKET_QUEUE_LENGTH);
         let (telem_tx, telem_rx) = mpsc::channel();
         let (cmd_tx, cmd_rx) = mpsc::channel();
 
@@ -94,7 +94,7 @@ impl Controller {
     fn tx_thread(
         callsign: String,
         image_rx: Receiver<FecPacket>,
-        vid_rx: Receiver<Vec<u8>>,
+        vid_rx: crossbeam_channel::Receiver<Vec<u8>>,
         telem_rx: Receiver<Packet>,
     ) {
         let mut radio = loop {
@@ -135,7 +135,7 @@ impl Controller {
     fn tx_thread_single_iter(
         callsign: &str,
         image_rx: &Receiver<FecPacket>,
-        vid_rx: &Receiver<Vec<u8>>,
+        vid_rx: &crossbeam_channel::Receiver<Vec<u8>>,
         telem_rx: &Receiver<Packet>,
         text_msg_id: &mut u16,
         last_got_temp: &mut Instant,
@@ -181,7 +181,15 @@ impl Controller {
 
             let temp = radio.get_temp()?;
 
-            let packet = Packet::new_text_message(callsign, &format!("Temp: {}", temp));
+            let packet = Packet::new_text_message(
+                callsign,
+                &format!(
+                    "Temp: {}, Vbuf: {} / {}",
+                    temp,
+                    vid_rx.len(),
+                    VIDEO_PACKET_QUEUE_LENGTH
+                ),
+            );
 
             radio.send_packet(&packet.into_raw(text_msg_id).into())?;
         }

src/video.rs

-use std::{
-    sync::mpsc::{SyncSender, TrySendError},
-    thread,
-    time::Duration,
-};
+use std::{thread, time::Duration};
 
 use crate::packet::{Packet, RawPacket, VIDEO_LEN};
 use anyhow::Context;
+use crossbeam_channel::{Sender, TrySendError};
 use gstreamer::{
     element_error,
     prelude::{Cast, ElementExtManual, GstBinExtManual},
@@ -73,7 +70,7 @@ impl VideoPacker {
     }
 }
 
-pub fn start_video(sender: SyncSender<Vec<u8>>) {
+pub fn start_video(sender: Sender<Vec<u8>>) {
     thread::spawn(move || loop {
         match init(sender.clone()) {
             Ok((pipeline, bus)) => handle_pipeline(pipeline, bus),
@@ -86,7 +83,7 @@ pub fn start_video(sender: SyncSender<Vec<u8>>) {
     });
 }
 
-fn init(sender: SyncSender<Vec<u8>>) -> anyhow::Result<(Pipeline, Bus)> {
+fn init(sender: Sender<Vec<u8>>) -> anyhow::Result<(Pipeline, Bus)> {
     gstreamer::init()?;
 
     let src = gstreamer::ElementFactory::make("libcamerasrc")