2021-11-17 15:15:16 +00:00
|
|
|
// pipeline.rs
|
|
|
|
//
|
|
|
|
// Copyright 2021 Stéphane Cerveau <scerveau@collabora.com>
|
|
|
|
//
|
|
|
|
// This program is free software: you can redistribute it and/or modify
|
|
|
|
// it under the terms of the GNU General Public License as published by
|
|
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
|
|
// (at your option) any later version.
|
|
|
|
//
|
|
|
|
// This program is distributed in the hope that it will be useful,
|
|
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
|
|
// GNU General Public License for more details.
|
|
|
|
//
|
|
|
|
// You should have received a copy of the GNU General Public License
|
|
|
|
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
//
|
|
|
|
// SPDX-License-Identifier: GPL-3.0-only
|
2022-01-20 14:06:14 +00:00
|
|
|
|
2021-12-15 12:25:13 +00:00
|
|
|
use crate::graphmanager::{GraphView, Node, NodeType, PortDirection};
|
2022-01-04 16:48:32 +00:00
|
|
|
use crate::logger;
|
2022-01-20 14:06:14 +00:00
|
|
|
use crate::ui::message as GPSMessage;
|
2022-01-04 16:48:32 +00:00
|
|
|
use crate::GPS_INFO;
|
|
|
|
|
2022-01-17 13:39:26 +00:00
|
|
|
use gst::glib;
|
2021-11-17 15:15:16 +00:00
|
|
|
use gst::prelude::*;
|
|
|
|
use gstreamer as gst;
|
2021-12-03 13:47:20 +00:00
|
|
|
use std::cell::{Cell, RefCell};
|
2021-12-08 13:15:22 +00:00
|
|
|
use std::collections::HashMap;
|
2021-12-03 13:47:20 +00:00
|
|
|
use std::fmt;
|
|
|
|
use std::ops;
|
|
|
|
use std::rc::{Rc, Weak};
|
2021-11-17 15:15:16 +00:00
|
|
|
|
2021-12-03 13:47:20 +00:00
|
|
|
#[derive(Debug, Clone, Copy, PartialEq)]
|
|
|
|
pub enum PipelineState {
|
|
|
|
Playing,
|
|
|
|
Paused,
|
|
|
|
Stopped,
|
|
|
|
}
|
|
|
|
|
|
|
|
impl fmt::Display for PipelineState {
|
|
|
|
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
|
|
|
|
write!(f, "{:?}", self)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Debug, Clone)]
|
|
|
|
pub struct Pipeline(Rc<PipelineInner>);
|
|
|
|
|
|
|
|
// Deref into the contained struct to make usage a bit more ergonomic
|
|
|
|
impl ops::Deref for Pipeline {
|
|
|
|
type Target = PipelineInner;
|
|
|
|
|
|
|
|
fn deref(&self) -> &PipelineInner {
|
|
|
|
&*self.0
|
|
|
|
}
|
2021-11-17 15:15:16 +00:00
|
|
|
}
|
|
|
|
|
2021-12-03 13:47:20 +00:00
|
|
|
#[derive(Debug, Clone)]
|
|
|
|
pub struct PipelineWeak(Weak<PipelineInner>);
|
|
|
|
|
|
|
|
impl PipelineWeak {
|
|
|
|
pub fn upgrade(&self) -> Option<Pipeline> {
|
|
|
|
self.0.upgrade().map(Pipeline)
|
2021-11-17 15:15:16 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-12-03 13:47:20 +00:00
|
|
|
#[derive(Debug)]
|
|
|
|
pub struct PipelineInner {
|
|
|
|
pipeline: RefCell<Option<gst::Pipeline>>,
|
|
|
|
current_state: Cell<PipelineState>,
|
|
|
|
}
|
|
|
|
|
2021-11-17 15:15:16 +00:00
|
|
|
impl Pipeline {
|
2022-01-11 13:38:18 +00:00
|
|
|
pub fn new() -> anyhow::Result<Self> {
|
2021-12-03 13:47:20 +00:00
|
|
|
let pipeline = Pipeline(Rc::new(PipelineInner {
|
|
|
|
pipeline: RefCell::new(None),
|
|
|
|
current_state: Cell::new(PipelineState::Stopped),
|
|
|
|
}));
|
|
|
|
|
|
|
|
Ok(pipeline)
|
|
|
|
}
|
|
|
|
|
2022-01-11 13:38:18 +00:00
|
|
|
pub fn create_pipeline(&self, description: &str) -> anyhow::Result<()> {
|
2022-01-04 16:48:32 +00:00
|
|
|
GPS_INFO!("Creating pipeline {}", description);
|
2021-12-03 13:47:20 +00:00
|
|
|
|
2022-01-11 13:38:18 +00:00
|
|
|
// Create pipeline from the description
|
2021-12-03 13:47:20 +00:00
|
|
|
let pipeline = gst::parse_launch(&description.to_string())?;
|
2022-01-10 17:33:08 +00:00
|
|
|
if let Ok(pipeline) = pipeline.downcast::<gst::Pipeline>() {
|
|
|
|
let bus = pipeline.bus().expect("Pipeline had no bus");
|
|
|
|
let pipeline_weak = self.downgrade();
|
|
|
|
bus.add_watch_local(move |_bus, msg| {
|
|
|
|
let pipeline = upgrade_weak!(pipeline_weak, glib::Continue(false));
|
|
|
|
pipeline.on_pipeline_message(msg);
|
|
|
|
glib::Continue(true)
|
|
|
|
})?;
|
2021-12-03 13:47:20 +00:00
|
|
|
|
2022-01-10 17:33:08 +00:00
|
|
|
*self.pipeline.borrow_mut() = Some(pipeline);
|
|
|
|
/* start playing */
|
|
|
|
} else {
|
2022-01-11 13:38:18 +00:00
|
|
|
GPS_ERROR!("Can not create a proper pipeline from gstreamer parse_launch");
|
2022-01-10 17:33:08 +00:00
|
|
|
}
|
2021-12-03 13:47:20 +00:00
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2022-01-11 13:38:18 +00:00
|
|
|
pub fn start_pipeline(
|
|
|
|
&self,
|
|
|
|
graphview: &GraphView,
|
|
|
|
new_state: PipelineState,
|
|
|
|
) -> anyhow::Result<PipelineState> {
|
|
|
|
if self.state() == PipelineState::Stopped {
|
|
|
|
self.create_pipeline(&self.render_gst_launch(graphview))
|
|
|
|
.map_err(|err| {
|
|
|
|
GPS_ERROR!("Unable to start a pipeline: {}", err);
|
2022-01-12 17:46:07 +00:00
|
|
|
err
|
|
|
|
})?;
|
2022-01-11 13:38:18 +00:00
|
|
|
}
|
|
|
|
|
2022-01-12 17:46:07 +00:00
|
|
|
self.set_state(new_state).map_err(|error| {
|
|
|
|
GPS_ERROR!("Unable to change state {}", error);
|
|
|
|
error
|
|
|
|
})?;
|
|
|
|
|
2022-01-11 13:38:18 +00:00
|
|
|
Ok(self.state())
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn set_state(&self, new_state: PipelineState) -> anyhow::Result<PipelineState> {
|
2021-12-03 13:47:20 +00:00
|
|
|
if let Some(pipeline) = self.pipeline.borrow().to_owned() {
|
2022-01-11 13:38:18 +00:00
|
|
|
match new_state {
|
2021-12-03 13:47:20 +00:00
|
|
|
PipelineState::Playing => pipeline.set_state(gst::State::Playing)?,
|
|
|
|
PipelineState::Paused => pipeline.set_state(gst::State::Paused)?,
|
|
|
|
PipelineState::Stopped => {
|
|
|
|
pipeline.set_state(gst::State::Null)?;
|
|
|
|
gst::StateChangeSuccess::Success
|
|
|
|
}
|
|
|
|
};
|
2022-01-11 13:38:18 +00:00
|
|
|
self.current_state.set(new_state);
|
2021-12-03 13:47:20 +00:00
|
|
|
}
|
2022-01-11 13:38:18 +00:00
|
|
|
Ok(new_state)
|
2021-12-03 13:47:20 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
pub fn state(&self) -> PipelineState {
|
|
|
|
self.current_state.get()
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn downgrade(&self) -> PipelineWeak {
|
|
|
|
PipelineWeak(Rc::downgrade(&self.0))
|
|
|
|
}
|
|
|
|
|
|
|
|
fn on_pipeline_message(&self, msg: &gst::MessageRef) {
|
|
|
|
use gst::MessageView;
|
|
|
|
match msg.view() {
|
|
|
|
MessageView::Error(err) => {
|
2022-01-10 17:33:08 +00:00
|
|
|
GPS_ERROR!(
|
|
|
|
"Error from {:?}: {} ({:?})",
|
|
|
|
err.src().map(|s| s.path_string()),
|
|
|
|
err.error(),
|
|
|
|
err.debug()
|
|
|
|
);
|
2022-01-20 14:06:14 +00:00
|
|
|
GPSMessage::display_error_dialog(
|
2021-12-03 13:47:20 +00:00
|
|
|
false,
|
|
|
|
format!(
|
|
|
|
"Error from {:?}: {} ({:?})",
|
|
|
|
err.src().map(|s| s.path_string()),
|
|
|
|
err.error(),
|
|
|
|
err.debug()
|
|
|
|
)
|
|
|
|
.as_str(),
|
|
|
|
);
|
|
|
|
}
|
|
|
|
MessageView::Application(msg) => match msg.structure() {
|
|
|
|
// Here we can send ourselves messages from any thread and show them to the user in
|
|
|
|
// the UI in case something goes wrong
|
|
|
|
Some(s) if s.name() == "warning" => {
|
|
|
|
let text = s.get::<&str>("text").expect("Warning message without text");
|
2022-01-10 17:33:08 +00:00
|
|
|
GPS_WARN!("{}", text);
|
2021-12-03 13:47:20 +00:00
|
|
|
}
|
|
|
|
_ => (),
|
|
|
|
},
|
|
|
|
_ => (),
|
|
|
|
};
|
2021-11-17 15:15:16 +00:00
|
|
|
}
|
|
|
|
|
2021-12-15 12:25:13 +00:00
|
|
|
// Render graph methods
|
2022-01-11 13:38:18 +00:00
|
|
|
fn process_gst_node(
|
2021-12-15 12:25:13 +00:00
|
|
|
&self,
|
|
|
|
graphview: &GraphView,
|
|
|
|
node: &Node,
|
2022-01-10 13:17:08 +00:00
|
|
|
elements: &mut HashMap<String, String>,
|
2021-12-15 12:25:13 +00:00
|
|
|
mut description: String,
|
|
|
|
) -> String {
|
|
|
|
let unique_name = node.unique_name();
|
2022-01-10 13:17:08 +00:00
|
|
|
description.push_str(&format!("{} name={} ", node.name(), unique_name));
|
|
|
|
elements.insert(unique_name.clone(), unique_name.clone());
|
2021-12-15 12:25:13 +00:00
|
|
|
for (name, value) in node.properties().iter() {
|
2022-01-20 17:18:52 +00:00
|
|
|
//This allow to have an index in front of a property such as an enum.
|
|
|
|
let value = value.split_once(':').unwrap_or((value, value));
|
2022-01-12 17:45:19 +00:00
|
|
|
if !node.hidden_property(name) {
|
2022-01-20 17:18:52 +00:00
|
|
|
description.push_str(&format!("{}={} ", name, value.1));
|
2022-01-12 17:45:19 +00:00
|
|
|
}
|
2021-12-15 12:25:13 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
let ports = node.all_ports(PortDirection::Output);
|
|
|
|
let n_ports = ports.len();
|
|
|
|
for port in ports {
|
|
|
|
if let Some((_port_to, node_to)) = graphview.port_connected_to(port.id()) {
|
|
|
|
if n_ports > 1 {
|
2022-01-10 13:17:08 +00:00
|
|
|
description.push_str(&format!("{}. ! ", unique_name));
|
2021-12-15 12:25:13 +00:00
|
|
|
} else {
|
2022-01-10 13:17:08 +00:00
|
|
|
description.push_str("! ");
|
2021-12-15 12:25:13 +00:00
|
|
|
}
|
2022-01-18 09:51:06 +00:00
|
|
|
if let Some(node) = graphview.node(node_to) {
|
2022-01-10 13:17:08 +00:00
|
|
|
if elements.contains_key(&node.unique_name()) {
|
|
|
|
description.push_str(&format!("{}. ", node.unique_name()));
|
|
|
|
} else {
|
|
|
|
description =
|
2022-01-11 13:38:18 +00:00
|
|
|
self.process_gst_node(graphview, &node, elements, description.clone());
|
2022-01-10 13:17:08 +00:00
|
|
|
}
|
2021-12-15 12:25:13 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
description
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn render_gst_launch(&self, graphview: &GraphView) -> String {
|
2022-01-10 13:17:08 +00:00
|
|
|
let source_nodes = graphview.all_nodes(NodeType::Source);
|
|
|
|
let mut elements: HashMap<String, String> = HashMap::new();
|
2021-12-15 12:25:13 +00:00
|
|
|
let mut description = String::from("");
|
2022-01-10 13:17:08 +00:00
|
|
|
for source_node in source_nodes {
|
|
|
|
description =
|
2022-01-11 13:38:18 +00:00
|
|
|
self.process_gst_node(graphview, &source_node, &mut elements, description.clone());
|
2021-12-15 12:25:13 +00:00
|
|
|
}
|
|
|
|
description
|
|
|
|
}
|
2021-11-17 15:15:16 +00:00
|
|
|
}
|
2021-12-03 13:47:20 +00:00
|
|
|
|
|
|
|
impl Drop for PipelineInner {
|
|
|
|
fn drop(&mut self) {
|
|
|
|
// TODO: If a recording is currently running we would like to finish that first
|
|
|
|
// before quitting the pipeline and shutting down the pipeline.
|
|
|
|
if let Some(pipeline) = self.pipeline.borrow().to_owned() {
|
|
|
|
// We ignore any errors here
|
|
|
|
let _ = pipeline.set_state(gst::State::Null);
|
|
|
|
|
|
|
|
// Remove the message watch from the bus
|
|
|
|
let bus = pipeline.bus().expect("Pipeline had no bus");
|
|
|
|
let _ = bus.remove_watch();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|