<commit_before>// Copyright (C) 2019 <NAME> <<EMAIL>>
//
// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Library General Public
// License as published by the Free Software Foundation; either
// version 2 of the License, or (at your option) any later version.
//
// This library 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
// Library General Public License for more details.
//
// You should have received a copy of the GNU Library General Public
// License along with this library; if not, write to the
// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
// Boston, MA 02110-1335, USA.

#[cfg(not(feature = "v1_18"))]
use self::gst_base::prelude::*;
#[cfg(not(feature = "v1_18"))]
use self::gst_base::subclass::prelude::*;
#[cfg(not(feature = "v1_18"))]
use super::gst_base_compat as gst_base;

#[cfg(feature = "v1_18")]
use gst_base::prelude::*;
#[cfg(feature = "v1_18")]
use gst_base::subclass::prelude::*;

use glib::subclass;
use glib::subclass::prelude::*;
use gst::subclass::prelude::*;

use once_cell::sync::Lazy;

use std::sync::{Mutex, RwLock};

struct FallbackSwitch {
    sinkpad: gst_base::AggregatorPad,
    fallback_sinkpad: RwLock<Option<gst_base::AggregatorPad>>,
    active_sinkpad: Mutex<gst::Pad>,
    output_state: Mutex<OutputState>,
    pad_states: RwLock<PadStates>,
    settings: Mutex<Settings>,
}

static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
    gst::DebugCategory::new(
        "fallbackswitch",
        gst::DebugColorFlags::empty(),
        Some("Fallback switch Element"),
    )
});

#[derive(Debug)]
struct OutputState {
    last_sinkpad_time: gst::ClockTime,
}

#[derive(Debug, Default)]
struct PadStates {
    sinkpad: PadState,
    fallback_sinkpad: Option<PadState>,
}

#[derive(Debug, Default)]
struct PadState {
    caps: Option<gst::Caps>,
    audio_info: Option<gst_audio::AudioInfo>,
    video_info: Option<gst_video::VideoInfo>,
}

const DEFAULT_TIMEOUT: u64 = 5 * gst::SECOND_VAL;

#[derive(Debug, Clone)]
struct Settings {
    timeout: gst::ClockTime,
}

impl Default for OutputState {
    fn default() -> Self {
        OutputState {
            last_sinkpad_time: gst::CLOCK_TIME_NONE,
        }
    }
}

impl Default for Settings {
    fn default() -> Self {
        Settings {
            timeout: DEFAULT_TIMEOUT.into(),
        }
    }
}

static PROPERTIES: [subclass::Property; 2] = [
    subclass::Property("timeout", |name| {
        glib::ParamSpec::uint64(
            name,
            "Timeout",
            "Timeout in nanoseconds",
            0,
            std::u64::MAX,
            DEFAULT_TIMEOUT,
            glib::ParamFlags::READWRITE,
        )
    }),
    subclass::Property("active-pad", |name| {
        glib::ParamSpec::object(
            name,
            "Active Pad",
            "Currently active pad",
            gst::Pad::static_type(),
            glib::ParamFlags::READABLE,
        )
    }),
];

