sink-input, source-output: Add support for nodes

src/pulsecore/sink-input.c

@@ -109,11 +109,13 @@ pa_sink_input_new_data* pa_sink_input_new_data_init(pa_sink_input_new_data *data
     data->resample_method = PA_RESAMPLER_INVALID;
     data->proplist = pa_proplist_new();
     data->volume_writable = true;
-
     data->volume_factor_items = pa_hashmap_new_full(pa_idxset_string_hash_func, pa_idxset_string_compare_func, NULL,
                                                     (pa_free_cb_t) volume_factor_entry_free);
     data->volume_factor_sink_items = pa_hashmap_new_full(pa_idxset_string_hash_func, pa_idxset_string_compare_func, NULL,
                                                          (pa_free_cb_t) volume_factor_entry_free);
+    pa_node_new_data_init(&data->node_data);
+    pa_node_new_data_set_type(&data->node_data, PA_NODE_TYPE_SINK_INPUT);
+    pa_node_new_data_set_direction(&data->node_data, PA_DIRECTION_INPUT);
 
     return data;
 }
@@ -231,9 +233,17 @@ bool pa_sink_input_new_data_set_formats(pa_sink_input_new_data *data, pa_idxset
     return true;
 }
 
+void pa_sink_input_new_data_set_create_node(pa_sink_input_new_data *data, bool create) {
+    pa_assert(data);
+
+    data->create_node = create;
+}
+
 void pa_sink_input_new_data_done(pa_sink_input_new_data *data) {
     pa_assert(data);
 
+    pa_node_new_data_done(&data->node_data);
+
     if (data->req_formats)
         pa_idxset_free(data->req_formats, (pa_free_cb_t) pa_format_info_free);
 
@@ -284,7 +294,7 @@ int pa_sink_input_new(
         pa_core *core,
         pa_sink_input_new_data *data) {
 
-    pa_sink_input *i;
+    pa_sink_input *i = NULL;
     pa_resampler *resampler = NULL;
     char st[PA_SAMPLE_SPEC_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX], fmt[PA_FORMAT_INFO_SNPRINT_MAX];
     pa_channel_map original_cm;
@@ -293,6 +303,7 @@ int pa_sink_input_new(
     char *memblockq_name;
     pa_sample_spec ss;
     pa_channel_map map;
+    int ret = 0;
 
     pa_assert(_i);
     pa_assert(core);
@@ -582,6 +593,19 @@ int pa_sink_input_new(
             &i->sink->silence);
     pa_xfree(memblockq_name);
 
+    if (data->create_node) {
+        if (!data->node_data.description)
+            pa_node_new_data_set_description(&data->node_data, pa_sink_input_get_description(i));
+
+        if (!(i->node = pa_node_new(i->core, &data->node_data))) {
+            pa_log("Failed to create a node for sink input \"%s\".", pa_sink_input_get_description(i));
+            ret = -PA_ERR_INTERNAL;
+            goto fail;
+        }
+
+        i->node->owner = i;
+    }
+
     pt = pa_proplist_to_string_sep(i->proplist, "\n    ");
     pa_log_info("Created input %u \"%s\" on %s with sample spec %s and channel map %s\n    %s",
                 i->index,
@@ -596,6 +620,14 @@ int pa_sink_input_new(
 
     *_i = i;
     return 0;
+
+fail:
+    if (i) {
+        pa_sink_input_unlink(i);
+        pa_sink_input_unref(i);
+    }
+
+    return ret;
 }
 
 /* Called from main context */
@@ -679,6 +711,9 @@ void pa_sink_input_unlink(pa_sink_input *i) {
     if (linked)
         pa_hook_fire(&i->core->hooks[PA_CORE_HOOK_SINK_INPUT_UNLINK], i);
 
+    if (i->node)
+        pa_node_unlink(i->node);
+
     if (i->sync_prev)
         i->sync_prev->sync_next = i->sync_next;
     if (i->sync_next)
@@ -753,6 +788,9 @@ static void sink_input_free(pa_object *o) {
      * "half-moved" or are connected to sinks that have no asyncmsgq
      * and are hence half-destructed themselves! */
 
+    if (i->node)
+        pa_node_free(i->node);
+
     if (i->thread_info.render_memblockq)
         pa_memblockq_free(i->thread_info.render_memblockq);
 
@@ -820,6 +858,9 @@ void pa_sink_input_put(pa_sink_input *i) {
 
     pa_assert_se(pa_asyncmsgq_send(i->sink->asyncmsgq, PA_MSGOBJECT(i->sink), PA_SINK_MESSAGE_ADD_INPUT, i, 0, NULL) == 0);
 
+    if (i->node)
+        pa_node_put(i->node);
+
     pa_subscription_post(i->core, PA_SUBSCRIPTION_EVENT_SINK_INPUT|PA_SUBSCRIPTION_EVENT_NEW, i->index);
     pa_hook_fire(&i->core->hooks[PA_CORE_HOOK_SINK_INPUT_PUT], i);
 

src/pulsecore/sink-input.h

@@ -70,6 +70,8 @@ struct pa_sink_input {
     uint32_t index;
     pa_core *core;
 
+    pa_node *node;
+
     /* Please note that this state should only be read with
      * pa_sink_input_get_state(). That function will transparently
      * merge the thread_info.drained value in. */
@@ -314,6 +316,9 @@ typedef struct pa_sink_input_new_data {
     bool volume_writable:1;
 
     bool save_sink:1, save_volume:1, save_muted:1;
+
+    bool create_node;
+    pa_node_new_data node_data;
 } pa_sink_input_new_data;
 
 pa_sink_input_new_data* pa_sink_input_new_data_init(pa_sink_input_new_data *data);
@@ -326,6 +331,7 @@ void pa_sink_input_new_data_add_volume_factor_sink(pa_sink_input_new_data *data,
 void pa_sink_input_new_data_set_muted(pa_sink_input_new_data *data, bool mute);
 bool pa_sink_input_new_data_set_sink(pa_sink_input_new_data *data, pa_sink *s, bool save);
 bool pa_sink_input_new_data_set_formats(pa_sink_input_new_data *data, pa_idxset *formats);
+void pa_sink_input_new_data_set_create_node(pa_sink_input_new_data *data, bool create);
 void pa_sink_input_new_data_done(pa_sink_input_new_data *data);
 
 /* To be called by the implementing module only */

src/pulsecore/source-output.c

@@ -55,6 +55,9 @@ pa_source_output_new_data* pa_source_output_new_data_init(pa_source_output_new_d
     data->resample_method = PA_RESAMPLER_INVALID;
     data->proplist = pa_proplist_new();
     data->volume_writable = true;
+    pa_node_new_data_init(&data->node_data);
+    pa_node_new_data_set_type(&data->node_data, PA_NODE_TYPE_SOURCE_OUTPUT);
+    pa_node_new_data_set_direction(&data->node_data, PA_DIRECTION_OUTPUT);
 
     return data;
 }
@@ -174,9 +177,17 @@ bool pa_source_output_new_data_set_formats(pa_source_output_new_data *data, pa_i
     return true;
 }
 
+void pa_source_output_new_data_set_create_node(pa_source_output_new_data *data, bool create) {
+    pa_assert(data);
+
+    data->create_node = create;
+}
+
 void pa_source_output_new_data_done(pa_source_output_new_data *data) {
     pa_assert(data);
 
+    pa_node_new_data_done(&data->node_data);
+
     if (data->req_formats)
         pa_idxset_free(data->req_formats, (pa_free_cb_t) pa_format_info_free);
 
@@ -219,7 +230,7 @@ int pa_source_output_new(
         pa_core *core,
         pa_source_output_new_data *data) {
 
-    pa_source_output *o;
+    pa_source_output *o = NULL;
     pa_resampler *resampler = NULL;
     char st[PA_SAMPLE_SPEC_SNPRINT_MAX], cm[PA_CHANNEL_MAP_SNPRINT_MAX];
     pa_channel_map original_cm;
@@ -227,6 +238,7 @@ int pa_source_output_new(
     char *pt;
     pa_sample_spec ss;
     pa_channel_map map;
+    int ret = 0;
 
     pa_assert(_o);
     pa_assert(core);
@@ -497,6 +509,19 @@ int pa_source_output_new(
     if (o->direct_on_input)
         pa_assert_se(pa_idxset_put(o->direct_on_input->direct_outputs, o, NULL) == 0);
 
+    if (data->create_node) {
+        if (!data->node_data.description)
+            pa_node_new_data_set_description(&data->node_data, pa_source_output_get_description(o));
+
+        if (!(o->node = pa_node_new(o->core, &data->node_data))) {
+            pa_log("Failed to create a node for source output \"%s\".", pa_source_output_get_description(o));
+            ret = -PA_ERR_INTERNAL;
+            goto fail;
+        }
+
+        o->node->owner = o;
+    }
+
     pt = pa_proplist_to_string_sep(o->proplist, "\n    ");
     pa_log_info("Created output %u \"%s\" on %s with sample spec %s and channel map %s\n    %s",
                 o->index,
@@ -511,6 +536,14 @@ int pa_source_output_new(
 
     *_o = o;
     return 0;
+
+fail:
+    if (o) {
+        pa_source_output_unlink(o);
+        pa_source_output_unref(o);
+    }
+
+    return ret;
 }
 
 /* Called from main context */
@@ -574,6 +607,9 @@ void pa_source_output_unlink(pa_source_output*o) {
     if (linked)
         pa_hook_fire(&o->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_UNLINK], o);
 
+    if (o->node)
+        pa_node_unlink(o->node);
+
     if (o->direct_on_input)
         pa_idxset_remove_by_data(o->direct_on_input->direct_outputs, o, NULL);
 
@@ -633,6 +669,9 @@ static void source_output_free(pa_object* mo) {
 
     pa_log_info("Freeing output %u \"%s\"", o->index, pa_strnull(pa_proplist_gets(o->proplist, PA_PROP_MEDIA_NAME)));
 
+    if (o->node)
+        pa_node_free(o->node);
+
     if (o->thread_info.delay_memblockq)
         pa_memblockq_free(o->thread_info.delay_memblockq);
 
@@ -687,6 +726,9 @@ void pa_source_output_put(pa_source_output *o) {
 
     pa_assert_se(pa_asyncmsgq_send(o->source->asyncmsgq, PA_MSGOBJECT(o->source), PA_SOURCE_MESSAGE_ADD_OUTPUT, o, 0, NULL) == 0);
 
+    if (o->node)
+        pa_node_put(o->node);
+
     pa_subscription_post(o->core, PA_SUBSCRIPTION_EVENT_SOURCE_OUTPUT|PA_SUBSCRIPTION_EVENT_NEW, o->index);
     pa_hook_fire(&o->core->hooks[PA_CORE_HOOK_SOURCE_OUTPUT_PUT], o);
 

src/pulsecore/source-output.h

@@ -68,6 +68,8 @@ struct pa_source_output {
     uint32_t index;
     pa_core *core;
 
+    pa_node *node;
+
     pa_source_output_state_t state;
     pa_source_output_flags_t flags;
 
@@ -270,6 +272,9 @@ typedef struct pa_source_output_new_data {
     bool volume_writable:1;
 
     bool save_source:1, save_volume:1, save_muted:1;
+
+    bool create_node;
+    pa_node_new_data node_data;
 } pa_source_output_new_data;
 
 pa_source_output_new_data* pa_source_output_new_data_init(pa_source_output_new_data *data);
@@ -282,6 +287,7 @@ void pa_source_output_new_data_apply_volume_factor_source(pa_source_output_new_d
 void pa_source_output_new_data_set_muted(pa_source_output_new_data *data, bool mute);
 bool pa_source_output_new_data_set_source(pa_source_output_new_data *data, pa_source *s, bool save);
 bool pa_source_output_new_data_set_formats(pa_source_output_new_data *data, pa_idxset *formats);
+void pa_source_output_new_data_set_create_node(pa_source_output_new_data *data, bool create);
 void pa_source_output_new_data_done(pa_source_output_new_data *data);
 
 /* To be called by the implementing module only */