Cleanup

include/pipewire.hpp

@@ -1 +1,112 @@
+#ifndef __GSR_PIPEWIRE_HPP__
+#define __GSR_PIPEWIRE_HPP__
+
+#include <pipewire/pipewire.h>
+#include <spa/param/audio/format-utils.h>
+#include <spa/debug/types.h>
+#include <spa/param/audio/type-info.h>
+
+#include <vector>
+#include <string>
+#include <optional>
+
+struct capture_config {
+    // The node_name to look for. If not set, then every node_name matches.
+    std::optional<std::string> name;
+
+    // Wether to look for an match (false) or match everything that does not
+    // match name (true).
+    bool exclude;
+
+    // Whether name refers to an application (false) or an input device (true).
+    bool device;
+
+    // The amount of channels to create.
+    int channels;
+};
+
+struct target_port {
+    uint32_t id;
+    uint32_t node_id;
+    const char *channel_name;
+};
+
+struct target_node {
+    uint32_t client_id;
+    uint32_t id;
+    const char *app_name;
+};
+
+struct target_input {
+    uint32_t id;
+};
+
+struct sink_port {
+    uint32_t id;
+    const char* channel;
+};
+
+struct capture_stream {
+        struct pw_core *core;
+
+        // The stream we will capture
+        struct pw_stream *stream;
+
+        // The context to use.
+        struct pw_context *context;
+
+        // Object to accessing global events.
+        struct pw_registry *registry;
+
+        // Listener for global events.
+        struct spa_hook registry_listener;
+
+        // The capture sink.
+        struct pw_proxy *sink_proxy;
+
+        // Listener for the sink events.
+        struct spa_hook sink_proxy_listener;
+
+        // The event loop to use.
+        struct pw_thread_loop *thread_loop;
+
+        // The id of the sink that we created.
+        uint32_t sink_id;
+
+        // The serial of the sink.
+        uint32_t sink_serial;
+
+        // Sequence number for forcing a server round trip
+        int seq;
+
+        std::vector<struct sink_port> sink_ports;
+        std::vector<struct target_node> nodes;
+        std::vector<struct target_port> ports;
+        std::vector<struct target_input> inputs;
+
+        struct capture_config config;
+
+        struct spa_audio_info format;
+};
+
+/*
+ * Returns whether the capture stream is ready to have other ports attached to it.
+ **/
+bool capture_stream_is_ready(struct capture_stream *data);
+
+/*
+ * Initialises the PipeWire API.
+ **/
 void init_pipewire();
+
+/*
+ * Creates a capture stream using the provided config.
+ **/
+struct capture_stream create_capture_stream(struct capture_config config);
+
+/*
+ * Frees the resources held by the capture stream.
+ **/
+void free_capture_stream(struct capture_stream *data);
+
+#endif /*__GSR_PIPEWIRE_HPP__*/

src/main.cpp

