Send DTMF events typed

src/main.rs

@@ -4,41 +4,51 @@ use anyhow::Result;
 use gst::glib;
 use gst::glib::once_cell::sync::Lazy;
 use gst::prelude::*;
+use rand::prelude::*;
 use std::thread;
 use std::thread::sleep;
-use rand::prelude::*;
 use tokio::runtime::Builder;
 
 static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
     gst::DebugCategory::new("main", gst::DebugColorFlags::empty(), Some("Main function"))
 });
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Copy)]
-enum DtmfEvent {
+struct DtmfEvent(i32);
-    Start(DtmfStart),
+
-    End(i32),
+impl TryFrom<&gst::StructureRef> for DtmfEvent {
+    type Error = anyhow::Error;
+
+    fn try_from(structure: &gst::StructureRef) -> anyhow::Result<Self> {
+        let name = structure.name().to_string();
+        if name != "dtmf-event" {
+            anyhow::bail!("Not a dtmf-event structure: {name}");
+        }
+        let number = structure.get::<i32>("number")?;
+        Ok(Self(number))
+    }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Copy)]
-struct DtmfStart {
+enum DtmfCommand {
-    number: i32,
+    Start(i32),
-    volume: i32,
+    End(Option<i32>),
 }
 
-impl DtmfEvent {
+impl DtmfCommand {
     fn start(number: i32) -> Self {
-        Self::Start(DtmfStart { number, volume: 36 })
+        Self::Start(number)
     }
 
     fn end(self) -> Self {
         match self {
-            Self::Start(start) => Self::End(start.number),
+            Self::Start(number) => Self::End(Some(number)),
             Self::End(_) => self,
         }
     }
 }
 
-impl TryFrom<&gst::StructureRef> for DtmfEvent {
+impl TryFrom<&gst::StructureRef> for DtmfCommand {
     type Error = anyhow::Error;
 
     fn try_from(structure: &gst::StructureRef) -> anyhow::Result<Self> {
@@ -46,59 +56,60 @@ impl TryFrom<&gst::StructureRef> for DtmfEvent {
         if !name.starts_with("dtmf-event") {
             anyhow::bail!("Not a dtmf-event structure: {name}");
         }
-        let number = structure.get::<i32>("number")?;
+        let number = structure.get_optional::<i32>("number")?;
-        return if structure.get::<bool>("start")? {
+        if structure.get::<bool>("start")? {
-            let volume = structure.get::<i32>("volume")?;
+            Ok(Self::Start(number.ok_or_else(|| {
-            Ok(Self::Start(DtmfStart { number, volume }))
+                anyhow::anyhow!("No number specified for start DTMF command")
+            })?))
         } else {
             Ok(Self::End(number))
         }
     }
 }
 
-impl From<DtmfEvent> for gst::Structure {
+impl TryFrom<DtmfCommand> for gst::Structure {
-    fn from(event: DtmfEvent) -> Self {
+    type Error = anyhow::Error;
-        match event {
+
-            DtmfEvent::Start(start) => {
+    fn try_from(event: DtmfCommand) -> anyhow::Result<Self> {
-                gst::Structure::builder("dtmf-event")
+        let structure = match event {
+            DtmfCommand::Start(number) => gst::Structure::builder("dtmf-event")
                 .field("type", 1)
                 .field("start", true)
-                    .field("number", start.number)
+                .field("number", number)
-                    .field("volume", start.volume)
+                .field("volume", 36)
-                    .build()
+                .build(),
-            }
+            DtmfCommand::End(number) => {
-            DtmfEvent::End(number) => {
+                let Some(number) = number else {
+                    anyhow::bail!("Cannot send end DTMF command without a specified number");
+                };
                 gst::Structure::builder("dtmf-event")
                     .field("type", 1)
                     .field("start", false)
                     .field("number", number)
                     .build()
             }
-        }
+        };
+        Ok(structure)
     }
 }
 
 fn main() -> Result<()> {
     gst::init()?;
-    //videotestsrc ! videoconvert ! timeoverlay shaded-background=true ! gtksink
 
     let pipeline = gst::parse_launch(
         r#"
 
     dtmfsrc name=src ! mix.
-    audiotestsrc freq=20 ! audiomixer name=mix ! tee name=t ! queue ! audioconvert ! autoaudiosink name=audiosink
+    audiotestsrc freq=0 ! audiomixer name=mix ! dtmfdetect ! audioconvert ! autoaudiosink name=audiosink
 
     "#,
     )?
     .downcast::<gst::Pipeline>()
     .unwrap();
-    // t.! queue ! audioconvert ! dtmfdetect
 
     let context = glib::MainContext::default();
     let main_loop = glib::MainLoop::new(Some(&context), false);
 
-    pipeline.set_state(gst::State::Playing)?;
-
     let bus = pipeline.bus().unwrap();
     bus.add_watch({
         let main_loop = main_loop.clone();
@@ -108,38 +119,50 @@ fn main() -> Result<()> {
             match msg.view() {
                 MessageView::Eos(..) => main_loop.quit(),
                 MessageView::Error(err) => {
-                    gst::error!(CAT, obj: &err.src().unwrap(),
+                    gst::error!(CAT, obj: err.src().unwrap(),
                         "Error from {:?}: {} ({:?})",
                         err.src().map(|s| s.path_string()),
                         err.error(),
                         err.debug()
                     );
                     main_loop.quit();
-                },
+                }
                 MessageView::Element(element) => {
-                    let dtmf_event = match DtmfEvent::try_from(element.structure().unwrap()) {
+                    match element.structure().unwrap().name().as_str() {
+                        "dtmf-event" => {
+                            let dtmf_event = DtmfEvent::try_from(element.structure().unwrap())
+                                .expect("Failed to parse DTMF event");
+                            gst::info!(CAT, "Detected DTMF event: {:?}", dtmf_event);
+                        }
+                        "dtmf-event-processed" => {
+                            let dtmf_cmd = match DtmfCommand::try_from(element.structure().unwrap())
+                            {
                                 Ok(ev) => ev,
                                 Err(err) => {
-                            gst::info!(CAT, "Failed to parse DTMF event {:?} with error: {:?}", element.structure().unwrap(), err);
+                                    gst::error!(
+                                        CAT,
+                                        "Failed to parse DTMF event {:?} with error: {:?}",
+                                        element.structure().unwrap(),
+                                        err
+                                    );
                                     return glib::Continue(true);
                                 }
                             };
-                    match element.structure().unwrap().name().as_str() {
+                            match dtmf_cmd {
-                        "dtmf-event" => {
+                                DtmfCommand::Start(number) => {
-                            gst::info!(CAT, "Detected DTMF event");
+                                    gst::info!(CAT, "Processed DTMF event {}", number);
                                 }
-                        "dtmf-event-processed" => {
+                                DtmfCommand::End(_) => {
-                            match dtmf_event {
+                                    gst::info!(CAT, "Processed ending DTMF event");
-                                DtmfEvent::Start(start) => {
-                                    gst::info!(CAT, "DTMF event {} processed", start.number);
-                                }
-                                DtmfEvent::End(number) => {
-                                    gst::info!(CAT, "Ending DTMF event {number} processed");
                                 }
                             }
                         }
                         _ => {
-                            gst::info!(CAT, "Received unknown event: {:?}", element.structure().unwrap().name());
+                            gst::error!(
+                                CAT,
+                                "Received unknown event: {:?}",
+                                element.structure().unwrap().name()
+                            );
                         }
                     }
                 }
@@ -160,16 +183,34 @@ fn main() -> Result<()> {
 
             let source = pipeline.by_name("src").unwrap();
 
+            // wait pipeline to be running
+            let bus = pipeline.bus().unwrap();
+            while let Some(msg) = bus.timed_pop(None) {
+                use gst::MessageView;
+                if let MessageView::StateChanged(state_changed) = msg.view() {
+                    if state_changed.src().unwrap() == &source
+                        && state_changed.current() == gst::State::Playing
+                    {
+                        break;
+                    }
+                }
+            }
+            gst::info!(CAT, "Pipeline is running");
+
             let mut rng = rand::thread_rng();
             loop {
-                let dtmf_event = DtmfEvent::start(rng.gen_range(0..15));
+                let dtmf_cmd = DtmfCommand::start(rng.gen_range(0..15));
 
-                source.send_event(gst::event::CustomUpstream::new(dtmf_event.clone().into()));
+                source.send_event(gst::event::CustomUpstream::new(
-                gst::info!(CAT, "Sent DTMF event {:?}", dtmf_event);
+                    dtmf_cmd.try_into().unwrap(),
+                ));
+                gst::info!(CAT, "Sent DTMF event {:?}", dtmf_cmd);
 
                 sleep(std::time::Duration::from_millis(10));
 
-                source.send_event(gst::event::CustomUpstream::new(dtmf_event.end().into()));
+                source.send_event(gst::event::CustomUpstream::new(
+                    dtmf_cmd.end().try_into().unwrap(),
+                ));
 
                 sleep(std::time::Duration::from_millis(1000));
             }
@@ -189,6 +230,8 @@ fn main() -> Result<()> {
         }
     });
 
+    pipeline.set_state(gst::State::Playing)?;
+
     ctrlc::set_handler({
         let main_loop = main_loop.clone();
         move || {