parking_lot & 'to' parameter in TranscodeMinigroupToHls task

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dispatcher"
-version = "0.4.21"
+version = "0.4.22"
 authors = ["Andrei Nesterov <[email protected]>"]
 description = "Service managing classrooms"
 readme = "README.md"
@@ -68,6 +68,7 @@ tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
 url = { version = "2.2.1", features = ["serde"] }
 uuid = { version = "0.8", features = ["v4", "serde"] }
 vec1 = { version = "1.8.0", features = ["serde"] }
+parking_lot = "0.12"
 
 [dev-dependencies]
 lazy_static = "1.4"

src/clients/event/layer/metrics.rs

@@ -1,6 +1,6 @@
+use parking_lot::RwLock;
 use std::collections::HashMap;
 use std::pin::Pin;
-use std::sync::RwLock;
 use std::task::{Context, Poll};
 
 use anyhow::Result;

src/clients/tq.rs

@@ -63,7 +63,9 @@ impl<'a> TaskWithOptions<'a> {
             Task::ConvertMjrDumpsToStream { .. } => {
                 self.to = settings.to.as_deref();
             }
-            _ => {}
+            Task::TranscodeMinigroupToHls { .. } => {
+                self.to = settings.to.as_deref();
+            }
         }
     }
 }

src/test_helpers/state.rs

@@ -1,4 +1,5 @@
-use std::sync::{Arc, Mutex};
+use parking_lot::Mutex;
+use std::sync::Arc;
 
 use anyhow::Result;
 use async_trait::async_trait;
@@ -252,10 +253,7 @@ impl TestPublisher {
     }
 
     pub fn flush(&self) -> Vec<OutgoingEnvelope> {
-        let mut messages_lock = self
-            .messages
-            .lock()
-            .expect("Failed to obtain messages lock");
+        let mut messages_lock = self.messages.lock();
 
         (*messages_lock).drain(0..).collect::<Vec<_>>()
     }
@@ -269,11 +267,7 @@ impl Publisher for TestPublisher {
             .expect("Failed to parse dumped message");
 
         parsed_message.set_topic(dump.topic());
-
-        let mut messages_lock = self
-            .messages
-            .lock()
-            .expect("Failed to obtain messages lock");
+        let mut messages_lock = self.messages.lock();
 
         (*messages_lock).push(parsed_message);
         Ok(())

src/utils.rs

@@ -39,10 +39,8 @@ where
 
 #[cfg(test)]
 mod tests {
-    use std::{
-        sync::{Arc, Mutex},
-        time::Duration,
-    };
+    use parking_lot::Mutex;
+    use std::{sync::Arc, time::Duration};
 
     use super::single_retry;
 
@@ -52,7 +50,7 @@ mod tests {
         let create_fut = || {
             let call_count = call_count.clone();
             async move {
-                let mut mutex_guard = call_count.lock().unwrap();
+                let mut mutex_guard = call_count.lock();
                 *mutex_guard += 1;
                 if *mutex_guard == 1 {
                     Err(())
@@ -64,7 +62,7 @@ mod tests {
 
         let result = single_retry(create_fut, Duration::from_secs(10000)).await;
 
-        let guard = call_count.lock().unwrap();
+        let guard = call_count.lock();
         assert_eq!(*guard, 2);
         assert!(result.is_ok());
     }
@@ -75,7 +73,7 @@ mod tests {
         let create_fut = || {
             let call_count = call_count.clone();
             async move {
-                let mut mutex_guard = call_count.lock().unwrap();
+                let mut mutex_guard = call_count.lock();
                 *mutex_guard += 1;
                 if *mutex_guard == 1 {
                     Err::<(), _>(*mutex_guard)
@@ -87,7 +85,7 @@ mod tests {
 
         let result = single_retry(create_fut, Duration::from_secs(10000)).await;
 
-        let guard = call_count.lock().unwrap();
+        let guard = call_count.lock();
         assert_eq!(*guard, 2);
         assert_eq!(result, Err(2));
     }
@@ -99,7 +97,7 @@ mod tests {
             let call_count = call_count.clone();
             async move {
                 let call_count = {
-                    let mut mutex_guard = call_count.lock().unwrap();
+                    let mut mutex_guard = call_count.lock();
                     *mutex_guard += 1;
                     *mutex_guard
                 };
@@ -114,7 +112,7 @@ mod tests {
 
         let result = single_retry(create_fut, Duration::from_millis(1)).await;
 
-        let guard = call_count.lock().unwrap();
+        let guard = call_count.lock();
         assert_eq!(*guard, 2);
         assert!(result.is_ok());
     }
@@ -126,7 +124,7 @@ mod tests {
             let call_count = call_count.clone();
             async move {
                 let call_count = {
-                    let mut mutex_guard = call_count.lock().unwrap();
+                    let mut mutex_guard = call_count.lock();
                     *mutex_guard += 1;
                     *mutex_guard
                 };
@@ -140,7 +138,7 @@ mod tests {
 
         let result = single_retry(create_fut, Duration::from_millis(1)).await;
 
-        let guard = call_count.lock().unwrap();
+        let guard = call_count.lock();
         assert_eq!(*guard, 1);
         assert!(result.is_ok());
     }