Update GStreamer to 0.18 and clean up (#964)

* Update GStreamer backend to 0.18
* Don't manually go through all intermediate states when shutting down the GStreamer backend; that happens automatically
* Don't initialize GStreamer twice
* Use less stringly-typed API for configuring the appsrc
* Create our own main context instead of stealing the default one; if the application somewhere else uses the default main context this would otherwise fail in interesting ways
* Create GStreamer pipeline more explicitly instead of going via strings for everything
* Add an audioresample element before the sink in case the sink doesn't support the sample rate
* Remove unnecessary `as_bytes()` call
* Use a GStreamer bus sync handler instead of spawning a new thread with a mainloop; it's only used for printing errors or when the end of the stream is reached, which can also be done as well when synchronously handling messages.
* Change `expect()` calls to proper error returns wherever possible in GStreamer backend
* Store asynchronously reported error in GStreamer backend and return them on next write
* Update MSRV to 1.56

.github/workflows/test.yml

@@ -55,7 +55,7 @@ jobs:
       matrix:
         os: [ubuntu-latest]
         toolchain:
-          - 1.53 # MSRV (Minimum supported rust version)
+          - 1.56 # MSRV (Minimum supported rust version)
           - stable
         experimental: [false]
         # Ignore failures in beta
@@ -113,7 +113,7 @@ jobs:
       matrix:
         os: [windows-latest]
         toolchain:
-          - 1.53 # MSRV (Minimum supported rust version)
+          - 1.56 # MSRV (Minimum supported rust version)
           - stable
     steps:
       - name: Checkout code
@@ -160,7 +160,7 @@ jobs:
         os: [ubuntu-latest]
         target: [armv7-unknown-linux-gnueabihf]
         toolchain:
-          - 1.53 # MSRV (Minimum supported rust version)
+          - 1.56 # MSRV (Minimum supported rust version)
           - stable
     steps:
       - name: Checkout code

COMPILING.md

@@ -7,7 +7,7 @@ In order to compile librespot, you will first need to set up a suitable Rust bui
 ### Install Rust
 The easiest, and recommended way to get Rust is to use [rustup](https://rustup.rs). Once that’s installed, Rust's standard tools should be set up and ready to use.
 
-*Note: The current minimum required Rust version at the time of writing is 1.53.*
+*Note: The current minimum required Rust version at the time of writing is 1.56.*
 
 #### Additional Rust tools - `rustfmt`
 To ensure a consistent codebase, we utilise [`rustfmt`](https://github.com/rust-lang/rustfmt) and [`clippy`](https://github.com/rust-lang/rust-clippy), which are installed by default with `rustup` these days, else they can be installed manually with:

Cargo.lock

@@ -67,6 +67,12 @@ version = "1.0.53"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "94a45b455c14666b85fc40a019e8ab9eb75e3a124e05494f5397122bc9eb06e0"
 
+[[package]]
+name = "array-init"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6945cc5422176fc5e602e590c2878d2c2acd9a4fe20a4baa7c28022521698ec6"
+
 [[package]]
 name = "arrayvec"
 version = "0.7.2"
@@ -221,9 +227,9 @@ dependencies = [
 
 [[package]]
 name = "cfg-expr"
-version = "0.8.1"
+version = "0.9.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b412e83326147c2bb881f8b40edfbf9905b9b8abaebd0e47ca190ba62fda8f0e"
+checksum = "3431df59f28accaf4cb4eed4a9acc66bea3f3c3753aa6cdc2f024174ef232af7"
 dependencies = [
  "smallvec",
 ]
@@ -502,12 +508,6 @@ dependencies = [
  "pkg-config",
 ]
 
-[[package]]
-name = "either"
-version = "1.6.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e78d4f1cc4ae33bbfc157ed5d5a5ef3bc29227303d595861deb238fcec4e9457"
-
 [[package]]
 name = "encoding_rs"
 version = "0.8.30"
@@ -733,9 +733,9 @@ dependencies = [
 
 [[package]]
 name = "glib"
-version = "0.14.8"
+version = "0.15.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7c515f1e62bf151ef6635f528d05b02c11506de986e43b34a5c920ef0b3796a4"
+checksum = "e385b6c17a1add7d0fbc64d38e2e742346d3e8b22e5fa3734e5cdca2be24028d"
 dependencies = [
  "bitflags",
  "futures-channel",
@@ -748,13 +748,14 @@ dependencies = [
  "libc",
  "once_cell",
  "smallvec",
+ "thiserror",
 ]
 
 [[package]]
 name = "glib-macros"
-version = "0.14.1"
+version = "0.15.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2aad66361f66796bfc73f530c51ef123970eb895ffba991a234fcf7bea89e518"
+checksum = "e58b262ff65ef771003873cea8c10e0fe854f1c508d48d62a4111a1ff163f7d1"
 dependencies = [
  "anyhow",
  "heck",
@@ -767,9 +768,9 @@ dependencies = [
 
 [[package]]
 name = "glib-sys"
-version = "0.14.0"
+version = "0.15.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c1d60554a212445e2a858e42a0e48cece1bd57b311a19a9468f70376cf554ae"
+checksum = "0c4f08dd67f74b223fedbbb30e73145b9acd444e67cc4d77d0598659b7eebe7e"
 dependencies = [
  "libc",
  "system-deps",
@@ -783,9 +784,9 @@ checksum = "9b919933a397b79c37e33b77bb2aa3dc8eb6e165ad809e58ff75bc7db2e34574"
 
 [[package]]
 name = "gobject-sys"
-version = "0.14.0"
+version = "0.15.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aa92cae29759dae34ab5921d73fff5ad54b3d794ab842c117e36cafc7994c3f5"
+checksum = "6edb1f0b3e4c08e2a0a490d1082ba9e902cdff8ff07091e85c6caec60d17e2ab"
 dependencies = [
  "glib-sys",
  "libc",
@@ -794,9 +795,9 @@ dependencies = [
 
 [[package]]
 name = "gstreamer"
-version = "0.17.4"
+version = "0.18.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c6a255f142048ba2c4a4dce39106db1965abe355d23f4b5335edea43a553faa4"
+checksum = "a54229ced7e44752bff52360549cd412802a4b1a19852b87346625ca9f6d4330"
 dependencies = [
  "bitflags",
  "cfg-if",
@@ -810,6 +811,7 @@ dependencies = [
  "num-integer",
  "num-rational",
  "once_cell",
+ "option-operations",
  "paste",
  "pretty-hex",
  "thiserror",
@@ -817,9 +819,9 @@ dependencies = [
 
 [[package]]
 name = "gstreamer-app"
-version = "0.17.2"
+version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f73b8d33b1bbe9f22d0cf56661a1d2a2c9a0e099ea10e5f1f347be5038f5c043"
+checksum = "653b14862e385f6a568a5c54aee830c525277418d765e93cdac1c1b97e25f300"
 dependencies = [
  "bitflags",
  "futures-core",
@@ -834,9 +836,9 @@ dependencies = [
 
 [[package]]
 name = "gstreamer-app-sys"
-version = "0.17.0"
+version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "41865cfb8a5ddfa1161734a0d068dcd4689da852be0910b40484206408cfeafa"
+checksum = "c3b401f21d731b3e5de802487f25507fabd34de2dd007d582f440fb1c66a4fbb"
 dependencies = [
  "glib-sys",
  "gstreamer-base-sys",
@@ -846,10 +848,41 @@ dependencies = [
 ]
 
 [[package]]
-name = "gstreamer-base"
+name = "gstreamer-audio"
-version = "0.17.2"
+version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2c0c1d8c62eb5d08fb80173609f2eea71d385393363146e4e78107facbd67715"
+checksum = "75cc407516c2f36576060767491f1134728af6d335a59937f09a61aab7abb72c"
+dependencies = [
+ "array-init",
+ "bitflags",
+ "cfg-if",
+ "glib",
+ "gstreamer",
+ "gstreamer-audio-sys",
+ "gstreamer-base",
+ "libc",
+ "once_cell",
+]
+
+[[package]]
+name = "gstreamer-audio-sys"
+version = "0.18.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a34258fb53c558c0f41dad194037cbeaabf49d347570df11b8bd1c4897cf7d7c"
+dependencies = [
+ "glib-sys",
+ "gobject-sys",
+ "gstreamer-base-sys",
+ "gstreamer-sys",
+ "libc",
+ "system-deps",
+]
+
+[[package]]
+name = "gstreamer-base"
+version = "0.18.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "224f35f36582407caf58ded74854526beeecc23d0cf64b8d1c3e00584ed6863f"
 dependencies = [
  "bitflags",
  "cfg-if",
@@ -861,9 +894,9 @@ dependencies = [
 
 [[package]]
 name = "gstreamer-base-sys"
-version = "0.17.0"
+version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "28169a7b58edb93ad8ac766f0fa12dcd36a2af4257a97ee10194c7103baf3e27"
+checksum = "a083493c3c340e71fa7c66eebda016e9fafc03eb1b4804cf9b2bad61994b078e"
 dependencies = [
  "glib-sys",
  "gobject-sys",
@@ -874,9 +907,9 @@ dependencies = [
 
 [[package]]
 name = "gstreamer-sys"
-version = "0.17.3"
+version = "0.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a81704feeb3e8599913bdd1e738455c2991a01ff4a1780cb62200993e454cc3e"
+checksum = "e3517a65d3c2e6f8905b456eba5d53bda158d664863aef960b44f651cb7d33e2"
 dependencies = [
  "glib-sys",
  "gobject-sys",
@@ -936,12 +969,9 @@ dependencies = [
 
 [[package]]
 name = "heck"
-version = "0.3.3"
+version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d621efb26863f0e9924c6ac577e8275e5e6b77455db64ffa6c65c904e9e132c"
+checksum = "2540771e65fc8cb83cd6e8a237f70c319bd5c29f78ed1084ba5d50eeac86f7f9"
-dependencies = [
- "unicode-segmentation",
-]
 
 [[package]]
 name = "hermit-abi"
@@ -1150,15 +1180,6 @@ dependencies = [
  "cfg-if",
 ]
 
-[[package]]
-name = "itertools"
-version = "0.10.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9a9d19fa1e79b6215ff29b9d6880b706147f16e9b1dbb1e4e5947b5b02bc5e3"
-dependencies = [
- "either",
-]
-
 [[package]]
 name = "itoa"
 version = "0.4.8"
@@ -1532,6 +1553,7 @@ dependencies = [
  "glib",
  "gstreamer",
  "gstreamer-app",
+ "gstreamer-audio",
  "jack 0.8.4",
  "libpulse-binding",
  "libpulse-simple-binding",
@@ -2003,6 +2025,15 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf"
 
+[[package]]
+name = "option-operations"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "95d6113415f41b268f1195907427519769e40ee6f28cbb053795098a2c16f447"
+dependencies = [
+ "paste",
+]
+
 [[package]]
 name = "parking_lot"
 version = "0.11.2"
@@ -2534,7 +2565,7 @@ checksum = "94cb479353c0603785c834e2307440d83d196bf255f204f7f6741358de8d6a2f"
 dependencies = [
  "cfg-if",
  "libc",
- "version-compare 0.1.0",
+ "version-compare",
 ]
 
 [[package]]
@@ -2700,24 +2731,6 @@ version = "0.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"
 
-[[package]]
-name = "strum"
-version = "0.21.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aaf86bbcfd1fa9670b7a129f64fc0c9fcbbfe4f1bc4210e9e98fe71ffc12cde2"
-
-[[package]]
-name = "strum_macros"
-version = "0.21.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d06aaeeee809dbc59eb4556183dd927df67db1540de5be8d3ec0b6636358a5ec"
-dependencies = [
- "heck",
- "proc-macro2",
- "quote",
- "syn",
-]
-
 [[package]]
 name = "subtle"
 version = "2.4.1"
@@ -2834,20 +2847,15 @@ dependencies = [
 
 [[package]]
 name = "system-deps"
-version = "3.2.0"
+version = "6.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "480c269f870722b3b08d2f13053ce0c2ab722839f472863c3e2d61ff3a1c2fa6"
+checksum = "ad3a97fdef3daf935d929b3e97e5a6a680cd4622e40c2941ca0875d6566416f8"
 dependencies = [
- "anyhow",
  "cfg-expr",
  "heck",
- "itertools",
  "pkg-config",
- "strum",
- "strum_macros",
- "thiserror",
  "toml",
- "version-compare 0.0.11",
+ "version-compare",
 ]
 
 [[package]]
@@ -3115,12 +3123,6 @@ dependencies = [
  "tinyvec",
 ]
 
-[[package]]
-name = "unicode-segmentation"
-version = "1.8.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8895849a949e7845e06bd6dc1aa51731a103c42707010a5b591c0038fb73385b"
-
 [[package]]
 name = "unicode-width"
 version = "0.1.9"
@@ -3188,12 +3190,6 @@ dependencies = [
  "thiserror",
 ]
 
-[[package]]
-name = "version-compare"
-version = "0.0.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1c18c859eead79d8b95d09e4678566e8d70105c4e7b251f707a03df32442661b"
-
 [[package]]
 name = "version-compare"
 version = "0.1.0"

playback/Cargo.toml

@@ -34,9 +34,10 @@ libpulse-binding        = { version = "2", optional = true, default-features = f
 libpulse-simple-binding = { version = "2", optional = true, default-features = false }
 jack            = { version = "0.8", optional = true }
 sdl2            = { version = "0.35", optional = true }
-gstreamer       = { version = "0.17", optional = true }
+gst             = { package = "gstreamer", version = "0.18", optional = true }
-gstreamer-app   = { version = "0.17", optional = true }
+gst-app         = { package = "gstreamer-app", version = "0.18", optional = true }
-glib            = { version = "0.14", optional = true }
+gst-audio       = { package = "gstreamer-audio", version = "0.18", optional = true }
+glib            = { version = "0.15", optional = true }
 
 # Rodio dependencies
 rodio           = { version = "0.15", optional = true, default-features = false }
@@ -60,6 +61,6 @@ jackaudio-backend = ["jack"]
 rodio-backend = ["rodio", "cpal"]
 rodiojack-backend = ["rodio", "cpal/jack"]
 sdl-backend = ["sdl2"]
-gstreamer-backend = ["gstreamer", "gstreamer-app", "glib"]
+gstreamer-backend = ["gst", "gst-app", "gst-audio", "glib"]
 
 passthrough-decoder = ["ogg"]

playback/src/audio_backend/gstreamer.rs

@@ -1,14 +1,11 @@
-use std::{ops::Drop, thread};
-
-use gstreamer as gst;
-use gstreamer_app as gst_app;
-
 use gst::{
     event::{FlushStart, FlushStop},
     prelude::*,
     State,
 };
-use zerocopy::AsBytes;
+
+use parking_lot::Mutex;
+use std::sync::Arc;
 
 use super::{Open, Sink, SinkAsBytes, SinkError, SinkResult};
 
@@ -16,12 +13,12 @@ use crate::{
     config::AudioFormat, convert::Converter, decoder::AudioPacket, NUM_CHANNELS, SAMPLE_RATE,
 };
 
-#[allow(dead_code)]
 pub struct GstreamerSink {
     appsrc: gst_app::AppSrc,
     bufferpool: gst::BufferPool,
     pipeline: gst::Pipeline,
     format: AudioFormat,
+    async_error: Arc<Mutex<Option<String>>>,
 }
 
 impl Open for GstreamerSink {
@@ -29,63 +26,74 @@ impl Open for GstreamerSink {
         info!("Using GStreamer sink with format: {:?}", format);
         gst::init().expect("failed to init GStreamer!");
 
-        // GStreamer calls S24 and S24_3 different from the rest of the world
         let gst_format = match format {
-            AudioFormat::S24 => "S24_32".to_string(),
+            AudioFormat::F64 => gst_audio::AUDIO_FORMAT_F64,
-            AudioFormat::S24_3 => "S24".to_string(),
+            AudioFormat::F32 => gst_audio::AUDIO_FORMAT_F32,
-            _ => format!("{:?}", format),
+            AudioFormat::S32 => gst_audio::AUDIO_FORMAT_S32,
+            AudioFormat::S24 => gst_audio::AUDIO_FORMAT_S2432,
+            AudioFormat::S24_3 => gst_audio::AUDIO_FORMAT_S24,
+            AudioFormat::S16 => gst_audio::AUDIO_FORMAT_S16,
         };
+
+        let gst_info = gst_audio::AudioInfo::builder(gst_format, SAMPLE_RATE, NUM_CHANNELS as u32)
+            .build()
+            .expect("Failed to create GStreamer audio format");
+        let gst_caps = gst_info.to_caps().expect("Failed to create GStreamer caps");
+
         let sample_size = format.size();
         let gst_bytes = NUM_CHANNELS as usize * 1024 * sample_size;
 
-        #[cfg(target_endian = "little")]
+        let pipeline = gst::Pipeline::new(None);
-        const ENDIANNESS: &str = "LE";
+        let appsrc = gst::ElementFactory::make("appsrc", None)
-        #[cfg(target_endian = "big")]
+            .expect("Failed to create GStreamer appsrc element")
-        const ENDIANNESS: &str = "BE";
+            .downcast::<gst_app::AppSrc>()
-
-        let pipeline_str_preamble = format!(
-            "appsrc caps=\"audio/x-raw,format={}{},layout=interleaved,channels={},rate={}\" block=true max-bytes={} name=appsrc0 ",
-            gst_format, ENDIANNESS, NUM_CHANNELS, SAMPLE_RATE, gst_bytes
-        );
-        // no need to dither twice; use librespot dithering instead
-        let pipeline_str_rest = r#" ! audioconvert dithering=none ! autoaudiosink"#;
-        let pipeline_str: String = match device {
-            Some(x) => format!("{}{}", pipeline_str_preamble, x),
-            None => format!("{}{}", pipeline_str_preamble, pipeline_str_rest),
-        };
-        info!("Pipeline: {}", pipeline_str);
-
-        gst::init().unwrap();
-        let pipelinee = gst::parse_launch(&*pipeline_str).expect("Couldn't launch pipeline; likely a GStreamer issue or an error in the pipeline string you specified in the 'device' argument to librespot.");
-        let pipeline = pipelinee
-            .dynamic_cast::<gst::Pipeline>()
-            .expect("couldn't cast pipeline element at runtime!");
-        let bus = pipeline.bus().expect("couldn't get bus from pipeline");
-        let mainloop = glib::MainLoop::new(None, false);
-        let appsrce: gst::Element = pipeline
-            .by_name("appsrc0")
-            .expect("couldn't get appsrc from pipeline");
-        let appsrc: gst_app::AppSrc = appsrce
-            .dynamic_cast::<gst_app::AppSrc>()
             .expect("couldn't cast AppSrc element at runtime!");
-        let appsrc_caps = appsrc.caps().expect("couldn't get appsrc caps");
+        appsrc.set_caps(Some(&gst_caps));
+        appsrc.set_max_bytes(gst_bytes as u64);
+        appsrc.set_block(true);
+
+        let sink = match device {
+            None => {
+                // no need to dither twice; use librespot dithering instead
+                gst::parse_bin_from_description(
+                    "audioconvert dithering=none ! audioresample ! autoaudiosink",
+                    true,
+                )
+                .expect("Failed to create default GStreamer sink")
+            }
+            Some(ref x) => gst::parse_bin_from_description(x, true)
+                .expect("Failed to create custom GStreamer sink"),
+        };
+        pipeline
+            .add(&appsrc)
+            .expect("Failed to add GStreamer appsrc to pipeline");
+        pipeline
+            .add(&sink)
+            .expect("Failed to add GStreamer sink to pipeline");
+        appsrc
+            .link(&sink)
+            .expect("Failed to link GStreamer source to sink");
+
+        let bus = pipeline.bus().expect("couldn't get bus from pipeline");
 
         let bufferpool = gst::BufferPool::new();
 
         let mut conf = bufferpool.config();
-        conf.set_params(Some(&appsrc_caps), gst_bytes as u32, 0, 0);
+        conf.set_params(Some(&gst_caps), gst_bytes as u32, 0, 0);
         bufferpool
             .set_config(conf)
             .expect("couldn't configure the buffer pool");
 
-        thread::spawn(move || {
+        let async_error = Arc::new(Mutex::new(None));
-            let thread_mainloop = mainloop;
+        let async_error_clone = async_error.clone();
-            let watch_mainloop = thread_mainloop.clone();
+
-            bus.add_watch(move |_, msg| {
+        bus.set_sync_handler(move |_bus, msg| {
             match msg.view() {
                 gst::MessageView::Eos(_) => {
                     println!("gst signaled end of stream");
-                        watch_mainloop.quit();
+
+                    let mut async_error_storage = async_error_clone.lock();
+                    *async_error_storage = Some(String::from("gst signaled end of stream"));
                 }
                 gst::MessageView::Error(err) => {
                     println!(
@@ -94,15 +102,19 @@ impl Open for GstreamerSink {
                         err.error(),
                         err.debug()
                     );
-                        watch_mainloop.quit();
+
+                    let mut async_error_storage = async_error_clone.lock();
+                    *async_error_storage = Some(format!(
+                        "Error from {:?}: {} ({:?})",
+                        err.src().map(|s| s.path_string()),
+                        err.error(),
+                        err.debug()
+                    ));
                 }
                 _ => (),
-                };
+            }
 
-                glib::Continue(true)
+            gst::BusSyncReply::Drop
-            })
-            .expect("failed to add bus watch");
-            thread_mainloop.run();
         });
 
         pipeline
@@ -114,16 +126,18 @@ impl Open for GstreamerSink {
             bufferpool,
             pipeline,
             format,
+            async_error,
         }
     }
 }
 
 impl Sink for GstreamerSink {
     fn start(&mut self) -> SinkResult<()> {
+        *self.async_error.lock() = None;
         self.appsrc.send_event(FlushStop::new(true));
         self.bufferpool
             .set_active(true)
-            .expect("couldn't activate buffer pool");
+            .map_err(|e| SinkError::StateChange(e.to_string()))?;
         self.pipeline
             .set_state(State::Playing)
             .map_err(|e| SinkError::StateChange(e.to_string()))?;
@@ -131,13 +145,14 @@ impl Sink for GstreamerSink {
     }
 
     fn stop(&mut self) -> SinkResult<()> {
+        *self.async_error.lock() = None;
         self.appsrc.send_event(FlushStart::new());
         self.pipeline
             .set_state(State::Paused)
             .map_err(|e| SinkError::StateChange(e.to_string()))?;
         self.bufferpool
             .set_active(false)
-            .expect("couldn't deactivate buffer pool");
+            .map_err(|e| SinkError::StateChange(e.to_string()))?;
         Ok(())
     }
 
@@ -146,15 +161,16 @@ impl Sink for GstreamerSink {
 
 impl Drop for GstreamerSink {
     fn drop(&mut self) {
-        // Follow the state transitions documented at:
-        // https://gstreamer.freedesktop.org/documentation/additional/design/states.html?gi-language=c
-        let _ = self.pipeline.set_state(State::Ready);
         let _ = self.pipeline.set_state(State::Null);
     }
 }
 
 impl SinkAsBytes for GstreamerSink {
     fn write_bytes(&mut self, data: &[u8]) -> SinkResult<()> {
+        if let Some(async_error) = &*self.async_error.lock() {
+            return Err(SinkError::OnWrite(async_error.to_string()));
+        }
+
         let mut buffer = self
             .bufferpool
             .acquire_buffer(None)
@@ -163,8 +179,8 @@ impl SinkAsBytes for GstreamerSink {
         let mutbuf = buffer.make_mut();
         mutbuf.set_size(data.len());
         mutbuf
-            .copy_from_slice(0, data.as_bytes())
+            .copy_from_slice(0, data)
-            .expect("Failed to copy from slice");
+            .map_err(|e| SinkError::OnWrite(e.to_string()))?;
 
         self.appsrc
             .push_buffer(buffer)