diff --git a/src/pipewire/filter.c b/src/pipewire/filter.c
index ea3e74cd2..58dd4c998 100644
--- a/src/pipewire/filter.c
+++ b/src/pipewire/filter.c
@@ -1389,6 +1389,28 @@ static void free_port(struct filter *impl, struct port *port)
 	free(port);
 }
 
+static void filter_free(struct pw_filter *filter)
+{
+	struct filter *impl = SPA_CONTAINER_OF(filter, struct filter, this);
+
+	pw_log_debug("%p: free", filter);
+	clear_params(impl, NULL, SPA_ID_INVALID);
+
+	free(filter->error);
+
+	pw_properties_free(filter->properties);
+
+	pw_map_clear(&impl->ports[SPA_DIRECTION_INPUT]);
+	pw_map_clear(&impl->ports[SPA_DIRECTION_OUTPUT]);
+
+	free(filter->name);
+
+	if (impl->data.context)
+		pw_context_destroy(impl->data.context);
+
+	free(impl);
+}
+
 SPA_EXPORT
 void pw_filter_destroy(struct pw_filter *filter)
 {
@@ -1411,26 +1433,10 @@ void pw_filter_destroy(struct pw_filter *filter)
 		spa_hook_remove(&filter->core_listener);
 		spa_list_remove(&filter->link);
 	}
-
-	clear_params(impl, NULL, SPA_ID_INVALID);
-
-	pw_log_debug("%p: free", filter);
-	free(filter->error);
-
-	pw_properties_free(filter->properties);
-
 	spa_hook_list_clean(&impl->hooks);
 	spa_hook_list_clean(&filter->listener_list);
 
-	pw_map_clear(&impl->ports[SPA_DIRECTION_INPUT]);
-	pw_map_clear(&impl->ports[SPA_DIRECTION_OUTPUT]);
-
-	free(filter->name);
-
-	if (impl->data.context)
-		pw_context_destroy(impl->data.context);
-
-	free(impl);
+	filter_free(filter);
 }
 
 static int
diff --git a/src/pipewire/stream.c b/src/pipewire/stream.c
index 20b9e90f2..cfe01d079 100644
--- a/src/pipewire/stream.c
+++ b/src/pipewire/stream.c
@@ -715,8 +715,9 @@ static int impl_send_command(void *object, const struct spa_command *command)
 	case SPA_NODE_COMMAND_Suspend:
 	case SPA_NODE_COMMAND_Flush:
 	case SPA_NODE_COMMAND_Pause:
-		pw_loop_invoke(impl->main_loop,
-			NULL, 0, NULL, 0, false, impl);
+		/* this ensures we don't have any pending invokes in the queue after we
+		 * emit the state change and/or command. */
+		pw_loop_invoke(impl->main_loop, NULL, 0, NULL, 0, false, impl);
 		if (stream->state == PW_STREAM_STATE_STREAMING && id != SPA_NODE_COMMAND_Flush) {
 			pw_log_debug("%p: pause", stream);
 			stream_set_state(stream, PW_STREAM_STATE_PAUSED, 0, NULL);
@@ -1748,11 +1749,35 @@ static int stream_disconnect(struct stream *impl)
 	return 0;
 }
 
+static void stream_free(struct pw_stream *stream)
+{
+	struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this);
+	struct control *c;
+
+	pw_log_debug("%p: free", stream);
+	clear_params(impl, SPA_ID_INVALID, 0);
+
+	free(stream->error);
+
+	pw_properties_free(stream->properties);
+
+	free(stream->name);
+
+	spa_list_consume(c, &stream->controls, link) {
+		spa_list_remove(&c->link);
+		free(c);
+	}
+	if (impl->data.context)
+		pw_context_destroy(impl->data.context);
+
+	pw_properties_free(impl->port_props);
+	free(impl);
+}
+
 SPA_EXPORT
 void pw_stream_destroy(struct pw_stream *stream)
 {
 	struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this);
-	struct control *c;
 
 	ensure_loop(impl->main_loop, return);
 
@@ -1768,29 +1793,10 @@ void pw_stream_destroy(struct pw_stream *stream)
 		spa_list_remove(&stream->link);
 		stream->core = NULL;
 	}
-
-	clear_params(impl, SPA_ID_INVALID, 0);
-
-	pw_log_debug("%p: free", stream);
-	free(stream->error);
-
-	pw_properties_free(stream->properties);
-
-	free(stream->name);
-
-	spa_list_consume(c, &stream->controls, link) {
-		spa_list_remove(&c->link);
-		free(c);
-	}
-
 	spa_hook_list_clean(&impl->hooks);
 	spa_hook_list_clean(&stream->listener_list);
 
-	if (impl->data.context)
-		pw_context_destroy(impl->data.context);
-
-	pw_properties_free(impl->port_props);
-	free(impl);
+	stream_free(stream);
 }
 
 static int