Merge branch 'master' into feature/shared_gst

Dockerfile

@@ -8,8 +8,6 @@ FROM docker.io/rust:slim-bookworm AS build
 ARG TARGETPLATFORM
 
 ENV DEBIAN_FRONTEND=noninteractive
-
-
 WORKDIR /usr/local/src/neolink
 COPY . /usr/local/src/neolink
 
@@ -85,6 +83,8 @@ RUN gst-inspect-1.0; \
     "/usr/local/bin/neolink" --version && \
     mkdir -m 0700 /root/.config/ # Location that the push notifications are cached to
 
-CMD ["/usr/local/bin/neolink", "rtsp", "--config", "/etc/neolink.toml"]
+ENV NEO_LINK_MODE="rtsp" NEO_LINK_PORT=8554
+
+CMD /usr/local/bin/neolink ${NEO_LINK_MODE} --config /etc/neolink.toml
 ENTRYPOINT ["/entrypoint.sh"]
-EXPOSE 8554
+EXPOSE ${NEO_LINK_PORT}

README.md

@@ -33,7 +33,7 @@ features not yet in upstream master.
 **Minor Features**:
 
 - Improved error messages when missing gstreamer plugins
-- Protocol more closely follows offical reolink format
+- Protocol more closely follows official reolink format
   - Possibly can handle more simulatenous connections
 - More ways to connect to the camera. Including Relaying through reolink
 servers
