diff --git a/src/pipewire/filter.c b/src/pipewire/filter.c
index 5ac7b1d83..934368080 100644
--- a/src/pipewire/filter.c
+++ b/src/pipewire/filter.c
@@ -969,10 +969,10 @@ static void call_process(struct filter *impl)
 {
 	pw_log_trace_fp("%p: call process", impl);
 	if (SPA_FLAG_IS_SET(impl->flags, PW_FILTER_FLAG_RT_PROCESS)) {
-		spa_callbacks_call_fast(&impl->rt_callbacks, struct pw_filter_events,
-				process, 0, impl->rt.position);
-	}
-	else {
+		if (impl->rt_callbacks.funcs)
+			spa_callbacks_call_fast(&impl->rt_callbacks, struct pw_filter_events,
+					process, 0, impl->rt.position);
+	} else {
 		pw_loop_invoke(impl->main_loop,
 			do_call_process, 1, NULL, 0, false, impl);
 	}
@@ -1436,10 +1436,19 @@ void pw_filter_destroy(struct pw_filter *filter)
 	free(impl);
 }
 
+static int
+do_remove_callbacks(struct spa_loop *loop,
+                 bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct filter *impl = user_data;
+	spa_zero(impl->rt_callbacks);
+	return 0;
+}
+
 static void hook_removed(struct spa_hook *hook)
 {
 	struct filter *impl = hook->priv;
-	spa_zero(impl->rt_callbacks);
+	pw_loop_invoke(impl->data_loop, do_remove_callbacks, 1, NULL, 0, true, impl);
 	hook->priv = NULL;
 	hook->removed = NULL;
 }
diff --git a/src/pipewire/stream.c b/src/pipewire/stream.c
index 5572bff29..c19a2c523 100644
--- a/src/pipewire/stream.c
+++ b/src/pipewire/stream.c
@@ -426,11 +426,13 @@ static inline void call_process(struct stream *impl)
 	pw_log_trace_fp("%p: call process rt:%u", impl, impl->process_rt);
 	if (impl->direction == SPA_DIRECTION_OUTPUT && update_requested(impl) <= 0)
 		return;
-	if (impl->process_rt)
-		spa_callbacks_call_fast(&impl->rt_callbacks, struct pw_stream_events, process, 0);
-	else
+	if (impl->process_rt) {
+		if (impl->rt_callbacks.funcs)
+			spa_callbacks_call_fast(&impl->rt_callbacks, struct pw_stream_events, process, 0);
+	} else {
 		pw_loop_invoke(impl->main_loop,
 			do_call_process, 1, NULL, 0, false, impl);
+	}
 }
 
 static int
@@ -1693,10 +1695,19 @@ void pw_stream_destroy(struct pw_stream *stream)
 	free(impl);
 }
 
+static int
+do_remove_callbacks(struct spa_loop *loop,
+                 bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct stream *impl = user_data;
+	spa_zero(impl->rt_callbacks);
+	return 0;
+}
+
 static void hook_removed(struct spa_hook *hook)
 {
 	struct stream *impl = hook->priv;
-	spa_zero(impl->rt_callbacks);
+	pw_loop_invoke(impl->data_loop, do_remove_callbacks, 1, NULL, 0, true, impl);
 	hook->priv = NULL;
 	hook->removed = NULL;
 }