@@ -18,6 +18,7 @@ extern "C" {
 #include <mutex>
 #include <map>
 #include <signal.h>
+#include <optional>
 #include <sys/stat.h>
 #include <unistd.h>
 #include <sys/wait.h>
@@ -1376,6 +1377,20 @@ struct Arg {
 
 int main(int argc, char **argv) {
     init_pipewire();
+    /*struct capture_config config = {
+        .name = std::optional<std::string>("Firefox"),
+        .exclude = false,
+        .device = false,
+        .channels = 2,
+    };*/
+    struct capture_config config = {
+        .name = std::optional<std::string>("alsa_input.usb-DCMT_Technology_USB_Condenser_Microphone_214b206000000178-00.mono-fallback"),
+        .exclude = false,
+        .device = true,
+        .channels = 1,
+    };
+    auto cstream = create_capture_stream(config);
+    free_capture_stream(&cstream);
 
     return 0;
     signal(SIGINT, stop_handler);

src/pipewire.cpp

@@ -3,81 +3,13 @@
 #include <spa/debug/types.h>
 #include <spa/param/audio/type-info.h>
 #include <vector>
-
+#include <string>
-#define STR(x) #x
+#include <optional>
-#define AUDIO_CHANNELS 2
+#include "../include/pipewire.hpp"
-
-struct target_client {
-    const char *app_name;
-    const char *binary;
-    uint32_t id;
-
-    struct spa_hook client_listener;
-};
-
-struct target_port {
-    uint32_t id;
-    struct target_node *node;
-};
-
-struct target_node {
-    uint32_t client_id;
-    uint32_t id;
-    const char *app_name;
-
-    std::vector<struct target_port> ports;
-};
-
-struct sink_port {
-    uint32_t id;
-    const char* channel;
-};
-
-struct data {
-        struct pw_core *core;
-
-        // The stream we will capture
-        struct pw_stream *stream;
-
-        // The context to use.
-        struct pw_context *context;
-
-        // Object to accessing global events.
-        struct pw_registry *registry;
-
-        // Listener for global events.
-        struct spa_hook registry_listener;
-
-        // The capture sink.
-        struct pw_proxy *sink_proxy;
-
-        // Listener for the sink events.
-        struct spa_hook sink_proxy_listener;
-
-        // The event loop to use.
-        struct pw_thread_loop *thread_loop;
-
-        // The id of the sink that we created.
-        uint32_t sink_id;
-
-        // The serial of the sink.
-        uint32_t sink_serial;
-
-        // Sequence number for forcing a server round trip
-        int seq;
-
-        std::vector<struct sink_port> sink_ports;
-
-        std::vector<struct target_client> targets;
-        std::vector<struct target_node> nodes;
-        std::vector<struct target_port> ports;
- 
-        struct spa_audio_info format;
-};
 
 static void on_process(void *userdata)
 {
-        struct data *data = static_cast<struct data *>(userdata);
+        struct capture_stream *data = static_cast<struct capture_stream *>(userdata);
         struct pw_buffer *b;
         struct spa_buffer *buf;
  
@@ -90,7 +22,7 @@ static void on_process(void *userdata)
         if (buf->datas[0].data == NULL)
                 return;
  
-        printf("got a frame of size %d\n", buf->datas[0].chunk->size);
+        //printf("got a frame of size %d\n", buf->datas[0].chunk->size);
  
         pw_stream_queue_buffer(data->stream, b);
 }
@@ -98,7 +30,7 @@ static void on_process(void *userdata)
  
 static void on_param_changed(void *userdata, uint32_t id, const struct spa_pod *param)
 {
-        struct data *data = static_cast<struct data *>(userdata);
+        struct capture_stream *data = static_cast<struct capture_stream *>(userdata);
  
         if (param == NULL || id != SPA_PARAM_Format)
                 return;
@@ -121,16 +53,7 @@ static void on_param_changed(void *userdata, uint32_t id, const struct spa_pod *
 
 }
 
-void register_target_client(struct data *data, uint32_t id, const char* app_name) {
+void register_target_node(struct capture_stream *data, uint32_t id, uint32_t client_id, const char* app_name) {
-    struct target_client client = {};
-    client.binary = NULL;
-    client.app_name = strdup(app_name);
-    client.id = id;
-
-    data->targets.push_back(client);
-}
-
-void register_target_node(struct data *data, uint32_t id, uint32_t client_id, const char* app_name) {
     struct target_node node = {};
     node.app_name = strdup(app_name);
     node.id = id;
@@ -139,14 +62,114 @@ void register_target_node(struct data *data, uint32_t id, uint32_t client_id, co
     data->nodes.push_back(node);
 }
 
-void register_target_port(struct data *data, struct target_node *node, uint32_t id) {
+void register_target_port(struct capture_stream *data, uint32_t id, uint32_t node_id, const char *channel_name) {
     struct target_port port = {};
     port.id = id;
-    port.node = node;
+    port.node_id = node_id;
+    port.channel_name = strdup(channel_name);
 
     data->ports.push_back(port);
 }
 
+void register_target_input(struct capture_stream *data, uint32_t id) {
+    struct target_input port = {
+        .id = id,
+    };
+    data->inputs.push_back(port);
+}
+
+bool has_matching_node_or_input(struct capture_stream *capture, uint32_t node_id) {
+    // Find the corresponding node.
+    bool has_matching_node = false;
+    for (auto t : capture->nodes) {
+        if (t.id == node_id) {
+            has_matching_node = true;
+            break;
+        }
+    }
+    if (has_matching_node) {
+        return true;
+    }
+
+    // Find the corresponding input.
+    bool has_matching_input = false;
+    for (auto t : capture->inputs) {
+        if (t.id == node_id) {
+            has_matching_input = true;
+            break;
+        }
+    }
+    if (has_matching_input) {
+        return true;
+    }
+
+    return false;
+}
+
+bool connect_port_to_sink(struct capture_stream *data, uint32_t node_id, uint32_t id, const char *channel_name) {
+    // Find the corresponding node.
+    if (!has_matching_node_or_input(data, node_id)) {
+        printf("No matching node found\n");
+        return false;
+    }
+
+    // Find the correct sink port to attach to.
+    uint32_t sink_dst_port_id = 0;
+    for (auto sink_port : data->sink_ports) {
+        printf("%s = %s\n", sink_port.channel, channel_name);
+        if (strcmp(sink_port.channel, channel_name) == 0) {
+            sink_dst_port_id = sink_port.id;
+            break;
+        }
+    }
+    if (!sink_dst_port_id) {
+        return false;
+    }
+
+    // Connect the port to the sink.
+    struct pw_properties *link_props = pw_properties_new(
+        PW_KEY_OBJECT_LINGER, "false",
+        NULL
+    );
+    pw_properties_setf(link_props, PW_KEY_LINK_OUTPUT_NODE, "%u", node_id);
+    pw_properties_setf(link_props, PW_KEY_LINK_OUTPUT_PORT, "%u", id);
+
+    pw_properties_setf(link_props, PW_KEY_LINK_INPUT_NODE, "%u", data->sink_id);
+    pw_properties_setf(link_props, PW_KEY_LINK_INPUT_PORT, "%u", sink_dst_port_id);
+
+    printf(
+        "[DBG] Connecting (%d, %d) -> (%d, %d)\n",
+        node_id, id,
+        data->sink_id, sink_dst_port_id
+    );
+
+    struct pw_proxy *link_proxy = static_cast<struct pw_proxy *>(
+        pw_core_create_object(
+            data->core, "link-factory",
+            PW_TYPE_INTERFACE_Link, PW_VERSION_LINK, &link_props->dict, 0
+        )
+    );
+    data->seq = pw_core_sync(data->core, PW_ID_CORE, data->seq);
+    pw_properties_free(link_props);
+
+    if (!link_proxy) {
+        printf("[ERR] Failed to connect port %u of node %u to capture sink\n", id, node_id);
+        return false;
+    }
+}
+
+void connect_ports_to_sink(struct capture_stream *data) {
+    for (auto port : data->ports) {
+        printf("[DBG] Attempting to connect port %d\n", port.id);
+        connect_port_to_sink(
+            data,
+            port.node_id,
+            port.id,
+            port.channel_name
+        );
+    }
+}
+
 static void registry_event_global(void *raw_data, uint32_t id,
                 uint32_t permissions, const char *type, uint32_t version,
                 const struct spa_dict *props)
@@ -154,13 +177,12 @@ static void registry_event_global(void *raw_data, uint32_t id,
     if (!type || !props)
         return;
 
-    struct data *data = static_cast<struct data *>(raw_data);
+    struct capture_stream *data = static_cast<struct capture_stream *>(raw_data);
 
     if (id == data->sink_id) {
         const char *serial = spa_dict_lookup(props, PW_KEY_OBJECT_SERIAL);
         if (!serial) {
             data->sink_serial = 0;
-            printf("No serial found on capture sink\n");
         } else {
             data->sink_serial = strtoul(serial, NULL, 10);
         }
@@ -173,91 +195,36 @@ static void registry_event_global(void *raw_data, uint32_t id,
             !(dir = spa_dict_lookup(props, PW_KEY_PORT_DIRECTION)) ||
             !(chn = spa_dict_lookup(props, PW_KEY_AUDIO_CHANNEL))
         ) {
-            printf("One or more props not set\n");
 	    	return;
 	    }
 
         uint32_t node_id = strtoul(nid, NULL, 10);
-        printf("Port: node id %u\n", node_id);
         if (strcmp(dir, "in") == 0 && node_id == data->sink_id && data->sink_id != SPA_ID_INVALID) {
-            printf("=======\n");
+            // This port belongs to our own capture sink.
-            printf("Found our own sink's port: %d sink_id %d channel %s\n", id, data->sink_id, chn);
+            printf("[DBG] Own sink %d (%s) found\n", id, chn);
-            printf("=======\n");
-
-
             data->sink_ports.push_back(
                 { id, strdup(chn), }
             );
         } else if (strcmp(dir, "out") == 0) {
-            if (data->sink_id == SPA_ID_INVALID) {
+            if (!capture_stream_is_ready(data)) {
-                printf("Want to process port %d but sink_id is invalid\n", id);
+                // We're not ready to connect streams, so just track it for later.
-                return;
+                printf("[DBG] Capture sink is not ready yet for %d\n", id);
-            }
+                register_target_port(
-            struct target_node *n = NULL;
+                    data,
-            for (auto t : data->nodes) {
+                    id,
-                if (t.id == node_id) {
+                    node_id,
-                    n = &t;
+                    chn
-                    break;
+                );
-                }
-            }
-            if (!n) {
-                printf("Target not found\n");
-                return;
-            }
-            printf("Target found\n");
-
-            uint32_t p = 0;
-            for (auto sink_port : data->sink_ports) {
-                printf("%s = %s\n", sink_port.channel, chn);
-                if (strcmp(sink_port.channel, chn) == 0) {
-                    p = sink_port.id;
-                    break;
-                }
-            }
-            if (!p) {
-                printf("Failed to find port for channel %s of port %d\n", chn, id);
                 return;
             }
 
-            struct pw_properties *link_props = pw_properties_new(
+            connect_port_to_sink(
-                PW_KEY_OBJECT_LINGER, "false",
+                data,
-                NULL
+                node_id,
+                id,
+                chn
             );
-            pw_properties_setf(link_props, PW_KEY_LINK_OUTPUT_NODE, "%u", node_id);
-            pw_properties_setf(link_props, PW_KEY_LINK_OUTPUT_PORT, "%u", id);
-
-            pw_properties_setf(link_props, PW_KEY_LINK_INPUT_NODE, "%u", data->sink_id);
-            pw_properties_setf(link_props, PW_KEY_LINK_INPUT_PORT, "%u", p);
-
-            printf(
-                "Connecting (%d, %d) -> (%d, %d)\n",
-                node_id, id,
-                data->sink_id, p
-            );
-
-            struct pw_proxy *link_proxy = static_cast<struct pw_proxy *>(
-                pw_core_create_object(
-                    data->core, "link-factory",
-                    PW_TYPE_INTERFACE_Link, PW_VERSION_LINK, &link_props->dict, 0
-                )
-            );
-            data->seq = pw_core_sync(data->core, PW_ID_CORE, data->seq);
-            pw_properties_free(link_props);
-
-            if (!link_proxy) {
-                printf("!!!!! Failed to connect port %u of node %u to capture sink\n", id, node_id);
-                return;
-            }
-            printf("Connected!\n");
         }
-    } else if (strcmp(type, PW_TYPE_INTERFACE_Client) == 0) {
-        const char *client_app_name = spa_dict_lookup(props, PW_KEY_APP_NAME);
-        printf("Client: app name %s id %d\n", client_app_name, id);
-        register_target_client(
-            data,
-            id,
-            client_app_name
-        );
     } else if (strcmp(type, PW_TYPE_INTERFACE_Node) == 0) {
 		const char *node_name, *media_class;
 		if (!(node_name = spa_dict_lookup(props, PW_KEY_NODE_NAME)) ||
@@ -265,26 +232,47 @@ static void registry_event_global(void *raw_data, uint32_t id,
 			return;
 		}
 
-        printf("Node: media_class %s node_app %s id %d\n", media_class, node_name, id);
         if (strcmp(media_class, "Stream/Output/Audio") == 0) {
 			const char *node_app_name = spa_dict_lookup(props, PW_KEY_APP_NAME);
 			if (!node_app_name) {
 				node_app_name = node_name;
 			}
 
+            if (data->config.name.has_value()) {
+                bool matches = strcmp(
+                    node_app_name,
+                    data->config.name.value().c_str()
+                ) == 0;
+
+                printf("[DBG] Node: name %s matches %d exclude %d\n", node_app_name, matches, data->config.exclude);
+                if (!(matches ^ data->config.exclude)) {
+                    return;
+                }
+            }
+
 			uint32_t client_id = 0;
 			const char *client_id_str = spa_dict_lookup(props, PW_KEY_CLIENT_ID);
 			if (client_id_str) {
 				client_id = strtoul(client_id_str, NULL, 10);
 			}
             
-            
             register_target_node(
                 data,
                 id,
                 client_id,
                 node_app_name
             );
+        } else if (strcmp(media_class, "Audio/Source") == 0) {
+            if (!data->config.device || data->config.exclude || !data->config.name.has_value()) {
+                return;
+            }
+
+            if (strcmp(node_name, data->config.name.value().c_str()) != 0) {
+                return;
+            }
+
+            printf("[DBG] Tracking input source %d (%s)\n", id, node_name);
+            register_target_input(data, id);
         }
     }
 }
@@ -301,9 +289,9 @@ static const struct pw_registry_events registry_events = {
 };
 
 static void on_sink_proxy_bound(void *userdata, uint32_t global_id) {
-    struct data *data = static_cast<struct data*>(userdata);
+    struct capture_stream *data = static_cast<struct capture_stream *>(userdata);
     data->sink_id = global_id;
-    printf("Got id %d\n", global_id);
+    printf("[DBG] Got proxy sink id %d\n", global_id);
 }
 
 static void on_sink_proxy_error(void *data, int seq, int res, const char *message)
@@ -318,15 +306,61 @@ static const struct pw_proxy_events sink_proxy_events = {
 };
 
 void init_pipewire() {
-    struct data data = {
+    pw_init(NULL, NULL);
+}
+
+bool capture_stream_is_ready(struct capture_stream *data) {
+    return data->sink_id != SPA_ID_INVALID &&
+        data->sink_serial != SPA_ID_INVALID &&
+        data->sink_ports.size() == data->config.channels;
+}
+
+enum spa_audio_channel *channel_num_to_channels(int num) {
+    enum spa_audio_channel channels[8];
+    if (num == 1) {
+        // Mono
+        channels[0] = SPA_AUDIO_CHANNEL_MONO;
+        channels[1] = SPA_AUDIO_CHANNEL_UNKNOWN;
+        channels[2] = SPA_AUDIO_CHANNEL_UNKNOWN;
+        channels[3] = SPA_AUDIO_CHANNEL_UNKNOWN;
+        channels[4] = SPA_AUDIO_CHANNEL_UNKNOWN;
+        channels[5] = SPA_AUDIO_CHANNEL_UNKNOWN;
+        channels[6] = SPA_AUDIO_CHANNEL_UNKNOWN;
+        channels[7] = SPA_AUDIO_CHANNEL_UNKNOWN;
+    } else if (num == 2) {
+        // Probably FL,FR.
+        channels[0] = SPA_AUDIO_CHANNEL_FL;
+        channels[1] = SPA_AUDIO_CHANNEL_FL;
+        channels[2] = SPA_AUDIO_CHANNEL_UNKNOWN;
+        channels[3] = SPA_AUDIO_CHANNEL_UNKNOWN;
+        channels[4] = SPA_AUDIO_CHANNEL_UNKNOWN;
+        channels[5] = SPA_AUDIO_CHANNEL_UNKNOWN;
+        channels[6] = SPA_AUDIO_CHANNEL_UNKNOWN;
+        channels[7] = SPA_AUDIO_CHANNEL_UNKNOWN;
+    }
+
+    return channels;
+}
+
+const char *channel_num_to_position(int num) {
+    if (num == 1) {
+        return "MONO";
+    } else if (num == 2) {
+        return "FL,FR";
+    }
+}
+
+struct capture_stream create_capture_stream(struct capture_config config) {
+    struct capture_stream data = {
         0,
         sink_id: SPA_ID_INVALID,
-        sink_serial: 0,
+        sink_serial: SPA_ID_INVALID,
         seq: 0,
         sink_ports: std::vector<struct sink_port> {},
-        targets: std::vector<struct target_client> {},
         nodes: std::vector<struct target_node> {},
         ports: std::vector<struct target_port> {},
+        inputs: std::vector<struct target_input> {},
+        config: config,
     };
     const struct spa_pod *params[1];
     uint8_t buffer[2048];
@@ -346,6 +380,8 @@ void init_pipewire() {
     pw_core_sync(data.core, PW_ID_CORE, 0);
     //pw_thread_loop_wait(data.thread_loop);
     pw_thread_loop_unlock(data.thread_loop);
+    char numbuf[2];
+    sprintf(numbuf, "%d", data.config.channels);
 
     props = pw_properties_new(
         PW_KEY_MEDIA_TYPE, "Audio",
@@ -353,8 +389,8 @@ void init_pipewire() {
         PW_KEY_MEDIA_ROLE, "Screen",
         PW_KEY_NODE_NAME, "GSR",
         PW_KEY_NODE_VIRTUAL, "true",
-        PW_KEY_AUDIO_CHANNELS, "" STR(AUDIO_CHANNELS) "",
+        PW_KEY_AUDIO_CHANNELS, numbuf,
-        SPA_KEY_AUDIO_POSITION, "FL,FR",
+        SPA_KEY_AUDIO_POSITION, channel_num_to_position(data.config.channels),
         PW_KEY_FACTORY_NAME, "support.null-audio-sink",
         PW_KEY_MEDIA_CLASS, "Audio/Sink/Internal",
         NULL
@@ -379,29 +415,25 @@ void init_pipewire() {
     printf("Listener registered\n");
 
     printf("Waiting for id\n");
-    while (data.sink_id == SPA_ID_INVALID || data.sink_serial == 0) {
+    while (!capture_stream_is_ready(&data)) {
         printf("Poll\n");
         pw_loop_iterate(pw_thread_loop_get_loop(data.thread_loop), -1);
     }
-    printf("Got id\n");
+    printf("Node Setup complete\n");
 
-    enum spa_audio_channel channels[8];
+    // Connect delayed ports to the sink
-    channels[0] = SPA_AUDIO_CHANNEL_FL;
+    //pw_thread_loop_lock(data.thread_loop);
-    channels[1] = SPA_AUDIO_CHANNEL_FL;
+    connect_ports_to_sink(&data);
-    channels[2] = SPA_AUDIO_CHANNEL_UNKNOWN;
+    //pw_thread_loop_unlock(data.thread_loop);
-    channels[3] = SPA_AUDIO_CHANNEL_UNKNOWN;
-    channels[4] = SPA_AUDIO_CHANNEL_UNKNOWN;
-    channels[5] = SPA_AUDIO_CHANNEL_UNKNOWN;
-    channels[6] = SPA_AUDIO_CHANNEL_UNKNOWN;
-    channels[7] = SPA_AUDIO_CHANNEL_UNKNOWN;
 
+    auto channels = channel_num_to_channels(data.config.channels);
     params[0] = spa_pod_builder_add_object(
         &b,
         SPA_TYPE_OBJECT_Format, SPA_PARAM_EnumFormat,
         SPA_FORMAT_mediaType, SPA_POD_Id(SPA_MEDIA_TYPE_audio),
         SPA_FORMAT_mediaSubtype, SPA_POD_Id(SPA_MEDIA_SUBTYPE_raw),
-        SPA_FORMAT_AUDIO_channels, SPA_POD_Int(AUDIO_CHANNELS),
+        SPA_FORMAT_AUDIO_channels, SPA_POD_Int(data.config.channels),
-        SPA_FORMAT_AUDIO_position, SPA_POD_Array(sizeof(enum spa_audio_channel), SPA_TYPE_Id, AUDIO_CHANNELS, channels),
+        SPA_FORMAT_AUDIO_position, SPA_POD_Array(sizeof(enum spa_audio_channel), SPA_TYPE_Id, data.config.channels, channels),
         SPA_FORMAT_AUDIO_format, SPA_POD_CHOICE_ENUM_Id(
             8, SPA_AUDIO_FORMAT_U8, SPA_AUDIO_FORMAT_S16_LE, SPA_AUDIO_FORMAT_S32_LE,
 			SPA_AUDIO_FORMAT_F32_LE, SPA_AUDIO_FORMAT_U8P, SPA_AUDIO_FORMAT_S16P,
@@ -448,10 +480,12 @@ void init_pipewire() {
     while (true) {
         pw_loop_iterate(pw_thread_loop_get_loop(data.thread_loop), -1);
     }
-
+}
-    pw_proxy_destroy((struct pw_proxy *) data.registry);
+
-    pw_proxy_destroy(data.sink_proxy);
+void free_capture_stream(struct capture_stream *data) {
-    pw_stream_destroy(data.stream);
+    pw_proxy_destroy((struct pw_proxy *) data->registry);
-    pw_context_destroy(data.context);
+    pw_proxy_destroy(data->sink_proxy);
-    pw_thread_loop_destroy(data.thread_loop);
+    pw_stream_destroy(data->stream);
+    pw_context_destroy(data->context);
+    pw_thread_loop_destroy(data->thread_loop);
 }