@@ -50,7 +50,7 @@ Install the latest [gstreamer](https://gstreamer.freedesktop.org/download/)
 (1.20.5 as of writing this).
 
 - **Windows**: ensure you install `full` when prompted in the MSI options.
-- **Mac**: Install the dpkg version on the offical gstreamer website over
+- **Mac**: Install the dpkg version on the official gstreamer website over
   the brew version
 - **Ubuntu/Debian**: These packages should work
 
@@ -269,7 +269,7 @@ reflected in the rtsp
 
 These include changing the:
 
-- Avaliable users
+- Available users
 
 ```toml
 [[users]]
@@ -284,7 +284,7 @@ These include changing the:
   permitted_users = [ "me" ]
 ```
 
-- Avaliable streams
+- Available streams
 
 ```toml
 [[cameras]]
@@ -324,7 +324,7 @@ enable_motion = false        # motion detection
                              # (limited battery drain since it
                              # is a passive listening connection)
                              #
-enable_light = false         # flood lights only avaliable on some camera
+enable_light = false         # flood lights only available on some camera
                              # (limited battery drain since it
                              # is a passive listening connection)
                              #
@@ -355,7 +355,7 @@ must be manually specified.
   features = ["floodlight"]
 ```
 
-Avaliable features are:
+Available features are:
 
 - `floodlight`: This adds a light control to home assistant
 - `camera`: This adds a camera preview to home assistant. It is only updated
@@ -456,6 +456,10 @@ docker pull quantumentangledandy/neolink
 # network=host, notably macos lacks this option.
 docker run --network host --volume=$PWD/config.toml:/etc/neolink.toml quantumentangledandy/neolink
 ```
+#### Environmental Variables 
+There are currently 2 environmental variables available as part of the container: 
+- `NEO_LINK_MODE`: defaults to `"rtsp"` if not set, other options are "mqtt" or "mqtt-rtsp".
+- `NEO_LINK_PORT`: defaults to `8554`, set this to your required port value.
 
 ### Image
 

crates/core/src/bc/de.rs

@@ -448,7 +448,7 @@ mod tests {
 
     #[test]
     // B800 seems to have a different header to the E1 and swann cameras
-    // the stream_type and message_num do not seem to set in the offical clients
+    // the stream_type and message_num do not seem to set in the official clients
     //
     // They also have extra streams
     fn test_bc_b800_externstream() {
@@ -492,7 +492,7 @@ mod tests {
 
     #[test]
     // B800 seems to have a different header to the E1 and swann cameras
-    // the stream_type and message_num do not seem to set in the offical clients
+    // the stream_type and message_num do not seem to set in the official clients
     //
     // They also have extra streams
     fn test_bc_b800_substream() {
@@ -536,7 +536,7 @@ mod tests {
 
     #[test]
     // B800 seems to have a different header to the E1 and swann cameras
-    // the stream_type and message_num do not seem to set in the offical clients
+    // the stream_type and message_num do not seem to set in the official clients
     //
     // They also have extra streams
     fn test_bc_b800_mainstream() {

crates/core/src/bc/xml.rs

@@ -280,7 +280,7 @@ pub struct Extension {
     #[serde(rename = "binaryData", skip_serializing_if = "Option::is_none")]
     pub binary_data: Option<u32>,
     /// Certain requests such `AbilitySupport` require to know which user this
-    /// ability support request is for (why camera dosen't know this based on who
+    /// ability support request is for (why camera doesn't know this based on who
     /// is logged in is unknown... Possible security hole)
     #[serde(rename = "userName", skip_serializing_if = "Option::is_none")]
     pub user_name: Option<String>,

crates/core/src/bc_protocol/abilityinfo.rs

@@ -30,7 +30,7 @@ impl BcCamera {
         sub_get.send(get).await?;
         let msg = sub_get.recv().await?;
         if msg.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
         }
 
         if let BcBody::ModernMsg(ModernMsg {

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

@@ -248,7 +248,7 @@ impl Discoverer {
         tokio::select! {
             v = async {
                 loop {
-                    let (reply, addr) = reply.next().await.ok_or(Error::ConnectionUnavaliable)??;
+                    let (reply, addr) = reply.next().await.ok_or(Error::ConnectionUnavailable)??;
                     if let Some(result) = map(reply, addr) {
                         return Ok(result);
                     }

crates/core/src/bc_protocol/credentials.rs

@@ -7,7 +7,7 @@ use std::convert::TryInto;
 pub struct Credentials {
     /// The username to login to the camera with
     pub username: String,
-    /// The password to use for login. Some camera allow this to be ommited
+    /// The password to use for login. Some camera allow this to be omitted
     pub password: Option<String>,
 }
 

crates/core/src/bc_protocol/errors.rs

@@ -36,8 +36,8 @@ pub enum Error {
     },
 
     /// Raised when the camera responds with a status code over than OK
-    #[error("Camera responded with Service Unavaliable: {}", _0)]
-    CameraServiceUnavaliable(u16),
+    #[error("Camera responded with Service Unavaliable: {0}")]
+    CameraServiceUnavailable(u16),
 
     /// Raised when the camera responds with a status code over than OK during login
     #[error("Camera responded with Err during login")]
@@ -115,7 +115,7 @@ pub enum Error {
 
     /// Raised when the camera cannot be found
     #[error("Camera Not Findable")]
-    ConnectionUnavaliable,
+    ConnectionUnavailable,
 
     /// Raised when the subscription id dropped too soon
     #[error("Dropped Subscriber")]

crates/core/src/bc_protocol/floodlight.rs

@@ -129,7 +129,7 @@ impl BcCamera {
         sub_get.send(get).await?;
         let msg = sub_get.recv().await?;
         if msg.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
         }
 
         if let BcBody::ModernMsg(ModernMsg {
@@ -181,7 +181,7 @@ impl BcCamera {
         sub_get.send(get).await?;
         let msg = sub_get.recv().await?;
         if msg.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
         }
 
         Ok(())

crates/core/src/bc_protocol/keepalive.rs

@@ -2,7 +2,7 @@ use super::{BcCamera, Result};
 use crate::bc::model::*;
 
 impl BcCamera {
-    /// Create a handller to respond to keep alive messages
+    /// Create a handler to respond to keep alive messages
     /// These messages are sent by the camera so we listen to
     /// a message ID rather than setting a message number and
     /// responding to it

crates/core/src/bc_protocol/ledstate.rs

@@ -29,7 +29,7 @@ impl BcCamera {
         sub_get.send(get).await?;
         let msg = sub_get.recv().await?;
         if msg.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
         }
 
         if let BcBody::ModernMsg(ModernMsg {

crates/core/src/bc_protocol/link.rs

@@ -27,7 +27,7 @@ impl BcCamera {
         sub_get.send(get).await?;
         let msg = sub_get.recv().await?;
         if msg.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
         }
 
         if let BcBody::ModernMsg(ModernMsg {

crates/core/src/bc_protocol/pirstate.rs

@@ -39,10 +39,10 @@ impl BcCamera {
                     reties += 1;
                     continue;
                 } else {
-                    return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+                    return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
                 }
             } else if msg.meta.response_code != 200 {
-                return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+                return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
             } else {
                 // Valid message with response_code == 200
                 if let BcBody::ModernMsg(ModernMsg {
@@ -101,7 +101,7 @@ impl BcCamera {
         {
             let msg = reply?;
             if msg.meta.response_code != 200 {
-                return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+                return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
             }
 
             if let BcMeta {

crates/core/src/bc_protocol/ptz.rs

@@ -325,7 +325,7 @@ impl BcCamera {
         sub_get.send(get).await?;
         let msg = sub_get.recv().await?;
         if msg.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
         }
 
         if let BcBody::ModernMsg(ModernMsg {

crates/core/src/bc_protocol/pushinfo.rs

@@ -64,7 +64,7 @@ impl BcCamera {
         sub.send(msg).await?;
         let msg = sub.recv().await?;
         if msg.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
         }
 
         Ok(())

crates/core/src/bc_protocol/siren.rs

@@ -39,7 +39,7 @@ impl BcCamera {
         sub_get.send(get).await?;
         let msg = sub_get.recv().await?;
         if msg.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
         }
 
         Ok(())

crates/core/src/bc_protocol/snap.rs

@@ -41,7 +41,7 @@ impl BcCamera {
         sub_get.send(get).await?;
         let msg = sub_get.recv().await?;
         if msg.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
         }
 
         if let BcBody::ModernMsg(ModernMsg {
@@ -132,7 +132,7 @@ impl BcCamera {
                 }
             } else {
                 // anything else is an error
-                return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+                return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
             }
 
             // let binary_stream = sub_get.payload_stream();

crates/core/src/bc_protocol/stream.rs

@@ -17,7 +17,7 @@ pub enum StreamKind {
     Sub,
     /// This stream represents a balance between SD and HD
     ///
-    /// It is only avaliable on some camera. If the camera dosen't
+    /// It is only available on some camera. If the camera doesn't
     /// support it the stream will be the same as the SD stream
     Extern,
 }
@@ -155,7 +155,7 @@ impl BcCamera {
                 StreamKind::Extern => 0,
             };
 
-            // Theses are the numbers used with the offical client
+            // Theses are the numbers used with the official client
             // On an E1 and swann cameras:
             //  - mainStream always has a value of 0
             //  - subStream always has a value of 1
@@ -271,7 +271,7 @@ impl BcCamera {
                             msg_id: MSG_ID_VIDEO_STOP,
                             ..
                         }   = msg.meta {
-                            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+                            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
                         }
                     }
                 } => v,
@@ -313,7 +313,7 @@ impl BcCamera {
             StreamKind::Extern => 0,
         };
 
-        // Theses are the numbers used with the offical client
+        // Theses are the numbers used with the official client
         // On an E1 and swann cameras:
         //  - mainStream always has a value of 0
         //  - subStream always has a value of 1
@@ -352,7 +352,7 @@ impl BcCamera {
 
         let reply = sub_video.recv().await?;
         if reply.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(reply.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(reply.meta.response_code));
         }
 
         Ok(())

crates/core/src/bc_protocol/stream_info.rs

@@ -27,7 +27,7 @@ impl BcCamera {
         sub_get.send(get).await?;
         let msg = sub_get.recv().await?;
         if msg.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
         }
 
         if let BcBody::ModernMsg(ModernMsg {

crates/core/src/bc_protocol/support.rs

@@ -26,7 +26,7 @@ impl BcCamera {
         sub_get.send(get).await?;
         let msg = sub_get.recv().await?;
         if msg.meta.response_code != 200 {
-            return Err(Error::CameraServiceUnavaliable(msg.meta.response_code));
+            return Err(Error::CameraServiceUnavailable(msg.meta.response_code));
         }
 
         if let BcBody::ModernMsg(ModernMsg {

crates/core/src/bc_protocol/talk.rs

@@ -323,7 +323,7 @@ impl BcCamera {
 
         let full_block_size = block_size + 4; // Block size + predictor state
         let msg_num = self.new_message_num();
-        let sub = connection.subscribe(MSG_ID_TALK, msg_num).await?;
+        let mut sub = connection.subscribe(MSG_ID_TALK, msg_num).await?;
 
         const BLOCK_PER_PAYLOAD: usize = 1;
         const BLOCK_HEADER_SIZE: usize = 4;
@@ -333,9 +333,10 @@ impl BcCamera {
 
         let target_chunks = full_block_size as usize * BLOCK_PER_PAYLOAD;
 
-        let mut payload_bytes = vec![];
         let mut end_of_stream = false;
+        let mut expected_stream_end = std::time::Instant::now();
         while !end_of_stream {
+            let mut payload_bytes = vec![];
             while payload_bytes.len() < target_chunks {
                 let mut buffer = vec![255; target_chunks - payload_bytes.len()];
                 if let Ok(read) = buffered_recv.read(&mut buffer) {
@@ -373,8 +374,6 @@ impl BcCamera {
                 break;
             };
 
-            payload_bytes = vec![];
-
             // Time to play the sample in seconds
             let play_length = samples_sent as f32 / sample_rate as f32;
 
@@ -397,11 +396,22 @@ impl BcCamera {
                 }),
             };
 
+            let time_sent = std::time::Instant::now();
             sub.send(msg).await?;
-
-            std::thread::sleep(std::time::Duration::from_secs_f32(play_length * 0.95));
+            let play_length = std::time::Duration::from_secs_f32(play_length);
+            if time_sent > expected_stream_end {
+                expected_stream_end = time_sent + play_length;
+            } else {
+                expected_stream_end += play_length;
+            }
+            let _ = sub.recv().await?;
         }
 
+        // Chunks are still being played, while talk_stop will interrupt them. Wait until we expect
+        // the stream to end (+ and extra 100ms) before issuing talk_stop.
+        let remaining_stream_duration = expected_stream_end - std::time::Instant::now();
+        std::thread::sleep(remaining_stream_duration + std::time::Duration::from_secs_f32(0.1));
+
         self.talk_stop().await?;
 
         Ok(())