Progress

src/pipeline/builder.rs

@@ -1,4 +1,5 @@
 use tokio::sync::mpsc::UnboundedReceiver;
+
 use crate::ingress::ConnectionInfo;
 use crate::pipeline::runner::PipelineRunner;
 use crate::webhook::Webhook;

src/pipeline/mod.rs

@@ -1,7 +1,10 @@
 use std::ops::{Deref, DerefMut};
 
 use async_trait::async_trait;
-use ffmpeg_sys_next::{AVFrame, AVPacket};
+use ffmpeg_sys_next::{
+    av_frame_alloc, av_frame_free, av_frame_ref, av_packet_alloc, av_packet_free, av_packet_ref,
+    AVFrame, AVPacket,
+};
 use serde::{Deserialize, Serialize};
 
 use crate::demux::info::DemuxStreamInfo;
@@ -34,7 +37,7 @@ pub struct PipelineConfig {
     pub egress: Vec<EgressType>,
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, PartialEq)]
 pub enum PipelinePayload {
     /// No output
     Empty,
@@ -49,8 +52,45 @@ pub enum PipelinePayload {
 }
 
 unsafe impl Send for PipelinePayload {}
+
 unsafe impl Sync for PipelinePayload {}
 
+impl Clone for PipelinePayload {
+    fn clone(&self) -> Self {
+        match self {
+            PipelinePayload::Empty => PipelinePayload::Empty,
+            PipelinePayload::Bytes(b) => PipelinePayload::Bytes(b.clone()),
+            PipelinePayload::AvPacket(p) => unsafe {
+                let new_pkt = av_packet_alloc();
+                av_packet_ref(new_pkt, *p);
+                PipelinePayload::AvPacket(new_pkt)
+            },
+            PipelinePayload::AvFrame(p) => unsafe {
+                let new_frame = av_frame_alloc();
+                av_frame_ref(new_frame, *p);
+                PipelinePayload::AvFrame(new_frame)
+            },
+            PipelinePayload::SourceInfo(i) => PipelinePayload::SourceInfo(i.clone()),
+        }
+    }
+}
+
+impl Drop for PipelinePayload {
+    fn drop(&mut self) {
+        match self {
+            PipelinePayload::Empty => {}
+            PipelinePayload::Bytes(_) => {}
+            PipelinePayload::AvPacket(p) => unsafe {
+                av_packet_free(p);
+            },
+            PipelinePayload::AvFrame(p) => unsafe {
+                av_frame_free(p);
+            },
+            PipelinePayload::SourceInfo(_) => {}
+        }
+    }
+}
+
 #[async_trait]
 pub trait PipelineStep {
     fn name(&self) -> String;

src/pipeline/runner.rs

@@ -1,15 +1,19 @@
+use std::ops::{Add, AddAssign};
+use std::time::{Duration, Instant};
+
+use anyhow::Error;
+use log::info;
+use tokio::sync::broadcast;
+use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
+
 use crate::decode::Decoder;
-use crate::demux::info::{DemuxStreamInfo, StreamChannelType};
 use crate::demux::Demuxer;
+use crate::demux::info::{DemuxStreamInfo, StreamChannelType};
 use crate::egress::hls::HlsEgress;
 use crate::encode::Encoder;
 use crate::pipeline::{EgressType, PipelineConfig, PipelinePayload, PipelineStep};
 use crate::scale::Scaler;
 use crate::variant::VariantStream;
-use anyhow::Error;
-use log::info;
-use tokio::sync::broadcast;
-use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
 
 struct ScalerEncoder {
     pub scaler: Scaler,
@@ -24,6 +28,9 @@ pub struct PipelineRunner {
     scalers: Vec<ScalerEncoder>,
     encoders: Vec<Encoder<broadcast::Receiver<PipelinePayload>>>,
     egress: Vec<HlsEgress>,
+    started: Instant,
+    frame_no: u64,
+    stream_info: Option<DemuxStreamInfo>,
 }
 
 impl PipelineRunner {
@@ -38,14 +45,39 @@ impl PipelineRunner {
             scalers: vec![],
             encoders: vec![],
             egress: vec![],
+            started: Instant::now(),
+            frame_no: 0,
+            stream_info: None,
         }
     }
 
     pub fn run(&mut self) -> Result<(), Error> {
+        if let Some(info) = &self.stream_info {
+            if let Some(v_stream) = info
+                .channels
+                .iter()
+                .find(|s| s.channel_type == StreamChannelType::Video)
+            {
+                let duration = self.frame_no as f64 / v_stream.fps as f64;
+                let target_time = self.started.add(Duration::from_secs_f64(duration));
+                let now = Instant::now();
+                if now < target_time {
+                    let poll_sleep = target_time - now;
+                    std::thread::sleep(poll_sleep);
+                }
+            }
+        }
         if let Some(cfg) = self.demuxer.process()? {
             self.configure_pipeline(cfg)?;
         }
-        self.decoder.process()?;
+        let frames = self.decoder.process()?;
+        if let Some(v) = self.frame_no.checked_add(frames as u64) {
+            self.frame_no = v;
+        } else {
+            panic!("Frame number overflowed, maybe you need a bigger number!");
+        }
+
+        // video scalar-encoder chains
         for sw in &mut self.scalers {
             sw.scaler.process()?;
             sw.encoder.process()?;
@@ -53,15 +85,22 @@ impl PipelineRunner {
                 eg.process()?;
             }
         }
+        // audio encoder chains
+        for enc in &mut self.encoders {
+            enc.process()?;
+            for eg in &mut self.egress {
+                eg.process()?;
+            }
+        }
         Ok(())
     }
 
     fn configure_pipeline(&mut self, info: DemuxStreamInfo) -> Result<(), Error> {
-        // configure scalers
-        if self.scalers.len() != 0 {
+        if self.stream_info.is_some() {
             return Err(Error::msg("Pipeline already configured!"));
         }
         info!("Configuring pipeline {:?}", info);
+        self.stream_info = Some(info.clone());
 
         let video_stream = info
             .channels
@@ -100,7 +139,7 @@ impl PipelineRunner {
                                     return Err(Error::msg(format!(
                                         "Variant config not supported {:?}",
                                         c
-                                    )))
+                                    )));
                                 }
                             }
                         }
@@ -109,6 +148,11 @@ impl PipelineRunner {
                 }
             }
         }
-        Ok(())
+
+        if self.egress.len() == 0 {
+            Err(Error::msg("No egress config, pipeline misconfigured!"))
+        } else {
+            Ok(())
+        }
     }
 }