audiofx: Add HRTF renderer element

Fixes https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/issues/128

6 files changed, 1304 insertions, 0 deletions

diff --git a/audio/audiofx/Cargo.toml b/audio/audiofx/Cargo.toml
index 6f6c9326..6ee70f61 100644
--- a/audio/audiofx/Cargo.toml
+++ b/audio/audiofx/Cargo.toml
@@ -12,13 +12,16 @@ rust-version = "1.56"
 gst = { package = "gstreamer", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_16"] }
 gst-base = { package = "gstreamer-base", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_16"] }
 gst-audio = { package = "gstreamer-audio", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_16"] }
+anyhow = "1"
 byte-slice-cast = "1.0"
 num-traits = "0.2"
 once_cell = "1.0"
 ebur128 = "0.1"
+hrtf = "0.7"
 nnnoiseless = { version = "0.3", default-features = false }
 smallvec = "1"
 atomic_refcell = "0.1"
+rayon = "1.5"
 
 [lib]
 name = "gstrsaudiofx"
diff --git a/audio/audiofx/examples/hrtfrender.rs b/audio/audiofx/examples/hrtfrender.rs
new file mode 100644
index 00000000..aee8d21c
--- /dev/null
+++ b/audio/audiofx/examples/hrtfrender.rs
@@ -0,0 +1,145 @@
+// Copyright (C) 2021 Tomasz Andrzejak <andreiltd@gmail.com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+use gst::prelude::*;
+
+use anyhow::{bail, Error};
+
+use std::sync::{Arc, Condvar, Mutex};
+use std::{env, thread, time};
+
+// Rotation in radians to apply to object position every 100 ms
+const ROTATION: f32 = 2.0 / 180.0 * std::f32::consts::PI;
+
+fn run() -> Result<(), Error> {
+    gst::init()?;
+    gstrsaudiofx::plugin_register_static()?;
+
+    let args: Vec<String> = env::args().collect();
+
+    // Ircam binaries that the hrtf plugin is using can be downloaded from:
+    // https://github.com/mrDIMAS/hrir_sphere_builder/tree/master/hrtf_base/IRCAM
+    //
+    // Ideally provide a mono input as this example is moving a single object. Otherwise
+    // the input will be downmixed to mono with the audioconvert element.
+    //
+    // e.g.: hrtfrender 'file:///path/to/my/awesome/mono/wav/awesome.wav' IRC_1002_C.bin
+    if args.len() != 3 {
+        bail!("Usage: {} URI HRIR", args[0].clone());
+    }
+
+    let uri = &args[1];
+    let hrir = &args[2];
+
+    let pipeline = gst::parse_launch(&format!(
+        "uridecodebin uri={} ! audioconvert ! audio/x-raw,channels=1 !
+            hrtfrender hrir-file={} name=hrtf ! audioresample ! autoaudiosink",
+        uri, hrir
+    ))?
+    .downcast::<gst::Pipeline>()
+    .expect("type error");
+
+    let hrtf = pipeline.by_name("hrtf").expect("hrtf element not found");
+
+    // At the beginning put an object in front of listener
+    let objs = [gst::Structure::builder("application/spatial-object")
+        .field("x", 0f32)
+        .field("y", 0f32)
+        .field("z", 1f32)
+        .field("distance-gain", 1f32)
+        .build()];
+
+    hrtf.set_property("spatial-objects", gst::Array::new(objs));
+
+    let state_cond = Arc::new((Mutex::new(gst::State::Null), Condvar::new()));
+    let state_cond_clone = Arc::clone(&state_cond);
+
+    thread::spawn(move || {
+        // Wait for the pipeline to start up
+        {
+            let (lock, cvar) = &*state_cond_clone;
+            let mut state = lock.lock().unwrap();
+
+            while *state != gst::State::Playing {
+                state = cvar.wait(state).unwrap();
+            }
+        }
+
+        loop {
+            // get current object position and rotate it clockwise
+            let s = hrtf.property::<gst::Array>("spatial-objects")[0]
+                .get::<gst::Structure>()
+                .expect("type error");
+
+            // positive values are on the right side of a listener
+            let x = s.get::<f32>("x").expect("type error");
+            // elevation, positive value is up
+            let y = s.get::<f32>("y").expect("type error");
+            // positive values are in front of a listener
+            let z = s.get::<f32>("z").expect("type error");
+            // gain
+            let gain = s.get::<f32>("distance-gain").expect("type error");
+
+            // rotate clockwise: https://en.wikipedia.org/wiki/Rotation_matrix
+            let new_x = x * f32::cos(ROTATION) + z * f32::sin(ROTATION);
+            let new_z = -x * f32::sin(ROTATION) + z * f32::cos(ROTATION);
+
+            let objs = [gst::Structure::builder("application/spatial-object")
+                .field("x", &new_x)
+                .field("y", &y)
+                .field("z", &new_z)
+                .field("distance-gain", &gain)
+                .build()];
+
+            hrtf.set_property("spatial-objects", gst::Array::new(objs));
+
+            thread::sleep(time::Duration::from_millis(100));
+        }
+    });
+
+    pipeline.set_state(gst::State::Playing)?;
+
+    let bus = pipeline.bus().unwrap();
+    for msg in bus.iter_timed(gst::ClockTime::NONE) {
+        use gst::MessageView;
+
+        match msg.view() {
+            MessageView::StateChanged(state_changed) => {
+                if state_changed.src().map(|s| s == pipeline).unwrap_or(false)
+                    && state_changed.current() == gst::State::Playing
+                {
+                    let (lock, cvar) = &*state_cond;
+                    let mut state = lock.lock().unwrap();
+
+                    *state = gst::State::Playing;
+                    cvar.notify_one();
+                }
+            }
+            MessageView::Eos(..) => break,
+            MessageView::Error(err) => {
+                println!(
+                    "Error from {:?}: {} ({:?})",
+                    msg.src().map(|s| s.path_string()),
+                    err.error(),
+                    err.debug()
+                );
+                break;
+            }
+            _ => (),
+        }
+    }
+
+    pipeline.set_state(gst::State::Null)?;
+
+    Ok(())
+}
+
+fn main() {
+    match run() {
+        Ok(r) => r,
+        Err(e) => eprintln!("Error! {}", e),
+    }
+}
diff --git a/audio/audiofx/src/hrtfrender/imp.rs b/audio/audiofx/src/hrtfrender/imp.rs
new file mode 100644
index 00000000..21cb6a1f
--- /dev/null
+++ b/audio/audiofx/src/hrtfrender/imp.rs
@@ -0,0 +1,841 @@
+// Copyright (C) 2021 Tomasz Andrzejak <andreiltd@gmail.com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+use gst::glib;
+use gst::prelude::*;
+use gst::subclass::prelude::*;
+use gst::{gst_debug, gst_error, gst_log};
+
+use gst_base::subclass::prelude::*;
+
+use hrtf::{HrirSphere, HrtfContext, HrtfProcessor, Vec3};
+
+use std::io::{Error, ErrorKind};
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex, Weak};
+
+use byte_slice_cast::*;
+use rayon::prelude::*;
+use rayon::{ThreadPool, ThreadPoolBuilder};
+
+use once_cell::sync::Lazy;
+static CAT: Lazy<gst::DebugCategory> = Lazy::new(|| {
+    gst::DebugCategory::new(
+        "hrtfrender",
+        gst::DebugColorFlags::empty(),
+        Some("Head-Related Transfer Function Renderer"),
+    )
+});
+
+static THREAD_POOL: Lazy<Mutex<Weak<ThreadPool>>> = Lazy::new(|| Mutex::new(Weak::new()));
+
+const DEFAULT_INTERPOLATION_STEPS: u64 = 8;
+const DEFAULT_BLOCK_LENGTH: u64 = 512;
+const DEFAULT_DISTANCE_GAIN: f32 = 1.0;
+
+#[derive(Clone, Copy)]
+struct SpatialObject {
+    /// Position values use a left-handed Cartesian coordinate system
+    position: Vec3,
+    /// Object attenuation by distance
+    distance_gain: f32,
+}
+
+impl From<gst::Structure> for SpatialObject {
+    fn from(s: gst::Structure) -> Self {
+        SpatialObject {
+            position: Vec3 {
+                x: s.get("x").expect("type checked upstream"),
+                y: s.get("y").expect("type checked upstream"),
+                z: s.get("z").expect("type checked upstream"),
+            },
+            distance_gain: s.get("distance-gain").expect("type checked upstream"),
+        }
+    }
+}
+
+impl From<SpatialObject> for gst::Structure {
+    fn from(obj: SpatialObject) -> Self {
+        gst::Structure::builder("application/spatial-object")
+            .field("x", obj.position.x)
+            .field("y", obj.position.y)
+            .field("z", obj.position.z)
+            .field("distance-gain", obj.distance_gain)
+            .build()
+    }
+}
+
+impl TryFrom<gst_audio::AudioChannelPosition> for SpatialObject {
+    type Error = gst::FlowError;
+
+    fn try_from(pos: gst_audio::AudioChannelPosition) -> Result<Self, gst::FlowError> {
+        use gst_audio::AudioChannelPosition::*;
+
+        let position = match pos {
+            FrontLeft => Vec3::new(-1.45, 0.0, 2.5),
+            FrontRight => Vec3::new(1.45, 0.0, 2.5),
+            FrontCenter | Mono => Vec3::new(0.0, 0.0, 2.5),
+            Lfe1 | Lfe2 => Vec3::new(0.0, 0.0, 0.0),
+            RearLeft => Vec3::new(-1.45, 0.0, -2.5),
+            RearRight => Vec3::new(1.45, 0.0, -2.5),
+            FrontLeftOfCenter => Vec3::new(-0.72, 0.0, 2.5),
+            FrontRightOfCenter => Vec3::new(0.72, 0.0, 2.5),
+            RearCenter => Vec3::new(0.0, 0.0, -2.5),
+            SideLeft => Vec3::new(-2.5, 0.0, -0.44),
+            SideRight => Vec3::new(2.5, 0.0, -0.44),
+            TopFrontLeft => Vec3::new(-0.72, 2.5, 1.25),
+            TopFrontRight => Vec3::new(0.72, 2.5, 1.25),
+            TopFrontCenter => Vec3::new(0.0, 2.5, 1.25),
+            TopCenter => Vec3::new(0.0, 2.5, 0.0),
+            TopRearLeft => Vec3::new(-0.72, 2.5, -1.25),
+            TopRearRight => Vec3::new(0.72, 2.5, -1.25),
+            TopSideLeft => Vec3::new(-1.25, 2.5, -0.22),
+            TopSideRight => Vec3::new(1.25, 2.5, -0.22),
+            TopRearCenter => Vec3::new(0.0, 2.5, -1.25),
+            BottomFrontCenter => Vec3::new(0.0, -2.5, 1.25),
+            BottomFrontLeft => Vec3::new(-0.72, -2.5, 1.25),
+            BottomFrontRight => Vec3::new(0.72, -2.5, 1.25),
+            WideLeft => Vec3::new(-2.5, 0.0, 1.45),
+            WideRight => Vec3::new(2.5, 0.0, 1.45),
+            SurroundLeft => Vec3::new(-2.5, 0.0, -1.45),
+            SurroundRight => Vec3::new(2.5, 0.0, -1.45),
+            _ => return Err(gst::FlowError::NotSupported),
+        };
+
+        Ok(SpatialObject {
+            position,
+            distance_gain: DEFAULT_DISTANCE_GAIN,
+        })
+    }
+}
+
+#[derive(Clone)]
+struct Settings {
+    interpolation_steps: u64,
+    block_length: u64,
+    spatial_objects: Option<Vec<SpatialObject>>,
+    hrir_raw_bytes: Option<glib::Bytes>,
+    hrir_file_location: Option<String>,
+}
+
+impl Default for Settings {
+    fn default() -> Self {
+        Settings {
+            interpolation_steps: DEFAULT_INTERPOLATION_STEPS,
+            block_length: DEFAULT_BLOCK_LENGTH,
+            spatial_objects: None,
+            hrir_raw_bytes: None,
+            hrir_file_location: None,
+        }
+    }
+}
+
+impl Settings {
+    fn position(&self, channel: usize) -> Result<Vec3, gst::FlowError> {
+        Ok(self
+            .spatial_objects
+            .as_ref()
+            .ok_or(gst::FlowError::NotNegotiated)?[channel]
+            .position)
+    }
+
+    fn distance_gain(&self, channel: usize) -> Result<f32, gst::FlowError> {
+        Ok(self
+            .spatial_objects
+            .as_ref()
+            .ok_or(gst::FlowError::NotNegotiated)?[channel]
+            .distance_gain)
+    }
+
+    fn sphere(&self, rate: u32) -> Result<hrtf::HrirSphere, hrtf::HrtfError> {
+        if let Some(bytes) = &self.hrir_raw_bytes {
+            return HrirSphere::new(bytes.as_byte_slice(), rate);
+        }
+
+        if let Some(path) = &self.hrir_file_location {
+            return HrirSphere::from_file(PathBuf::from(path), rate);
+        }
+
+        Err(Error::new(ErrorKind::Other, "Impulse response not set").into())
+    }
+}
+
+struct ChannelProcessor {
+    prev_left_samples: Vec<f32>,
+    prev_right_samples: Vec<f32>,
+    prev_sample_vector: Option<Vec3>,
+    prev_distance_gain: Option<f32>,
+    indata_scratch: Box<[f32]>,
+    outdata_scratch: Box<[(f32, f32)]>,
+    processor: HrtfProcessor,
+}
+
+struct State {
+    ininfo: gst_audio::AudioInfo,
+    outinfo: gst_audio::AudioInfo,
+    adapter: gst_base::UniqueAdapter,
+    block_samples: usize,
+    channel_processors: Vec<ChannelProcessor>,
+}
+
+impl State {
+    fn input_block_size(&self) -> usize {
+        self.block_samples * self.ininfo.bpf() as usize
+    }
+
+    fn output_block_size(&self) -> usize {
+        self.block_samples * self.outinfo.bpf() as usize
+    }
+
+    fn reset_processors(&mut self) {
+        for cp in self.channel_processors.iter_mut() {
+            cp.prev_left_samples.fill(0.0);
+            cp.prev_right_samples.fill(0.0);
+        }
+    }
+}
+
+#[derive(Default)]
+pub struct HrtfRender {
+    settings: Mutex<Settings>,
+    state: Mutex<Option<State>>,
+    thread_pool: Mutex<Option<Arc<ThreadPool>>>,
+}
+
+#[glib::object_subclass]
+impl ObjectSubclass for HrtfRender {
+    const NAME: &'static str = "HrtfRender";
+    type Type = super::HrtfRender;
+    type ParentType = gst_base::BaseTransform;
+}
+
+impl HrtfRender {
+    fn process(
+        &self,
+        element: &super::HrtfRender,
+        outbuf: &mut gst::BufferRef,
+        state: &mut State,
+        settings: &Settings,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        let mut outbuf =
+            gst_audio::AudioBufferRef::from_buffer_ref_writable(outbuf, &state.outinfo).map_err(
+                |err| {
+                    gst_error!(CAT, obj: element, "Failed to map buffer : {}", err);
+                    gst::FlowError::Error
+                },
+            )?;
+
+        let outdata = outbuf
+            .plane_data_mut(0)
+            .unwrap()
+            .as_mut_slice_of::<f32>()
+            .unwrap();
+
+        // prefill output with zeros so we can mix processed samples into it
+        outdata.fill(0.0);
+
+        let inblksz = state.input_block_size();
+        let channels = state.ininfo.channels();
+
+        let mut written_samples = 0;
+
+        let thread_pool_guard = self.thread_pool.lock().unwrap();
+        let thread_pool = thread_pool_guard.as_ref().ok_or(gst::FlowError::Error)?;
+
+        while state.adapter.available() >= inblksz {
+            let inbuf = state.adapter.take_buffer(inblksz).map_err(|_| {
+                gst_error!(CAT, obj: element, "Failed to map buffer");
+                gst::FlowError::Error
+            })?;
+
+            let inbuf = gst_audio::AudioBuffer::from_buffer_readable(inbuf, &state.ininfo)
+                .map_err(|_| {
+                    gst_error!(CAT, obj: element, "Failed to map buffer");
+                    gst::FlowError::Error
+                })?;
+
+            let indata = inbuf.plane_data(0).unwrap().as_slice_of::<f32>().unwrap();
+
+            thread_pool.install(|| -> Result<(), gst::FlowError> {
+                state
+                    .channel_processors
+                    .par_iter_mut()
+                    .enumerate()
+                    .try_for_each(|(i, cp)| -> Result<(), gst::FlowError> {
+                        let new_distance_gain = settings.distance_gain(i)?;
+                        let new_sample_vector = settings.position(i)?;
+
+                        // Convert to Right Handed, this is what HRTF crate expects
+                        let new_sample_vector = Vec3 {
+                            z: new_sample_vector.z * -1.0,
+                            ..new_sample_vector
+                        };
+
+                        // Deinterleave single channel to scratch buffer
+                        for (x, y) in Iterator::zip(
+                            indata.iter().skip(i).step_by(channels as usize),
+                            cp.indata_scratch.iter_mut(),
+                        ) {
+                            *y = *x;
+                        }
+
+                        cp.processor.process_samples(HrtfContext {
+                            source: &cp.indata_scratch,
+                            output: &mut cp.outdata_scratch,
+                            new_sample_vector,
+                            new_distance_gain,
+                            prev_sample_vector: cp.prev_sample_vector.unwrap_or(new_sample_vector),
+                            prev_distance_gain: cp.prev_distance_gain.unwrap_or(new_distance_gain),
+                            prev_left_samples: &mut cp.prev_left_samples,
+                            prev_right_samples: &mut cp.prev_right_samples,
+                        });
+
+                        cp.prev_sample_vector = Some(new_sample_vector);
+                        cp.prev_distance_gain = Some(new_distance_gain);
+
+                        Ok(())
+                    })
+            })?;
+
+            // unpack output scratch to output buffer
+            state.channel_processors.iter_mut().for_each(|cp| {
+                for (x, y) in Iterator::zip(
+                    cp.outdata_scratch.iter(),
+                    outdata[2 * written_samples..].chunks_exact_mut(2),
+                ) {
+                    y[0] += x.0;
+                    y[1] += x.1;
+                }
+
+                // HRTF is mixing processed samples with samples in output buffer, we need to
+                // reset scratch so it is not mixed with the next frame
+                cp.outdata_scratch.fill((0.0, 0.0));
+            });
+
+            written_samples += state.block_samples;
+        }
+
+        // we only support stereo output, we can assert that we filled the whole
+        // output buffer with stereo frames
+        assert_eq!(outdata.len(), written_samples * 2);
+
+        Ok(gst::FlowSuccess::Ok)
+    }
+
+    fn drain(&self, element: &super::HrtfRender) -> Result<gst::FlowSuccess, gst::FlowError> {
+        let settings = &self.settings.lock().unwrap();
+
+        let mut state_guard = self.state.lock().unwrap();
+        let state = state_guard.as_mut().ok_or(gst::FlowError::NotNegotiated)?;
+
+        let avail = state.adapter.available();
+
+        if avail == 0 {
+            return Ok(gst::FlowSuccess::Ok);
+        }
+
+        let inblksz = state.input_block_size();
+        let outblksz = state.output_block_size();
+        let inbpf = state.ininfo.bpf() as usize;
+        let outbpf = state.outinfo.bpf() as usize;
+
+        let inputsz = inblksz - avail;
+        let outputsz = avail / inbpf * outbpf;
+
+        let mut inbuf = gst::Buffer::with_size(inputsz).map_err(|_| gst::FlowError::Error)?;
+        let inbuf_mut = inbuf.get_mut().ok_or(gst::FlowError::Error)?;
+
+        let mut map = inbuf_mut
+            .map_writable()
+            .map_err(|_| gst::FlowError::Error)?;
+        let data = map
+            .as_mut_slice_of::<f32>()
+            .map_err(|_| gst::FlowError::Error)?;
+
+        data.fill(0.0);
+        drop(map);
+
+        let (pts, offset, duration) = {
+            let samples_to_time = |samples: u64| {
+                samples
+                    .mul_div_round(*gst::ClockTime::SECOND, state.ininfo.rate() as u64)
+                    .map(gst::ClockTime::from_nseconds)
+            };
+
+            let (prev_pts, distance) = state.adapter.prev_pts();
+            let distance_samples = distance / inbpf as u64;
+            let pts = prev_pts.opt_add(samples_to_time(distance_samples));
+
+            let (prev_offset, _) = state.adapter.prev_offset();
+            let offset = prev_offset.checked_add(distance_samples).unwrap_or(0);
+
+            let duration_samples = outputsz / outbpf as usize;
+            let duration = samples_to_time(duration_samples as u64);
+
+            (pts, offset, duration)
+        };
+
+        state.adapter.push(inbuf);
+
+        let mut outbuf = gst::Buffer::with_size(outblksz).map_err(|_| gst::FlowError::Error)?;
+        let outbuf_mut = outbuf.get_mut().unwrap();
+
+        self.process(element, outbuf_mut, state, settings)?;
+
+        outbuf_mut.set_size(outputsz);
+        outbuf_mut.set_pts(pts);
+        outbuf_mut.set_offset(offset);
+        outbuf_mut.set_duration(duration);
+
+        let srcpad = element.static_pad("src").unwrap();
+
+        state.reset_processors();
+
+        drop(state_guard);
+        srcpad.push(outbuf)
+    }
+}
+
+impl ObjectImpl for HrtfRender {
+    fn properties() -> &'static [glib::ParamSpec] {
+        static PROPERTIES: Lazy<Vec<glib::ParamSpec>> = Lazy::new(|| {
+            vec![
+                glib::ParamSpecBoxed::new(
+                    "hrir-raw",
+                    "Head Transform Impulse Response",
+                    "Head Transform Impulse Response raw bytes",
+                    glib::Bytes::static_type(),
+                    glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
+                ),
+                glib::ParamSpecString::new(
+                    "hrir-file",
+                    "Head Transform Impulse Response",
+                    "Head Transform Impulse Response file location to read from",
+                    None,
+                    glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
+                ),
+                glib::ParamSpecUInt64::new(
+                    "interpolation-steps",
+                    "Interpolation Steps",
+                    "Interpolation Steps is the amount of slices to cut source to",
+                    0,
+                    u64::MAX - 1,
+                    DEFAULT_INTERPOLATION_STEPS,
+                    glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
+                ),
+                glib::ParamSpecUInt64::new(
+                    "block-length",
+                    "Block Length",
+                    "Block Length is the length of each slice",
+                    0,
+                    u64::MAX - 1,
+                    DEFAULT_BLOCK_LENGTH,
+                    glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_READY,
+                ),
+                gst::ParamSpecArray::new(
+                    "spatial-objects",
+                    "Spatial Objects",
+                    "Spatial object Metadata to apply on input channels",
+                    Some(&glib::ParamSpecBoxed::new(
+                        "spatial-object",
+                        "Spatial Object",
+                        "Spatial Object Metadata",
+                        gst::Structure::static_type(),
+                        glib::ParamFlags::READWRITE,
+                    )),
+                    glib::ParamFlags::READWRITE | gst::PARAM_FLAG_MUTABLE_PLAYING,
+                ),
+            ]
+        });
+
+        PROPERTIES.as_ref()
+    }
+
+    fn set_property(
+        &self,
+        _obj: &Self::Type,
+        _id: usize,
+        value: &glib::Value,
+        pspec: &glib::ParamSpec,
+    ) {
+        match pspec.name() {
+            "hrir-raw" => {
+                let mut settings = self.settings.lock().unwrap();
+                settings.hrir_raw_bytes = value.get().expect("type checked upstream");
+            }
+            "hrir-file" => {
+                let mut settings = self.settings.lock().unwrap();
+                settings.hrir_file_location = value.get().expect("type checked upstream");
+            }
+            "interpolation-steps" => {
+                let mut settings = self.settings.lock().unwrap();
+                settings.interpolation_steps = value.get().expect("type checked upstream");
+            }
+            "block-length" => {
+                let mut settings = self.settings.lock().unwrap();
+                settings.block_length = value.get().expect("type checked upstream");
+            }
+            "spatial-objects" => {
+                let mut settings = self.settings.lock().unwrap();
+
+                let objs = value
+                    .get::<gst::Array>()
+                    .expect("type checked upstream")
+                    .iter()
+                    .map(|v| {
+                        let s = v.get::<gst::Structure>().expect("type checked upstream");
+                        SpatialObject::from(s)
+                    })
+                    .collect::<Vec<_>>();
+
+                let mut state_guard = self.state.lock().unwrap();
+
+                if let Some(state) = state_guard.as_mut() {
+                    if objs.len() != state.ininfo.channels() as usize {
+                        gst::gst_warning!(
+                            CAT,
+                            "Could not update spatial objects, expected {} channels, got {}",
+                            state.ininfo.channels(),
+                            objs.len()
+                        );
+                        return;
+                    }
+                }
+
+                settings.spatial_objects = if objs.is_empty() { None } else { Some(objs) };
+            }
+            _ => unimplemented!(),
+        }
+    }
+
+    fn property(&self, _obj: &Self::Type, _id: usize, pspec: &glib::ParamSpec) -> glib::Value {
+        match pspec.name() {
+            "hrir-raw" => {
+                let settings = self.settings.lock().unwrap();
+                settings.hrir_raw_bytes.to_value()
+            }
+            "hrir-file" => {
+                let settings = self.settings.lock().unwrap();
+                settings.hrir_file_location.to_value()
+            }
+            "interpolation-steps" => {
+                let settings = self.settings.lock().unwrap();
+                settings.interpolation_steps.to_value()
+            }
+            "block-length" => {
+                let settings = self.settings.lock().unwrap();
+                settings.block_length.to_value()
+            }
+            "spatial-objects" => {
+                let settings = self.settings.lock().unwrap();
+                let spatial_objects = settings
+                    .spatial_objects
+                    .as_ref()
+                    .unwrap_or(&Vec::new())
+                    .iter()
+                    .map(|x| gst::Structure::from(*x).to_send_value())
+                    .collect::<Vec<_>>();
+
+                gst::Array::from(spatial_objects).to_value()
+            }
+            _ => unimplemented!(),
+        }
+    }
+}
+
+impl GstObjectImpl for HrtfRender {}
+
+impl ElementImpl for HrtfRender {
+    fn metadata() -> Option<&'static gst::subclass::ElementMetadata> {
+        static ELEMENT_METADATA: Lazy<gst::subclass::ElementMetadata> = Lazy::new(|| {
+            gst::subclass::ElementMetadata::new(
+                "Head-Related Transfer Function (HRTF) renderer",
+                "Filter/Effect/Audio",
+                "Renders spatial sounds to a given position",
+                "Tomasz Andrzejak <andreiltd@gmail.com>",
+            )
+        });
+
+        Some(&*ELEMENT_METADATA)
+    }
+
+    fn pad_templates() -> &'static [gst::PadTemplate] {
+        static PAD_TEMPLATES: Lazy<Vec<gst::PadTemplate>> = Lazy::new(|| {
+            let src_caps = gst::Caps::builder("audio/x-raw")
+                .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+                .field("rate", gst::IntRange::new(1, i32::MAX))
+                .field("channels", 2i32)
+                .field("layout", "interleaved")
+                .build();
+
+            let src_pad_template = gst::PadTemplate::new(
+                "src",
+                gst::PadDirection::Src,
+                gst::PadPresence::Always,
+                &src_caps,
+            )
+            .unwrap();
+
+            let sink_caps = gst::Caps::builder("audio/x-raw")
+                .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+                .field("rate", gst::IntRange::new(1, i32::MAX))
+                .field("channels", gst::IntRange::new(1i32, 64))
+                .field("layout", "interleaved")
+                .build();
+
+            let sink_pad_template = gst::PadTemplate::new(
+                "sink",
+                gst::PadDirection::Sink,
+                gst::PadPresence::Always,
+                &sink_caps,
+            )
+            .unwrap();
+
+            vec![src_pad_template, sink_pad_template]
+        });
+
+        PAD_TEMPLATES.as_ref()
+    }
+}
+
+impl BaseTransformImpl for HrtfRender {
+    const MODE: gst_base::subclass::BaseTransformMode =
+        gst_base::subclass::BaseTransformMode::NeverInPlace;
+    const PASSTHROUGH_ON_SAME_CAPS: bool = false;
+    const TRANSFORM_IP_ON_PASSTHROUGH: bool = false;
+
+    fn transform(
+        &self,
+        element: &Self::Type,
+        inbuf: &gst::Buffer,
+        outbuf: &mut gst::BufferRef,
+    ) -> Result<gst::FlowSuccess, gst::FlowError> {
+        let settings = &self.settings.lock().unwrap();
+
+        let mut state_guard = self.state.lock().unwrap();
+        let state = state_guard.as_mut().ok_or(gst::FlowError::NotNegotiated)?;
+
+        state.adapter.push(inbuf.clone());
+
+        if state.adapter.available() >= state.input_block_size() {
+            return self.process(element, outbuf, state, settings);
+        }
+
+        Ok(gst::FlowSuccess::Ok)
+    }
+
+    fn transform_size(
+        &self,
+        element: &Self::Type,
+        _direction: gst::PadDirection,
+        _caps: &gst::Caps,
+        size: usize,
+        _othercaps: &gst::Caps,
+    ) -> Option<usize> {
+        assert_ne!(_direction, gst::PadDirection::Src);
+
+        let mut state_guard = self.state.lock().unwrap();
+        let state = state_guard.as_mut()?;
+
+        let othersize = {
+            let full_blocks = (size + state.adapter.available()) / (state.input_block_size());
+            full_blocks * state.output_block_size()
+        };
+
+        gst_log!(
+            CAT,
+            obj: element,
+            "Adapter size: {}, input size {}, transformed size {}",
+            state.adapter.available(),
+            size,
+            othersize,
+        );
+
+        Some(othersize)
+    }
+
+    fn transform_caps(
+        &self,
+        element: &Self::Type,
+        direction: gst::PadDirection,
+        caps: &gst::Caps,
+        filter: Option<&gst::Caps>,
+    ) -> Option<gst::Caps> {
+        let mut other_caps = {
+            let mut new_caps = caps.clone();
+
+            for s in new_caps.make_mut().iter_mut() {
+                s.set("format", gst_audio::AUDIO_FORMAT_F32.to_str());
+                s.set("layout", "interleaved");
+
+                if direction == gst::PadDirection::Sink {
+                    s.set("channels", 2);
+                    s.set("channel-mask", 0x3);
+                } else {
+                    let settings = self.settings.lock().unwrap();
+                    if let Some(objs) = &settings.spatial_objects {
+                        s.set("channels", objs.len() as i32);
+                    } else {
+                        s.set("channels", gst::IntRange::new(1, i32::MAX));
+                    }
+
+                    s.remove_field("channel-mask");
+                }
+            }
+            new_caps
+        };
+
+        if let Some(filter) = filter {
+            other_caps = filter.intersect_with_mode(&other_caps, gst::CapsIntersectMode::First);
+        }
+
+        gst_debug!(
+            CAT,
+            obj: element,
+            "Transformed caps from {} to {} in direction {:?}",
+            caps,
+            other_caps,
+            direction
+        );
+
+        Some(other_caps)
+    }
+
+    fn set_caps(
+        &self,
+        element: &Self::Type,
+        incaps: &gst::Caps,
+        outcaps: &gst::Caps,
+    ) -> Result<(), gst::LoggableError> {
+        let ininfo = gst_audio::AudioInfo::from_caps(incaps)
+            .map_err(|_| gst::loggable_error!(CAT, "Failed to parse input caps"))?;
+
+        let outinfo = gst_audio::AudioInfo::from_caps(outcaps)
+            .map_err(|_| gst::loggable_error!(CAT, "Failed to parse output caps"))?;
+
+        let settings = &mut self.settings.lock().unwrap();
+
+        if settings.spatial_objects.is_none() {
+            if let Some(positions) = ininfo.positions() {
+                let objs: Result<Vec<_>, _> = positions
+                    .iter()
+                    .map(|p| SpatialObject::try_from(*p))
+                    .collect();
+
+                if objs.is_err() {
+                    return Err(gst::loggable_error!(CAT, "Unsupported channel position"));
+                }
+
+                settings.spatial_objects = objs.ok();
+            } else {
+                return Err(gst::loggable_error!(CAT, "Cannot infer object positions"));
+            }
+        }
+
+        if settings.spatial_objects.as_ref().unwrap().len() != ininfo.channels() as usize {
+            return Err(gst::loggable_error!(CAT, "Wrong number of spatial objects"));
+        }
+
+        let sphere = settings
+            .sphere(ininfo.rate())
+            .map_err(|e| gst::loggable_error!(CAT, "Failed to load sphere {:?}", e))?;
+
+        let steps = settings.interpolation_steps as usize;
+        let blklen = settings.block_length as usize;
+
+        let block_samples = blklen
+            .checked_mul(steps)
+            .ok_or_else(|| gst::loggable_error!(CAT, "Not enough memory for frame allocation"))?;
+
+        let channel_processors = (0..ininfo.channels())
+            .map(|_| ChannelProcessor {
+                prev_left_samples: vec![0.0; block_samples],
+                prev_right_samples: vec![0.0; block_samples],
+                prev_sample_vector: None,
+                prev_distance_gain: None,
+                indata_scratch: vec![0.0; block_samples].into_boxed_slice(),
+                outdata_scratch: vec![(0.0, 0.0); block_samples].into_boxed_slice(),
+                processor: HrtfProcessor::new(sphere.to_owned(), steps, blklen),
+            })
+            .collect();
+
+        *self.state.lock().unwrap() = Some(State {
+            ininfo,
+            outinfo,
+            block_samples,
+            channel_processors,
+            adapter: gst_base::UniqueAdapter::new(),
+        });
+
+        gst_debug!(CAT, obj: element, "Configured for caps {}", incaps);
+
+        Ok(())
+    }
+
+    fn sink_event(&self, element: &Self::Type, event: gst::Event) -> bool {
+        use gst::EventView;
+
+        gst_debug!(CAT, "Handling event {:?}", event);
+
+        match event.view() {
+            EventView::FlushStop(_) => {
+                let mut state_guard = self.state.lock().unwrap();
+
+                if let Some(state) = state_guard.as_mut() {
+                    let avail = state.adapter.available();
+                    state.adapter.flush(avail);
+                    state.reset_processors();
+                }
+            }
+            EventView::Eos(_) => {
+                if self.drain(element).is_err() {
+                    gst::gst_warning!(CAT, "Failed to drain internal buffer");
+                    gst::element_warning!(
+                        element,
+                        gst::CoreError::Event,
+                        ["Failed to drain internal buffer"]
+                    );
+                }
+            }
+            _ => {}
+        }
+
+        self.parent_sink_event(element, event)
+    }
+
+    fn start(&self, _element: &Self::Type) -> Result<(), gst::ErrorMessage> {
+        // get global thread pool
+        let mut thread_pool_g = THREAD_POOL.lock().unwrap();
+        let mut thread_pool = self.thread_pool.lock().unwrap();
+
+        if let Some(tp) = thread_pool_g.upgrade() {
+            *thread_pool = Some(tp);
+        } else {
+            let tp = ThreadPoolBuilder::new().build().map_err(|_| {
+                gst::error_msg!(
+                    gst::CoreError::Failed,
+                    ["Could not create rayon thread pool"]
+                )
+            })?;
+
+            let tp = Arc::new(tp);
+
+            *thread_pool = Some(tp);
+            *thread_pool_g = Arc::downgrade(thread_pool.as_ref().unwrap());
+        }
+
+        Ok(())
+    }
+
+    fn stop(&self, _element: &Self::Type) -> Result<(), gst::ErrorMessage> {
+        // Drop state
+        let _ = self.state.lock().unwrap().take();
+        // Drop thread pool
+        let _ = self.thread_pool.lock().unwrap().take();
+
+        Ok(())
+    }
+}
diff --git a/audio/audiofx/src/hrtfrender/mod.rs b/audio/audiofx/src/hrtfrender/mod.rs
new file mode 100644
index 00000000..6f2349f9
--- /dev/null
+++ b/audio/audiofx/src/hrtfrender/mod.rs
@@ -0,0 +1,26 @@
+// Copyright (C) 2021 Tomasz Andrzejak <andreiltd@gmail.com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+use gst::glib;
+use gst::prelude::*;
+
+mod imp;
+
+glib::wrapper! {
+    pub struct HrtfRender(ObjectSubclass<imp::HrtfRender>) @extends gst_base::BaseTransform, gst::Element, gst::Object;
+}
+
+unsafe impl Send for HrtfRender {}
+unsafe impl Sync for HrtfRender {}
+
+pub fn register(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
+    gst::Element::register(
+        Some(plugin),
+        "hrtfrender",
+        gst::Rank::None,
+        HrtfRender::static_type(),
+    )
+}
diff --git a/audio/audiofx/src/lib.rs b/audio/audiofx/src/lib.rs
index 1d2cc11c..dcf2b177 100644
--- a/audio/audiofx/src/lib.rs
+++ b/audio/audiofx/src/lib.rs
@@ -12,12 +12,14 @@ mod audioecho;
 mod audioloudnorm;
 mod audiornnoise;
 mod ebur128level;
