Merge pull request #82 from QuantumEntangledAndy/fix/pause_buffers

Fix pause buffers

Cargo.toml

@@ -23,6 +23,8 @@ gstreamer = "0.20.3"
 gstreamer-app = "0.20.0"
 gstreamer-rtsp = "0.20.0"
 gstreamer-rtsp-server = { version = "0.20.3", features = ["v1_16"] }
+is_sorted = "0.1.1"
+itertools = "0.10.5"
 lazy_static = "1.4.0"
 log = { version = "0.4.17", features = [ "release_max_level_debug" ] }
 neolink_core = { path = "crates/core", version = "0.5.10" }

crates/core/src/bc_protocol/connection/discovery.rs

@@ -58,7 +58,7 @@ struct UidLookupResults {
 
 pub(crate) struct Discovery {}
 
-const MTU: u32 = 1030;
+const MTU: u32 = 1350;
 lazy_static! {
     static ref P2P_RELAY_HOSTNAMES: [&'static str; 10] = [
         "p2p.reolink.com",

crates/core/src/bc_protocol/connection/udpsource.rs

@@ -28,7 +28,7 @@ use tokio_util::{
     udp::UdpFramed,
 };
 
-const MTU: usize = 1030;
+const MTU: usize = 1350;
 const UDPDATA_HEADER_SIZE: usize = 20;
 
 pub(crate) struct UdpSource {

sample_config.toml

@@ -75,6 +75,22 @@ address = "192.168.1.187:9000"
 #
 # print_format = "None"
 
+# Internally neolink uses a buffer to improve delivery of frames at their actual times
+# A large buffer can work even for large interupts in the connection with the camera
+# A small buffer will reduce the latency introduced
+# buffer_size: 100
+
+# Over long periods of time the camera buffer can exhaust with no means of recovery
+# to combat this a time stretching stratery is employed where the stream is slowed
+# as the buffer is exhausted and sped up when full
+# this can keep the camera and rtsp streams in sync and reduce buffer exhaustions
+#
+# HOWEVER if there is real latency issues (such as too high a bitrate)
+# then smoothing can result in a stream appearing in slow motion
+#
+# you can disable and enable this feature here
+# use_smoothing: true
+
 
 [[cameras]]
 name = "storage shed"

src/config.rs

@@ -102,6 +102,22 @@ pub(crate) struct CameraConfig {
 
     #[serde(default = "default_update_time", alias = "time")]
     pub(crate) update_time: bool,
+
+    #[validate(range(
+        min = 10,
+        max = 500,
+        message = "Invalid buffer size",
+        code = "buffer_size"
+    ))]
+    #[serde(default = "default_buffer_size", alias = "size", alias = "buffer")]
+    pub(crate) buffer_size: usize,
+
+    #[serde(
+        default = "default_smoothing",
+        alias = "smoothing",
+        alias = "stretching"
+    )]
+    pub(crate) use_smoothing: bool,
 }
 
 #[derive(Debug, Deserialize, Validate, Clone)]
@@ -239,6 +255,14 @@ fn default_pause() -> PauseConfig {
     }
 }
 
+fn default_smoothing() -> bool {
+    true
+}
+
+fn default_buffer_size() -> usize {
+    100
+}
+
 pub(crate) static RESERVED_NAMES: &[&str] = &["anyone", "anonymous"];
 fn validate_username(name: &str) -> Result<(), ValidationError> {
     if name.trim().is_empty() {

src/rtsp/gst/factory.rs

@@ -41,14 +41,16 @@ glib::wrapper! {
 
 impl Default for NeoMediaFactory {
     fn default() -> Self {
-        Self::new()
+        Self::new(100, true)
     }
 }
 
 impl NeoMediaFactory {
-    pub(crate) fn new() -> Self {
+    pub(crate) fn new(buffer_size: usize, use_smoothing: bool) -> Self {
         let factory = Object::new::<NeoMediaFactory>();
         factory.set_shared(false);
+        factory.imp().shared.set_buffer_size(buffer_size);
+        factory.imp().shared.set_use_smoothing(use_smoothing);
         // factory.set_do_retransmission(false); // Can't use as the method is missing on the 32bit docker gst dll
         factory.set_launch("videotestsrc pattern=\"snow\" ! video/x-raw,width=896,height=512,framerate=25/1 ! textoverlay name=\"inittextoverlay\" text=\"Stream not Ready\" valignment=top halignment=left font-desc=\"Sans, 32\" ! jpegenc ! rtpjpegpay name=pay0");
         factory.set_suspend_mode(RTSPSuspendMode::None);
@@ -120,6 +122,8 @@ pub(crate) enum FactoryCommand {
     BcMedia(BcMedia),
     ClearBuffer,
     JumpToLive,
+    Pause,
+    Resume,
 }
 
 pub(crate) struct NeoMediaFactoryImpl {
@@ -149,6 +153,7 @@ impl Default for NeoMediaFactoryImpl {
             shared.clone(),
             ReceiverStream::new(datarx),
             ReceiverStream::new(rx_clientsender),
+            100,
         );
         threads.spawn(async move {
             loop {
@@ -265,7 +270,10 @@ impl NeoMediaFactoryImpl {
             bin.remove(&element)?;
         }
 
-        let mut client_data = NeoMediaSender::new();
+        let mut client_data = NeoMediaSender::new(
+            self.shared.get_buffer_size(),
+            self.shared.get_use_smoothing(),
+        );
 
         // Now contruct the actual ones
         match (