Update to use the new pad builders for safely setting pad functions

Only two uses of unsafely setting the pad functions is left:
- fallbacksrc for overriding the chain function of the proxy pad of a
  ghost pad
- threadshare for overriding the pad functions after creationg, which
  probably needs some fixing at some point

3 files changed, 316 insertions, 304 deletions

diff --git a/generic/sodium/src/decrypter.rs b/generic/sodium/src/decrypter.rs
index a52a83bd0..87a8fcbc5 100644
--- a/generic/sodium/src/decrypter.rs
+++ b/generic/sodium/src/decrypter.rs
@@ -269,39 +269,6 @@ struct Decrypter {
 }
 
 impl Decrypter {
-    fn set_pad_functions(_sinkpad: &gst::Pad, srcpad: &gst::Pad) {
-        srcpad.set_getrange_function(|pad, parent, offset, buffer, size| {
-            Decrypter::catch_panic_pad_function(
-                parent,
-                || Err(gst::FlowError::Error),
-                |decrypter, element| decrypter.get_range(pad, element, offset, buffer, size),
-            )
-        });
-
-        srcpad.set_activatemode_function(|pad, parent, mode, active| {
-            Decrypter::catch_panic_pad_function(
-                parent,
-                || {
-                    Err(gst_loggable_error!(
-                        CAT,
-                        "Panic activating srcpad with mode"
-                    ))
-                },
-                |decrypter, element| {
-                    decrypter.src_activatemode_function(pad, element, mode, active)
-                },
-            )
-        });
-
-        srcpad.set_query_function(|pad, parent, query| {
-            Decrypter::catch_panic_pad_function(
-                parent,
-                || false,
-                |decrypter, element| decrypter.src_query(pad, element, query),
-            )
-        });
-    }
-
     fn src_activatemode_function(
         &self,
         _pad: &gst::Pad,
@@ -597,10 +564,39 @@ impl ObjectSubclass for Decrypter {
     fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
         let templ = klass.get_pad_template("sink").unwrap();
         let sinkpad = gst::Pad::from_template(&templ, Some("sink"));
+
         let templ = klass.get_pad_template("src").unwrap();
-        let srcpad = gst::Pad::from_template(&templ, Some("src"));
+        let srcpad = gst::Pad::builder_with_template(&templ, Some("src"))
+            .getrange_function(|pad, parent, offset, buffer, size| {
+                Decrypter::catch_panic_pad_function(
+                    parent,
+                    || Err(gst::FlowError::Error),
+                    |decrypter, element| decrypter.get_range(pad, element, offset, buffer, size),
+                )
+            })
+            .activatemode_function(|pad, parent, mode, active| {
+                Decrypter::catch_panic_pad_function(
+                    parent,
+                    || {
+                        Err(gst_loggable_error!(
+                            CAT,
+                            "Panic activating srcpad with mode"
+                        ))
+                    },
+                    |decrypter, element| {
+                        decrypter.src_activatemode_function(pad, element, mode, active)
+                    },
+                )
+            })
+            .query_function(|pad, parent, query| {
+                Decrypter::catch_panic_pad_function(
+                    parent,
+                    || false,
+                    |decrypter, element| decrypter.src_query(pad, element, query),
+                )
+            })
+            .build();
 
-        Decrypter::set_pad_functions(&sinkpad, &srcpad);
         let props = Mutex::new(Props::default());
         let state = Mutex::new(None);
 
diff --git a/generic/sodium/src/encrypter.rs b/generic/sodium/src/encrypter.rs
index 129967bdf..2acf99061 100644
--- a/generic/sodium/src/encrypter.rs
+++ b/generic/sodium/src/encrypter.rs
@@ -198,38 +198,6 @@ struct Encrypter {
 }
 
 impl Encrypter {
-    fn set_pad_functions(sinkpad: &gst::Pad, srcpad: &gst::Pad) {
-        sinkpad.set_chain_function(|pad, parent, buffer| {
-            Encrypter::catch_panic_pad_function(
-                parent,
-                || Err(gst::FlowError::Error),
-                |encrypter, element| encrypter.sink_chain(pad, element, buffer),
-            )
-        });
-        sinkpad.set_event_function(|pad, parent, event| {
-            Encrypter::catch_panic_pad_function(
-                parent,
-                || false,
-                |encrypter, element| encrypter.sink_event(pad, element, event),
-            )
-        });
-
-        srcpad.set_query_function(|pad, parent, query| {
-            Encrypter::catch_panic_pad_function(
-                parent,
-                || false,
-                |encrypter, element| encrypter.src_query(pad, element, query),
-            )
-        });
-        srcpad.set_event_function(|pad, parent, event| {
-            Encrypter::catch_panic_pad_function(
-                parent,
-                || false,
-                |encrypter, element| encrypter.src_event(pad, element, event),
-            )
-        });
-    }
-
     fn sink_chain(
         &self,
         pad: &gst::Pad,
@@ -425,11 +393,41 @@ impl ObjectSubclass for Encrypter {
 
     fn with_class(klass: &subclass::simple::ClassStruct<Self>) -> Self {
         let templ = klass.get_pad_template("sink").unwrap();
-        let sinkpad = gst::Pad::from_template(&templ, Some("sink"));
+        let sinkpad = gst::Pad::builder_with_template(&templ, Some("sink"))
+            .chain_function(|pad, parent, buffer| {
+                Encrypter::catch_panic_pad_function(
+                    parent,
+                    || Err(gst::FlowError::Error),
+                    |encrypter, element| encrypter.sink_chain(pad, element, buffer),
+                )
+            })
+            .event_function(|pad, parent, event| {
+                Encrypter::catch_panic_pad_function(
+                    parent,
+                    || false,
+                    |encrypter, element| encrypter.sink_event(pad, element, event),
+                )
+            })
+            .build();
+
         let templ = klass.get_pad_template("src").unwrap();
-        let srcpad = gst::Pad::from_template(&templ, Some("src"));
+        let srcpad = gst::Pad::builder_with_template(&templ, Some("src"))
+            .query_function(|pad, parent, query| {
+                Encrypter::catch_panic_pad_function(
+                    parent,
+                    || false,
+                    |encrypter, element| encrypter.src_query(pad, element, query),
+                )
+            })
+            .event_function(|pad, parent, event| {
+                Encrypter::catch_panic_pad_function(
+                    parent,
+                    || false,
+                    |encrypter, element| encrypter.src_event(pad, element, event),
+                )
+            })
+            .build();
 
-        Encrypter::set_pad_functions(&sinkpad, &srcpad);
         let props = Mutex::new(Props::default());
         let state = Mutex::new(None);
 
diff --git a/generic/threadshare/src/runtime/pad.rs b/generic/threadshare/src/runtime/pad.rs
index d42663ef6..89e5e01d4 100644
--- a/generic/threadshare/src/runtime/pad.rs
+++ b/generic/threadshare/src/runtime/pad.rs
@@ -385,69 +385,71 @@ impl PadSrc {
     }
 
     fn init_pad_functions<H: PadSrcHandler>(&self, handler: H) {
-        let handler_clone = handler.clone();
-        let inner_arc = Arc::clone(&self.0);
-        self.0
-            .gst_pad()
-            .set_activate_function(move |gst_pad, parent| {
-                let handler = handler_clone.clone();
-                let inner_arc = inner_arc.clone();
-                H::ElementImpl::catch_panic_pad_function(
-                    parent,
-                    || {
-                        gst_error!(RUNTIME_CAT, obj: gst_pad, "Panic in PadSrc activate");
-                        Err(gst_loggable_error!(RUNTIME_CAT, "Panic in PadSrc activate"))
-                    },
-                    move |imp, element| {
-                        let this_ref = PadSrcRef::new(inner_arc);
-                        handler.src_activate(&this_ref, imp, element)
-                    },
-                )
-            });
-
-        let handler_clone = handler.clone();
-        let inner_arc = Arc::clone(&self.0);
-        self.gst_pad()
-            .set_activatemode_function(move |gst_pad, parent, mode, active| {
-                let handler = handler_clone.clone();
-                let inner_arc = inner_arc.clone();
-                H::ElementImpl::catch_panic_pad_function(
-                    parent,
-                    || {
-                        gst_error!(RUNTIME_CAT, obj: gst_pad, "Panic in PadSrc activatemode");
-                        Err(gst_loggable_error!(
-                            RUNTIME_CAT,
-                            "Panic in PadSrc activatemode"
-                        ))
-                    },
-                    move |imp, element| {
-                        let this_ref = PadSrcRef::new(inner_arc);
-                        this_ref.activate_mode_hook(mode, active)?;
-                        handler.src_activatemode(&this_ref, imp, element, mode, active)
-                    },
-                )
-            });
-
-        // No need to `set_event_function` since `set_event_full_function`
-        // overrides it and dispatches to `src_event` when necessary
-        let handler_clone = handler.clone();
-        let inner_arc = Arc::clone(&self.0);
-        self.gst_pad()
-            .set_event_full_function(move |_gst_pad, parent, event| {
-                let handler = handler_clone.clone();
-                let inner_arc = inner_arc.clone();
-                H::ElementImpl::catch_panic_pad_function(
-                    parent,
-                    || Err(FlowError::Error),
-                    move |imp, element| {
-                        let this_ref = PadSrcRef::new(inner_arc);
-                        handler.src_event_full(&this_ref, imp, &element, event)
-                    },
-                )
-            });
-
-        let inner_arc = Arc::clone(&self.0);
-        self.gst_pad()
+        // FIXME: Do this better
+        unsafe {
+            let handler_clone = handler.clone();
+            let inner_arc = Arc::clone(&self.0);
+            self.0
+                .gst_pad()
+                .set_activate_function(move |gst_pad, parent| {
+                    let handler = handler_clone.clone();
+                    let inner_arc = inner_arc.clone();
+                    H::ElementImpl::catch_panic_pad_function(
+                        parent,
+                        || {
+                            gst_error!(RUNTIME_CAT, obj: gst_pad, "Panic in PadSrc activate");
+                            Err(gst_loggable_error!(RUNTIME_CAT, "Panic in PadSrc activate"))
+                        },
+                        move |imp, element| {
+                            let this_ref = PadSrcRef::new(inner_arc);
+                            handler.src_activate(&this_ref, imp, element)
+                        },
+                    )
+                });
+
+            let handler_clone = handler.clone();
+            let inner_arc = Arc::clone(&self.0);
+            self.gst_pad()
+                .set_activatemode_function(move |gst_pad, parent, mode, active| {
+                    let handler = handler_clone.clone();
+                    let inner_arc = inner_arc.clone();
+                    H::ElementImpl::catch_panic_pad_function(
+                        parent,
+                        || {
+                            gst_error!(RUNTIME_CAT, obj: gst_pad, "Panic in PadSrc activatemode");
+                            Err(gst_loggable_error!(
+                                RUNTIME_CAT,
+                                "Panic in PadSrc activatemode"
+                            ))
+                        },
+                        move |imp, element| {
+                            let this_ref = PadSrcRef::new(inner_arc);
+                            this_ref.activate_mode_hook(mode, active)?;
+                            handler.src_activatemode(&this_ref, imp, element, mode, active)
+                        },
+                    )
+                });
+
+            // No need to `set_event_function` since `set_event_full_function`
+            // overrides it and dispatches to `src_event` when necessary
+            let handler_clone = handler.clone();
+            let inner_arc = Arc::clone(&self.0);
+            self.gst_pad()
+                .set_event_full_function(move |_gst_pad, parent, event| {
+                    let handler = handler_clone.clone();
+                    let inner_arc = inner_arc.clone();
+                    H::ElementImpl::catch_panic_pad_function(
+                        parent,
+                        || Err(FlowError::Error),
+                        move |imp, element| {
+                            let this_ref = PadSrcRef::new(inner_arc);
+                            handler.src_event_full(&this_ref, imp, &element, event)
+                        },
+                    )
+                });
+
+            let inner_arc = Arc::clone(&self.0);
+            self.gst_pad()
             .set_query_function(move |_gst_pad, parent, query| {
                 let handler = handler.clone();
                 let inner_arc = inner_arc.clone();
@@ -465,25 +467,29 @@ impl PadSrc {
                     },
                 )
             });
+        }
     }
 }
 
 impl Drop for PadSrc {
     fn drop(&mut self) {
-        self.gst_pad()
-            .set_activate_function(move |_gst_pad, _parent| {
-                Err(gst_loggable_error!(RUNTIME_CAT, "PadSrc no longer exists"))
-            });
-        self.gst_pad()
-            .set_activatemode_function(move |_gst_pad, _parent, _mode, _active| {
-                Err(gst_loggable_error!(RUNTIME_CAT, "PadSrc no longer exists"))
-            });
-        self.gst_pad()
-            .set_event_function(move |_gst_pad, _parent, _event| false);
-        self.gst_pad()
-            .set_event_full_function(move |_gst_pad, _parent, _event| Err(FlowError::Flushing));
-        self.gst_pad()
-            .set_query_function(move |_gst_pad, _parent, _query| false);
+        // FIXME: Do this better
+        unsafe {
+            self.gst_pad()
+                .set_activate_function(move |_gst_pad, _parent| {
+                    Err(gst_loggable_error!(RUNTIME_CAT, "PadSrc no longer exists"))
+                });
+            self.gst_pad()
+                .set_activatemode_function(move |_gst_pad, _parent, _mode, _active| {
+                    Err(gst_loggable_error!(RUNTIME_CAT, "PadSrc no longer exists"))
+                });
+            self.gst_pad()
+                .set_event_function(move |_gst_pad, _parent, _event| false);
+            self.gst_pad()
+                .set_event_full_function(move |_gst_pad, _parent, _event| Err(FlowError::Flushing));
+            self.gst_pad()
+                .set_query_function(move |_gst_pad, _parent, _query| false);
+        }
     }
 }
 
@@ -781,134 +787,64 @@ impl PadSink {
     }
 
     fn init_pad_functions<H: PadSinkHandler>(&self, handler: H) {
-        let handler_clone = handler.clone();
-        let inner_arc = Arc::clone(&self.0);
-        self.gst_pad()
-            .set_activate_function(move |gst_pad, parent| {
-                let handler = handler_clone.clone();
-                let inner_arc = inner_arc.clone();
-                H::ElementImpl::catch_panic_pad_function(
-                    parent,
-                    || {
-                        gst_error!(RUNTIME_CAT, obj: gst_pad, "Panic in PadSink activate");
-                        Err(gst_loggable_error!(
-                            RUNTIME_CAT,
-                            "Panic in PadSink activate"
-                        ))
-                    },
-                    move |imp, element| {
-                        let this_ref = PadSinkRef::new(inner_arc);
-                        handler.sink_activate(&this_ref, imp, element)
-                    },
-                )
-            });
-
-        let handler_clone = handler.clone();
-        let inner_arc = Arc::clone(&self.0);
-        self.gst_pad()
-            .set_activatemode_function(move |gst_pad, parent, mode, active| {
-                let handler = handler_clone.clone();
-                let inner_arc = inner_arc.clone();
-                H::ElementImpl::catch_panic_pad_function(
-                    parent,
-                    || {
-                        gst_error!(RUNTIME_CAT, obj: gst_pad, "Panic in PadSink activatemode");
-                        Err(gst_loggable_error!(
-                            RUNTIME_CAT,
-                            "Panic in PadSink activatemode"
-                        ))
-                    },
-                    move |imp, element| {
-                        let this_ref = PadSinkRef::new(inner_arc);
-                        this_ref.activate_mode_hook(mode, active)?;
-
-                        handler.sink_activatemode(&this_ref, imp, element, mode, active)
-                    },
-                )
-            });
-
-        let handler_clone = handler.clone();
-        let inner_arc = Arc::clone(&self.0);
-        self.gst_pad()
-            .set_chain_function(move |_gst_pad, parent, buffer| {
-                let handler = handler_clone.clone();
-                let inner_arc = inner_arc.clone();
-                H::ElementImpl::catch_panic_pad_function(
-                    parent,
-                    || Err(FlowError::Error),
-                    move |imp, element| {
-                        if Context::current_has_sub_tasks() {
-                            let this_weak = PadSinkWeak(Arc::downgrade(&inner_arc));
-                            let handler = handler.clone();
-                            let element = element.clone();
-                            let delayed_fut = async move {
-                                let imp =
-                                    <H::ElementImpl as ObjectSubclass>::from_instance(&element);
-                                let this_ref =
-                                    this_weak.upgrade().ok_or(gst::FlowError::Flushing)?;
-                                handler.sink_chain(&this_ref, imp, &element, buffer).await
-                            };
-                            let _ = Context::add_sub_task(delayed_fut.map(|res| res.map(drop)));
-
-                            Ok(gst::FlowSuccess::Ok)
-                        } else {
+        // FIXME: Do this better
+        unsafe {
+            let handler_clone = handler.clone();
+            let inner_arc = Arc::clone(&self.0);
+            self.gst_pad()
+                .set_activate_function(move |gst_pad, parent| {
+                    let handler = handler_clone.clone();
+                    let inner_arc = inner_arc.clone();
+                    H::ElementImpl::catch_panic_pad_function(
+                        parent,
+                        || {
+                            gst_error!(RUNTIME_CAT, obj: gst_pad, "Panic in PadSink activate");
+                            Err(gst_loggable_error!(
+                                RUNTIME_CAT,
+                                "Panic in PadSink activate"
+                            ))
+                        },
+                        move |imp, element| {
                             let this_ref = PadSinkRef::new(inner_arc);
-                            let chain_fut = handler.sink_chain(&this_ref, imp, &element, buffer);
-                            this_ref.handle_future(chain_fut)
-                        }
-                    },
-                )
-            });
-
-        let handler_clone = handler.clone();
-        let inner_arc = Arc::clone(&self.0);
-        self.gst_pad()
-            .set_chain_list_function(move |_gst_pad, parent, list| {
-                let handler = handler_clone.clone();
-                let inner_arc = inner_arc.clone();
-                H::ElementImpl::catch_panic_pad_function(
-                    parent,
-                    || Err(FlowError::Error),
-                    move |imp, element| {
-                        if Context::current_has_sub_tasks() {
-                            let this_weak = PadSinkWeak(Arc::downgrade(&inner_arc));
-                            let handler = handler.clone();
-                            let element = element.clone();
-                            let delayed_fut = async move {
-                                let imp =
-                                    <H::ElementImpl as ObjectSubclass>::from_instance(&element);
-                                let this_ref =
-                                    this_weak.upgrade().ok_or(gst::FlowError::Flushing)?;
-                                handler
-                                    .sink_chain_list(&this_ref, imp, &element, list)
-                                    .await
-                            };
-                            let _ = Context::add_sub_task(delayed_fut.map(|res| res.map(drop)));
-
-                            Ok(gst::FlowSuccess::Ok)
-                        } else {
+                            handler.sink_activate(&this_ref, imp, element)
+                        },
+                    )
+                });
+
+            let handler_clone = handler.clone();
+            let inner_arc = Arc::clone(&self.0);
+            self.gst_pad()
+                .set_activatemode_function(move |gst_pad, parent, mode, active| {
+                    let handler = handler_clone.clone();
+                    let inner_arc = inner_arc.clone();
+                    H::ElementImpl::catch_panic_pad_function(
+                        parent,
+                        || {
+                            gst_error!(RUNTIME_CAT, obj: gst_pad, "Panic in PadSink activatemode");
+                            Err(gst_loggable_error!(
+                                RUNTIME_CAT,
+                                "Panic in PadSink activatemode"
+                            ))
+                        },
+                        move |imp, element| {
                             let this_ref = PadSinkRef::new(inner_arc);
-                            let chain_list_fut =
-                                handler.sink_chain_list(&this_ref, imp, &element, list);
-                            this_ref.handle_future(chain_list_fut)
-                        }
-                    },
-                )
-            });
-
-        // No need to `set_event_function` since `set_event_full_function`
-        // overrides it and dispatches to `sink_event` when necessary
-        let handler_clone = handler.clone();
-        let inner_arc = Arc::clone(&self.0);
-        self.gst_pad()
-            .set_event_full_function(move |_gst_pad, parent, event| {
-                let handler = handler_clone.clone();
-                let inner_arc = inner_arc.clone();
-                H::ElementImpl::catch_panic_pad_function(
-                    parent,
-                    || Err(FlowError::Error),
-                    move |imp, element| {
-                        if event.is_serialized() {
+                            this_ref.activate_mode_hook(mode, active)?;
+
+                            handler.sink_activatemode(&this_ref, imp, element, mode, active)
+                        },
+                    )
+                });
+
+            let handler_clone = handler.clone();
+            let inner_arc = Arc::clone(&self.0);
+            self.gst_pad()
+                .set_chain_function(move |_gst_pad, parent, buffer| {
+                    let handler = handler_clone.clone();
+                    let inner_arc = inner_arc.clone();
+                    H::ElementImpl::catch_panic_pad_function(
+                        parent,
+                        || Err(FlowError::Error),
+                        move |imp, element| {
                             if Context::current_has_sub_tasks() {
                                 let this_weak = PadSinkWeak(Arc::downgrade(&inner_arc));
                                 let handler = handler.clone();
@@ -918,9 +854,42 @@ impl PadSink {
                                         <H::ElementImpl as ObjectSubclass>::from_instance(&element);
                                     let this_ref =
                                         this_weak.upgrade().ok_or(gst::FlowError::Flushing)?;
+                                    handler.sink_chain(&this_ref, imp, &element, buffer).await
+                                };
+                                let _ = Context::add_sub_task(delayed_fut.map(|res| res.map(drop)));
 
+                                Ok(gst::FlowSuccess::Ok)
+                            } else {
+                                let this_ref = PadSinkRef::new(inner_arc);
+                                let chain_fut =
+                                    handler.sink_chain(&this_ref, imp, &element, buffer);
+                                this_ref.handle_future(chain_fut)
+                            }
+                        },
+                    )
+                });
+
+            let handler_clone = handler.clone();
+            let inner_arc = Arc::clone(&self.0);
+            self.gst_pad()
+                .set_chain_list_function(move |_gst_pad, parent, list| {
+                    let handler = handler_clone.clone();
+                    let inner_arc = inner_arc.clone();
+                    H::ElementImpl::catch_panic_pad_function(
+                        parent,
+                        || Err(FlowError::Error),
+                        move |imp, element| {
+                            if Context::current_has_sub_tasks() {
+                                let this_weak = PadSinkWeak(Arc::downgrade(&inner_arc));
+                                let handler = handler.clone();
+                                let element = element.clone();
+                                let delayed_fut = async move {
+                                    let imp =
+                                        <H::ElementImpl as ObjectSubclass>::from_instance(&element);
+                                    let this_ref =
+                                        this_weak.upgrade().ok_or(gst::FlowError::Flushing)?;
                                     handler
-                                        .sink_event_full_serialized(&this_ref, imp, &element, event)
+                                        .sink_chain_list(&this_ref, imp, &element, list)
                                         .await
                                 };
                                 let _ = Context::add_sub_task(delayed_fut.map(|res| res.map(drop)));
@@ -928,20 +897,65 @@ impl PadSink {
                                 Ok(gst::FlowSuccess::Ok)
                             } else {
                                 let this_ref = PadSinkRef::new(inner_arc);
-                                let event_fut = handler
-                                    .sink_event_full_serialized(&this_ref, imp, &element, event);
-                                this_ref.handle_future(event_fut)
+                                let chain_list_fut =
+                                    handler.sink_chain_list(&this_ref, imp, &element, list);
+                                this_ref.handle_future(chain_list_fut)
                             }
-                        } else {
-                            let this_ref = PadSinkRef::new(inner_arc);
-                            handler.sink_event_full(&this_ref, imp, &element, event)
-                        }
-                    },
-                )
-            });
+                        },
+                    )
+                });
+
+            // No need to `set_event_function` since `set_event_full_function`
+            // overrides it and dispatches to `sink_event` when necessary
+            let handler_clone = handler.clone();
+            let inner_arc = Arc::clone(&self.0);
+            self.gst_pad()
+                .set_event_full_function(move |_gst_pad, parent, event| {
+                    let handler = handler_clone.clone();
+                    let inner_arc = inner_arc.clone();
+                    H::ElementImpl::catch_panic_pad_function(
+                        parent,
+                        || Err(FlowError::Error),
+                        move |imp, element| {
+                            if event.is_serialized() {
+                                if Context::current_has_sub_tasks() {
+                                    let this_weak = PadSinkWeak(Arc::downgrade(&inner_arc));
+                                    let handler = handler.clone();
+                                    let element = element.clone();
+                                    let delayed_fut = async move {
+                                        let imp = <H::ElementImpl as ObjectSubclass>::from_instance(
+                                            &element,
+                                        );
+                                        let this_ref =
+                                            this_weak.upgrade().ok_or(gst::FlowError::Flushing)?;
+
+                                        handler
+                                            .sink_event_full_serialized(
+                                                &this_ref, imp, &element, event,
+                                            )
+                                            .await
+                                    };
+                                    let _ =
+                                        Context::add_sub_task(delayed_fut.map(|res| res.map(drop)));
+
+                                    Ok(gst::FlowSuccess::Ok)
+                                } else {
+                                    let this_ref = PadSinkRef::new(inner_arc);
+                                    let event_fut = handler.sink_event_full_serialized(
+                                        &this_ref, imp, &element, event,
+                                    );
+                                    this_ref.handle_future(event_fut)
+                                }
+                            } else {
+                                let this_ref = PadSinkRef::new(inner_arc);
+                                handler.sink_event_full(&this_ref, imp, &element, event)
+                            }
+                        },
+                    )
+                });
 
-        let inner_arc = Arc::clone(&self.0);
-        self.gst_pad()
+            let inner_arc = Arc::clone(&self.0);
+            self.gst_pad()
             .set_query_function(move |_gst_pad, parent, query| {
                 let handler = handler.clone();
                 let inner_arc = inner_arc.clone();
@@ -959,29 +973,33 @@ impl PadSink {
                     },
                 )
             });
+        }
     }
 }
 
 impl Drop for PadSink {
     fn drop(&mut self) {
-        self.gst_pad()
-            .set_activate_function(move |_gst_pad, _parent| {
-                Err(gst_loggable_error!(RUNTIME_CAT, "PadSink no longer exists"))
-            });
-        self.gst_pad()
-            .set_activatemode_function(move |_gst_pad, _parent, _mode, _active| {
-                Err(gst_loggable_error!(RUNTIME_CAT, "PadSink no longer exists"))
-            });
-        self.gst_pad()
-            .set_chain_function(move |_gst_pad, _parent, _buffer| Err(FlowError::Flushing));
-        self.gst_pad()
-            .set_chain_list_function(move |_gst_pad, _parent, _list| Err(FlowError::Flushing));
-        self.gst_pad()
-            .set_event_function(move |_gst_pad, _parent, _event| false);
-        self.gst_pad()
-            .set_event_full_function(move |_gst_pad, _parent, _event| Err(FlowError::Flushing));
-        self.gst_pad()
-            .set_query_function(move |_gst_pad, _parent, _query| false);
+        // FIXME: Do this better
+        unsafe {
+            self.gst_pad()
+                .set_activate_function(move |_gst_pad, _parent| {
+                    Err(gst_loggable_error!(RUNTIME_CAT, "PadSink no longer exists"))
+                });
+            self.gst_pad()
+                .set_activatemode_function(move |_gst_pad, _parent, _mode, _active| {
+                    Err(gst_loggable_error!(RUNTIME_CAT, "PadSink no longer exists"))
+                });
+            self.gst_pad()
+                .set_chain_function(move |_gst_pad, _parent, _buffer| Err(FlowError::Flushing));
+            self.gst_pad()
+                .set_chain_list_function(move |_gst_pad, _parent, _list| Err(FlowError::Flushing));
+            self.gst_pad()
+                .set_event_function(move |_gst_pad, _parent, _event| false);
+            self.gst_pad()
+                .set_event_full_function(move |_gst_pad, _parent, _event| Err(FlowError::Flushing));
+            self.gst_pad()
+                .set_query_function(move |_gst_pad, _parent, _query| false);
+        }
     }
 }