+mod hrtfrender;
 
 fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
     audioecho::register(plugin)?;
     audioloudnorm::register(plugin)?;
     audiornnoise::register(plugin)?;
     ebur128level::register(plugin)?;
+    hrtfrender::register(plugin)?;
     Ok(())
 }
 
diff --git a/audio/audiofx/tests/hrtfrender.rs b/audio/audiofx/tests/hrtfrender.rs
new file mode 100644
index 00000000..3a3d6571
--- /dev/null
+++ b/audio/audiofx/tests/hrtfrender.rs
@@ -0,0 +1,287 @@
+// Copyright (C) 2021 Tomasz Andrzejak <andreiltd@gmail.com>
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+use gst::glib;
+use gst::prelude::*;
+
+use once_cell::sync::Lazy;
+use std::io::{Cursor, Write};
+
+static CONFIG: Lazy<glib::Bytes> = Lazy::new(|| {
+    let mut buff = Cursor::new(vec![0u8; 1024]);
+
+    write_config(&mut buff).unwrap();
+    glib::Bytes::from_owned(buff.get_ref().to_owned())
+});
+
+// Generates a fake config
+fn write_config(writer: &mut impl Write) -> std::io::Result<()> {
+    const SAMPLE_RATE: u32 = 44_100;
+    const LENGTH: u32 = 2;
+    const VERTEX_COUNT: u32 = 2;
+    const INDEX_COUNT: u32 = 2 * 3;
+
+    writer.write_all(b"HRIR")?;
+    writer.write_all(&SAMPLE_RATE.to_le_bytes())?;
+    writer.write_all(&LENGTH.to_le_bytes())?;
+    writer.write_all(&VERTEX_COUNT.to_le_bytes())?;
+    writer.write_all(&INDEX_COUNT.to_le_bytes())?;
+
+    // Write Indices
+    for _ in 0..INDEX_COUNT {
+        writer.write_all(&0u32.to_le_bytes())?;
+    }
+
+    // Write Vertices
+    for _ in 0..VERTEX_COUNT {
+        for _ in 0..3 {
+            writer.write_all(&0u32.to_le_bytes())?;
+        }
+
+        for _ in 0..LENGTH * 2 {
+            writer.write_all(&0f32.to_le_bytes())?;
+        }
+    }
+
+    Ok(())
+}
+
+fn init() {
+    use std::sync::Once;
+    static INIT: Once = Once::new();
+
+    INIT.call_once(|| {
+        gst::init().unwrap();
+        gstrsaudiofx::plugin_register_static().expect("Failed to register rsaudiofx plugin");
+    });
+}
+
+fn build_harness(src_caps: gst::Caps, sink_caps: gst::Caps) -> (gst_check::Harness, gst::Element) {
+    let hrtf = gst::ElementFactory::make("hrtfrender", None).unwrap();
+    hrtf.set_property("hrir-raw", &*CONFIG);
+
+    let mut h = gst_check::Harness::with_element(&hrtf, Some("sink"), Some("src"));
+    h.set_caps(src_caps, sink_caps);
+
+    (h, hrtf)
+}
+
+#[test]
+fn test_hrtfrender_samples_in_samples_out() {
+    init();
+
+    let src_caps = gst::Caps::builder("audio/x-raw")
+        .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+        .field("rate", 44_100i32)
+        .field("channels", 1i32)
+        .field("channel-mask", gst::Bitmask::new(0x1))
+        .field("layout", "interleaved")
+        .build();
+
+    let sink_caps = gst::Caps::builder("audio/x-raw")
+        .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+        .field("rate", 44_100i32)
+        .field("channels", 2i32)
+        .field("layout", "interleaved")
+        .build();
+
+    let (mut h, _) = build_harness(src_caps, sink_caps);
+    h.play();
+
+    let inbpf = 4;
+    let outbpf = 8;
+    let full_block = 512 * 8;
+
+    let mut buffer = gst::Buffer::with_size(full_block * inbpf + 20 * inbpf).unwrap();
+    let buffer_mut = buffer.get_mut().unwrap();
+
+    let full_block_time = (full_block as u64)
+        .mul_div_round(*gst::ClockTime::SECOND, 44_100)
+        .map(gst::ClockTime::from_nseconds);
+
+    buffer_mut.set_pts(gst::ClockTime::ZERO);
+    buffer_mut.set_duration(full_block_time);
+    buffer_mut.set_offset(0);
+
+    let ret = h.push(buffer);
+    assert!(ret.is_ok());
+
+    let buffer = h.pull().unwrap();
+    assert_eq!(buffer.size(), full_block * outbpf);
+
+    h.push_event(gst::event::Eos::new());
+    let buffer = h.pull().unwrap();
+
+    assert_eq!(buffer.size(), 20 * outbpf);
+    assert_eq!(buffer.offset(), full_block as u64);
+    assert_eq!(buffer.pts(), full_block_time);
+
+    let residue_time = 20
+        .mul_div_round(*gst::ClockTime::SECOND, 44_100)
+        .map(gst::ClockTime::from_nseconds);
+
+    assert_eq!(buffer.duration(), residue_time);
+}
+
+#[test]
+fn test_hrtfrender_implicit_spatial_objects() {
+    init();
+
+    let src_caps = gst::Caps::builder("audio/x-raw")
+        .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+        .field("rate", 44_100i32)
+        .field("channels", 8i32)
+        .field("channel-mask", gst::Bitmask::new(0xc3f))
+        .field("layout", "interleaved")
+        .build();
+
+    let sink_caps = gst::Caps::builder("audio/x-raw")
+        .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+        .field("rate", 44_100i32)
+        .field("channels", 2i32)
+        .field("layout", "interleaved")
+        .build();
+
+    let (mut h, hrtf) = build_harness(src_caps, sink_caps);
+    let objs = hrtf.property::<gst::Array>("spatial-objects");
+
+    h.play();
+
+    assert_eq!(objs.len(), 8);
+}
+
+#[test]
+fn test_hrtfrender_explicit_spatial_objects() {
+    init();
+
+    let src_caps = gst::Caps::builder("audio/x-raw")
+        .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+        .field("rate", 44_100i32)
+        .field("channels", 8i32)
+        .field("layout", "interleaved")
+        .build();
+
+    let sink_caps = gst::Caps::builder("audio/x-raw")
+        .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+        .field("rate", 44_100i32)
+        .field("channels", 2i32)
+        .field("layout", "interleaved")
+        .build();
+
+    let (mut h, hrtf) = build_harness(src_caps, sink_caps);
+
+    let objs = (0..8)
+        .map(|x| {
+            gst::Structure::builder("application/spatial-object")
+                .field("x", -1f32 + x as f32 / 8f32)
+                .field("y", 0f32)
+                .field("z", 1f32)
+                .field("distance-gain", 0.1f32)
+                .build()
+        })
+        .collect::<Vec<_>>();
+
+    hrtf.set_property("spatial-objects", gst::Array::new(objs));
+
+    h.play();
+
+    let objs = hrtf.property::<gst::Array>("spatial-objects");
+    assert_eq!(objs.len(), 8);
+}
+
+#[test]
+// Caps negotation should fail if we have mismatch between input channels and
+// of objects that we set via property. In this test case input has 6 channels
+// but the number of spatial objects set is 2.
+fn test_hrtfrender_caps_negotiation_fail() {
+    init();
+
+    let src_caps = gst::Caps::builder("audio/x-raw")
+        .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+        .field("rate", 44_100i32)
+        .field("channels", 6i32)
+        .field("layout", "interleaved")
+        .build();
+
+    let sink_caps = gst::Caps::builder("audio/x-raw")
+        .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+        .field("rate", 44_100i32)
+        .field("channels", 2i32)
+        .field("layout", "interleaved")
+        .build();
+
+    let (mut h, hrtf) = build_harness(src_caps, sink_caps);
+
+    let objs = (0..2)
+        .map(|_| {
+            gst::Structure::builder("application/spatial-object")
+                .field("x", 0f32)
+                .field("y", 0f32)
+                .field("z", 1f32)
+                .field("distance-gain", 0.1f32)
+                .build()
+        })
+        .collect::<Vec<_>>();
+
+    hrtf.set_property("spatial-objects", gst::Array::new(objs));
+
+    h.play();
+
+    let buffer = gst::Buffer::with_size(2048).unwrap();
+    assert_eq!(h.push(buffer), Err(gst::FlowError::NotNegotiated));
+
+    h.push_event(gst::event::Eos::new());
+
+    // The harness sinkpad end up not having defined caps so, the current_caps
+    // should be None
+    let current_caps = h.sinkpad().expect("harness has no sinkpad").current_caps();
+
+    assert!(current_caps.is_none());
+}
+
+#[test]
+fn test_hrtfrender_multiple_instances_sharing_thread_pool() {
+    init();
+
+    let src_caps = gst::Caps::builder("audio/x-raw")
+        .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+        .field("rate", 44_100i32)
+        .field("channels", 1i32)
+        .field("layout", "interleaved")
+        .build();
+
+    let sink_caps = gst::Caps::builder("audio/x-raw")
+        .field("format", gst_audio::AUDIO_FORMAT_F32.to_str())
+        .field("rate", 44_100i32)
+        .field("channels", 2i32)
+        .field("layout", "interleaved")
+        .build();
+
+    let (_, hrtf) = build_harness(src_caps.clone(), sink_caps.clone());
+
+    let block_length: u64 = hrtf.property::<u64>("block-length");
+    let steps: u64 = hrtf.property::<u64>("interpolation-steps");
+    let bps: u64 = 4;
+
+    drop(hrtf);
+    let blksz = (block_length * steps * bps) as usize;
+
+    let mut harnesses = (0..4)
+        .map(|_| build_harness(src_caps.clone(), sink_caps.clone()).0)
+        .collect::<Vec<_>>();
+
+    for h in harnesses.iter_mut() {
+        h.play();
+
+        let buffer = gst::Buffer::with_size(blksz).unwrap();
+        assert!(h.push(buffer).is_ok());
+
+        let buffer = h.pull().unwrap();
+        assert_eq!(buffer.size(), 2 * blksz);
+
+        h.push_event(gst::event::Eos::new());
+    }
+}