impl FallbackSwitch {
    fn handle_main_buffer(
        &self,
        agg: &gst_base::Aggregator,
        state: &mut OutputState,
        mut buffer: gst::Buffer,
        fallback_sinkpad: Option<&gst_base::AggregatorPad>,
    ) -> Result<Option<(gst::Buffer, gst::Caps, bool)>, gst::FlowError> {
        // If we got a buffer on the sinkpad just handle it
        gst_debug!(CAT, obj: agg, "Got buffer on sinkpad {:?}", buffer);

        if buffer.get_pts().is_none() {
            gst_error!(CAT, obj: agg, "Only buffers with PTS supported");
            return Err(gst::FlowError::Error);
        }

        let segment = self
            .sinkpad
            .get_segment()
            .downcast::<gst::ClockTime>()
            .map_err(|_| {
                gst_error!(CAT, obj: agg, "Only TIME segments supported");
                gst::FlowError::Error
            })?;

        {
            // FIXME: This will not work correctly for negative DTS
            let buffer = buffer.make_mut();
            buffer.set_pts(segment.to_running_time(buffer.get_pts()));
            buffer.set_dts(segment.to_running_time(buffer.get_dts()));
        }

        let mut active_sinkpad = self.active_sinkpad.lock().unwrap();
        let pad_change = &*active_sinkpad != self.sinkpad.upcast_ref::<gst::Pad>();
        if pad_change {
            if buffer.get_flags().contains(gst::BufferFlags::DELTA_UNIT) {
                gst_info!(
                    CAT,
                    obj: agg,
                    "Can't change back to sinkpad, waiting for keyframe"
                );
                self.sinkpad.push_event(
                    gst_video::new_upstream_force_key_unit_event()
                        .all_headers(true)
                        .build(),
                );
                return Ok(None);
            }

            gst_info!(CAT, obj: agg, "Active pad changed to sinkpad");
            *active_sinkpad = self.sinkpad.clone().upcast();
        }
        drop(active_sinkpad);

        state.last_sinkpad_time = segment.to_running_time(buffer.get_dts_or_pts());

        // Drop all older buffers from the fallback sinkpad
        if let Some(fallback_sinkpad) = fallback_sinkpad {
            let fallback_segment = self
                .sinkpad<commit_msg>fallbackswitch/fallbackswitch: Don't convert buffer PTS/DTS twice to running time<commit_after>// Copyright (C) 2019 <NAME> <<EMAIL>>
//
// This library is free software; you can redistribute it and/or
// modify it under the terms of the GNU Library General Public
// License as published by the Free Software Foundation; either
// version 2 of the License, or (at your option) any later version.
//
// This library 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
// Library General Public License for more details.
//
// You should have received a copy of the GNU Library General Public
// License along with this library; if not, write to the
// Free Software Foundation, Inc., 51 Franklin Street, Suite 500,
// Boston, MA 02110-1335, USA.

#[cfg(not(feature = "v1_18"))]
use self::gst_base::prelude::*;
#[cfg(not(feature = "v1_18"))]
use self::gst_base::subclass::prelude::*;
#[cfg(not(feature = "v1_18"))]
use super::gst_base_compat as gst_base;

#[cfg(feature = "v1_18")]
use gst_base::prelude::*;
#[cfg(feature = "v1_18")]
use gst_base::subclass::prelude::*;

use glib::subclass;
use glib::subclass::prelude::*;
use gst::subclass::prelude::*;

use once_cell::sync::Lazy;

use std::sync::{Mutex, RwLock};

struct FallbackSwitch {
    sinkpad: gst_base::AggregatorPad,
    fallback_sinkpad: RwLock<Option<gst_base::AggregatorPad>>,
    active_sinkpad: Mutex<gst::Pad>,
    output_state: Mutex<OutputState>,
    pad_states: RwLock<PadStates>,
    settings: Mutex<Settings>,
}

static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
    gst::DebugCategory::new(
        "fallbackswitch",
        gst::DebugColorFlags::empty(),
        Some("Fallback switch Element"),
    )
});

#[derive(Debug)]
struct OutputState {
    last_sinkpad_time: gst::ClockTime,
}

#[derive(Debug, Default)]
struct PadStates {
    sinkpad: PadState,
    fallback_sinkpad: Option<PadState>,
}

#[derive(Debug, Default)]
struct PadState {
    caps: Option<gst::Caps>,
    audio_info: Option<gst_audio::AudioInfo>,
    video_info: Option<gst_video::VideoInfo>,
}

const DEFAULT_TIMEOUT: u64 = 5 * gst::SECOND_VAL;

#[derive(Debug, Clone)]
struct Settings {
    timeout: gst::ClockTime,
}

impl Default for OutputState {
    fn default() -> Self {
        OutputState {
            last_sinkpad_time: gst::CLOCK_TIME_NONE,
        }
    }
}

impl Default for Settings {
    fn default() -> Self {
        Settings {
            timeout: DEFAULT_TIMEOUT.into(),
        }
    }
}

static PROPERTIES: [subclass::Property; 2] = [
    subclass::Property("timeout", |name| {
        glib::ParamSpec::uint64(
            name,
            "Timeout",
            "Timeout in nanoseconds",
            0,
            std::u64::MAX,
            DEFAULT_TIMEOUT,
            glib::ParamFlags::READWRITE,
        )
    }),
    subclass::Property("active-pad", |name| {
        glib::ParamSpec::object(
            name,
            "Active Pad",
            "Currently active pad",
            gst::Pad::static_type(),
            glib::ParamFlags::READABLE,
        )
    }),
];

impl FallbackSwitch {
    fn handle_main_buffer(
        &self,
        agg: &gst_base::Aggregator,
        state: &mut OutputState,
        mut buffer: gst::Buffer,
        fallback_sinkpad: Option<&gst_base::AggregatorPad>,
    ) -> Result<Option<(gst::Buffer, gst::Caps, bool)>, gst::FlowError> {
        // If we got a buffer on the sinkpad just handle it
        gst_debug!(CAT, obj: agg, "Got buffer on sinkpad {:?}", buffer);

        if buffer.get_pts().is_none() {
            gst_error!(CAT, obj: agg, "Only buffers with PTS supported");
            return Err(gst::FlowError::Error);
        }

        let segment = self
            .sinkpad
            .get_segment()
            .downcast::<gst::ClockTime>()
            .map_err(|_| {
                gst_error!(CAT, obj: agg, "Only TIME segments supported");
                gst::FlowError::Error
            })?;

        {
            // FIXME: This will not work correctly for negative DTS
            let buffer = buffer.make_mut();
            buffer.set_pts(segment.to_running_time(buffer.get_pts()));
            buffer.set_dts(segment.to_running_time(buffer.get_dts()));
        }

        let mut active_sinkpad = self.active_sinkpad.lock().unwrap();
        let pad_change = &*active_sinkpad != self.sinkpad.upcast_ref::<gst::Pad>();
        if pad_change {
            if buffer.get_flags().contains(gst::BufferFlags::DELTA_UNIT) {
                gst_info!(
                    CAT,
                    obj: agg,
                    "Can't change back to sinkpad, waiting for keyframe"
                );
                self.sinkpad.push_event(
                    gst_video::new_upstream_force_key_unit_event()
                        .all_headers(true)
                        .build(),
                );
                return Ok(None);
            }

            gst_info!(CAT, obj: agg, "Active pad changed to sinkpad");
            *active_sinkpad = self.sinkpad.clone().upcast();
        }
        drop(active_sinkpad);

        state.last_sinkpad_time = buffer.get_dts_or_pts();

        // Drop all older buffers from the fallback sinkpad
        if let Some(fallback_sinkpad) = fallback_sinkpad {
            let fallback_segment = self
                .sinkpad
{'fraction_non_alphanumeric': 0.12667467430472143, 'fraction_numerical': 0.00794604083895408, 'mean_word_length': 2.4692307692307693, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '663087', 'n_tokens_mistral': 3430, 'n_tokens_neox': 3356, 'n_words': 803}