diff --git a/spa/include/spa/graph.h b/spa/include/spa/graph.h
index 637c0d819..0cb951086 100644
--- a/spa/include/spa/graph.h
+++ b/spa/include/spa/graph.h
@@ -42,16 +42,16 @@ struct spa_graph {
 	struct spa_list nodes;
 };
 
-#define SPA_VERSION_GRAPH_NODE_METHODS	0
 struct spa_graph_node_methods {
+#define SPA_VERSION_GRAPH_NODE_METHODS	0
 	uint32_t version;
 
 	int (*process_input) (struct spa_graph_node *node, void *user_data);
 	int (*process_output) (struct spa_graph_node *node, void *user_data);
 };
 
-#define SPA_VERSION_GRAPH_PORT_METHODS	0
 struct spa_graph_port_methods {
+#define SPA_VERSION_GRAPH_PORT_METHODS	0
 	uint32_t version;
 
 	int (*reuse_buffer) (struct spa_graph_port *port, uint32_t buffer_id, void *user_data);
diff --git a/spa/include/spa/loop.h b/spa/include/spa/loop.h
index 18e719648..c613780d6 100644
--- a/spa/include/spa/loop.h
+++ b/spa/include/spa/loop.h
@@ -152,7 +152,8 @@ typedef void (*spa_source_io_func_t) (struct spa_loop_utils *utils,
 typedef void (*spa_source_idle_func_t) (struct spa_loop_utils *utils,
 					struct spa_source *source, void *data);
 typedef void (*spa_source_event_func_t) (struct spa_loop_utils *utils,
-					 struct spa_source *source, void *data);
+					 struct spa_source *source,
+					 uint64_t count, void *data);
 typedef void (*spa_source_timer_func_t) (struct spa_loop_utils *utils,
 					 struct spa_source *source, void *data);
 typedef void (*spa_source_signal_func_t) (struct spa_loop_utils *utils,
diff --git a/spa/plugins/support/loop.c b/spa/plugins/support/loop.c
index db71a61b1..9afb45d4e 100644
--- a/spa/plugins/support/loop.c
+++ b/spa/plugins/support/loop.c
@@ -263,7 +263,7 @@ loop_invoke(struct spa_loop *loop,
 	return res;
 }
 
-static void wakeup_func(struct spa_loop_utils *utils, struct spa_source *source, void *data)
+static void wakeup_func(struct spa_loop_utils *utils, struct spa_source *source, uint64_t count, void *data)
 {
 	struct impl *impl = data;
 	uint32_t index;
@@ -458,7 +458,7 @@ static void source_event_func(struct spa_source *source)
 	if (read(source->fd, &count, sizeof(uint64_t)) != sizeof(uint64_t))
 		spa_log_warn(impl->impl->log, NAME " %p: failed to read event fd: %s", source, strerror(errno));
 
-	impl->func.event(&impl->impl->utils, source, source->data);
+	impl->func.event(&impl->impl->utils, source, count, source->data);
 }
 
 static struct spa_source *loop_add_event(struct spa_loop_utils *utils,
diff --git a/src/modules/module-protocol-native.c b/src/modules/module-protocol-native.c
index c3c058702..a1bbe746f 100644
--- a/src/modules/module-protocol-native.c
+++ b/src/modules/module-protocol-native.c
@@ -69,6 +69,7 @@ struct connection {
         struct pw_protocol_native_connection *connection;
 
         bool disconnecting;
+	bool flush_signaled;
         struct pw_listener need_flush;
         struct spa_source *flush_event;
 };
@@ -473,9 +474,11 @@ on_remote_data(struct spa_loop_utils *utils,
 }
 
 
-static void do_flush_event(struct spa_loop_utils *utils, struct spa_source *source, void *data)
+static void do_flush_event(struct spa_loop_utils *utils, struct spa_source *source,
+			   uint64_t count, void *data)
 {
         struct connection *impl = data;
+	impl->flush_signaled = false;
         if (impl->connection)
                 if (!pw_protocol_native_connection_flush(impl->connection))
                         impl->this.disconnect(&impl->this);
@@ -485,7 +488,11 @@ static void on_need_flush(struct pw_listener *listener, struct pw_protocol_nativ
 {
         struct connection *impl = SPA_CONTAINER_OF(listener, struct connection, need_flush);
         struct pw_remote *remote = impl->this.remote;
-        pw_loop_signal_event(remote->core->main_loop, impl->flush_event);
+
+	if (!impl->flush_signaled) {
+		impl->flush_signaled = true;
+		pw_loop_signal_event(remote->core->main_loop, impl->flush_event);
+	}
 }
 
 static int impl_connect_fd(struct pw_protocol_connection *conn, int fd)
diff --git a/src/pipewire/data-loop.c b/src/pipewire/data-loop.c
index 83da38a60..4998c8143 100644
--- a/src/pipewire/data-loop.c
+++ b/src/pipewire/data-loop.c
@@ -101,7 +101,7 @@ static void *do_loop(void *user_data)
 }
 
 
-static void do_stop(struct spa_loop_utils *utils, struct spa_source *source, void *data)
+static void do_stop(struct spa_loop_utils *utils, struct spa_source *source, uint64_t count, void *data)
 {
 	struct impl *impl = data;
 	impl->running = false;
diff --git a/src/pipewire/port.h b/src/pipewire/port.h
index f682a36db..5b5b226e8 100644
--- a/src/pipewire/port.h
+++ b/src/pipewire/port.h
@@ -47,9 +47,8 @@ enum pw_port_state {
 
 struct pw_port;
 
-#define PW_VERSION_PORT_IMPLEMENTATION 0
-
 struct pw_port_implementation {
+#define PW_VERSION_PORT_IMPLEMENTATION 0
 	uint32_t version;
 
 	int (*enum_formats) (struct pw_port *port,
diff --git a/src/pipewire/thread-loop.c b/src/pipewire/thread-loop.c
index d2c153a60..7f57e87ff 100644
--- a/src/pipewire/thread-loop.c
+++ b/src/pipewire/thread-loop.c
@@ -64,7 +64,7 @@ static const struct spa_loop_control_hooks impl_hooks = {
 	after,
 };
 
-static void do_stop(struct spa_loop_utils *utils, struct spa_source *source, void *data)
+static void do_stop(struct spa_loop_utils *utils, struct spa_source *source, uint64_t count, void *data)
 {
 	struct thread_loop *impl = data;
 	impl->running = false;
diff --git a/src/pipewire/work-queue.c b/src/pipewire/work-queue.c
index 5c309ea8d..5eb0c3873 100644
--- a/src/pipewire/work-queue.c
+++ b/src/pipewire/work-queue.c
@@ -46,7 +46,8 @@ struct impl {
 };
 /** \endcond */
 
-static void process_work_queue(struct spa_loop_utils *utils, struct spa_source *source, void *data)
+static void process_work_queue(struct spa_loop_utils *utils, struct spa_source *source,
+			       uint64_t count, void *data)
 {
 	struct impl *impl = data;
 	struct pw_work_queue *this = &impl->this;