lib/graph/graph.c

   1 /*
   2  * Copyright 2017-2018 Philippe Proulx <pproulx@efficios.com>
   3  * Copyright 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining a copy
   6  * of this software and associated documentation files (the "Software"), to deal
   7  * in the Software without restriction, including without limitation the rights
   8  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   9  * copies of the Software, and to permit persons to whom the Software is
  10  * furnished to do so, subject to the following conditions:
  11  *
  12  * The above copyright notice and this permission notice shall be included in
  13  * all copies or substantial portions of the Software.
  14  *
  15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  21  * SOFTWARE.
  22  */
  23
  24 #define BT_LOG_TAG "GRAPH"
  25 #include <babeltrace/lib-logging-internal.h>
  26
  27 #include <babeltrace/assert-internal.h>
  28 #include <babeltrace/assert-pre-internal.h>
  29 #include <babeltrace/graph/component-internal.h>
  30 #include <babeltrace/graph/graph.h>
  31 #include <babeltrace/graph/graph-const.h>
  32 #include <babeltrace/graph/graph-internal.h>
  33 #include <babeltrace/graph/connection-internal.h>
  34 #include <babeltrace/graph/component-sink-internal.h>
  35 #include <babeltrace/graph/component-source-const.h>
  36 #include <babeltrace/graph/component-filter-const.h>
  37 #include <babeltrace/graph/port-const.h>
  38 #include <babeltrace/graph/message-internal.h>
  39 #include <babeltrace/graph/message-event-internal.h>
  40 #include <babeltrace/graph/message-packet-internal.h>
  41 #include <babeltrace/compiler-internal.h>
  42 #include <babeltrace/common-internal.h>
  43 #include <babeltrace/types.h>
  44 #include <babeltrace/value.h>
  45 #include <babeltrace/value-const.h>
  46 #include <babeltrace/value-internal.h>
  47 #include <unistd.h>
  48 #include <glib.h>
  49
  50 typedef void (*port_added_func_t)(const void *, const void *, void *);
  51
  52 typedef void (*port_removed_func_t)(const void *, const void *, void *);
  53
  54 typedef void (*ports_connected_func_t)(const void *, const void *, const void *,
  55                 const void *, void *);
  56
  57 typedef void (*ports_disconnected_func_t)(const void *, const void *,
  58                 const void *, const void *, void *);
  59
  60 typedef enum bt_self_component_status (*comp_init_method_t)(const void *,
  61                 const void *, void *);
  62
  63 struct bt_graph_listener {
  64         bt_graph_listener_removed_func removed;
  65         void *data;
  66 };
  67
  68 struct bt_graph_listener_port_added {
  69         struct bt_graph_listener base;
  70         port_added_func_t func;
  71 };
  72
  73 struct bt_graph_listener_port_removed {
  74         struct bt_graph_listener base;
  75         port_removed_func_t func;
  76 };
  77
  78 struct bt_graph_listener_ports_connected {
  79         struct bt_graph_listener base;
  80         ports_connected_func_t func;
  81 };
  82
  83 struct bt_graph_listener_ports_disconnected {
  84         struct bt_graph_listener base;
  85         ports_disconnected_func_t func;
  86 };
  87
  88 #define INIT_LISTENERS_ARRAY(_type, _listeners)                         \
  89         do {                                                            \
  90                 _listeners = g_array_new(FALSE, TRUE, sizeof(_type));   \
  91                 if (!(_listeners)) {                                    \
  92                         BT_LOGE_STR("Failed to allocate one GArray.");  \
  93                 }                                                       \
  94         } while (0)
  95
  96 #define CALL_REMOVE_LISTENERS(_type, _listeners)                        \
  97         do {                                                            \
  98                 size_t i;                                               \
  99                                                                         \
 100                 for (i = 0; i < (_listeners)->len; i++) {               \
 101                         _type *listener =                               \
 102                                 &g_array_index((_listeners), _type, i); \
 103                                                                         \
 104                         if (listener->base.removed) {                   \
 105                                 listener->base.removed(listener->base.data); \
 106                         }                                               \
 107                 }                                                       \
 108         } while (0)
 109
 110 static
 111 void destroy_graph(struct bt_object *obj)
 112 {
 113         struct bt_graph *graph = container_of(obj, struct bt_graph, base);
 114
 115         /*
 116          * The graph's reference count is 0 if we're here. Increment
 117          * it to avoid a double-destroy (possibly infinitely recursive)
 118          * in this situation:
 119          *
 120          * 1. We put and destroy a connection.
 121          * 2. This connection's destructor finalizes its active
 122          *    message iterators.
 123          * 3. A message iterator's finalization function gets a
 124          *    new reference on its component (reference count goes from
 125          *    0 to 1).
 126          * 4. Since this component's reference count goes to 1, it takes
 127          *    a reference on its parent (this graph). This graph's
 128          *    reference count goes from 0 to 1.
 129          * 5. The message iterator's finalization function puts its
 130          *    component reference (reference count goes from 1 to 0).
 131          * 6. Since this component's reference count goes from 1 to 0,
 132          *    it puts its parent (this graph). This graph's reference
 133          *    count goes from 1 to 0.
 134          * 7. Since this graph's reference count goes from 1 to 0,
 135          *    its destructor is called (this function).
 136          *
 137          * With the incrementation below, the graph's reference count at
 138          * step 4 goes from 1 to 2, and from 2 to 1 at step 6. This
 139          * ensures that this function is not called two times.
 140          */
 141         BT_LIB_LOGD("Destroying graph: %!+g", graph);
 142         obj->ref_count++;
 143
 144         /*
 145          * Cancel the graph to disallow some operations, like creating
 146          * message iterators and adding ports to components.
 147          */
 148         (void) bt_graph_cancel((void *) graph);
 149
 150         /* Call all remove listeners */
 151         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
 152                 graph->listeners.source_output_port_added);
 153         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
 154                 graph->listeners.filter_output_port_added);
 155         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
 156                 graph->listeners.filter_input_port_added);
 157         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_added,
 158                 graph->listeners.sink_input_port_added);
 159         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_removed,
 160                 graph->listeners.source_output_port_removed);
 161         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_removed,
 162                 graph->listeners.filter_output_port_removed);
 163         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_removed,
 164                 graph->listeners.filter_input_port_removed);
 165         CALL_REMOVE_LISTENERS(struct bt_graph_listener_port_removed,
 166                 graph->listeners.sink_input_port_removed);
 167         CALL_REMOVE_LISTENERS(struct bt_graph_listener_ports_connected,
 168                 graph->listeners.source_filter_ports_connected);
 169         CALL_REMOVE_LISTENERS(struct bt_graph_listener_ports_connected,
 170                 graph->listeners.source_sink_ports_connected);
 171         CALL_REMOVE_LISTENERS(struct bt_graph_listener_ports_connected,
 172                 graph->listeners.filter_sink_ports_connected);
 173         CALL_REMOVE_LISTENERS(struct bt_graph_listener_ports_disconnected,
 174                 graph->listeners.source_filter_ports_disconnected);
 175         CALL_REMOVE_LISTENERS(struct bt_graph_listener_ports_disconnected,
 176                 graph->listeners.source_sink_ports_disconnected);
 177         CALL_REMOVE_LISTENERS(struct bt_graph_listener_ports_disconnected,
 178                 graph->listeners.filter_sink_ports_disconnected);
 179
 180         if (graph->messages) {
 181                 g_ptr_array_free(graph->messages, TRUE);
 182                 graph->messages = NULL;
 183         }
 184
 185         if (graph->connections) {
 186                 BT_LOGD_STR("Destroying connections.");
 187                 g_ptr_array_free(graph->connections, TRUE);
 188                 graph->connections = NULL;
 189         }
 190
 191         if (graph->components) {
 192                 BT_LOGD_STR("Destroying components.");
 193                 g_ptr_array_free(graph->components, TRUE);
 194                 graph->components = NULL;
 195         }
 196
 197         if (graph->sinks_to_consume) {
 198                 g_queue_free(graph->sinks_to_consume);
 199                 graph->sinks_to_consume = NULL;
 200         }
 201
 202         if (graph->listeners.source_output_port_added) {
 203                 g_array_free(graph->listeners.source_output_port_added, TRUE);
 204                 graph->listeners.source_output_port_added = NULL;
 205         }
 206
 207         if (graph->listeners.filter_output_port_added) {
 208                 g_array_free(graph->listeners.filter_output_port_added, TRUE);
 209                 graph->listeners.filter_output_port_added = NULL;
 210         }
 211
 212         if (graph->listeners.filter_input_port_added) {
 213                 g_array_free(graph->listeners.filter_input_port_added, TRUE);
 214                 graph->listeners.filter_input_port_added = NULL;
 215         }
 216
 217         if (graph->listeners.sink_input_port_added) {
 218                 g_array_free(graph->listeners.sink_input_port_added, TRUE);
 219                 graph->listeners.sink_input_port_added = NULL;
 220         }
 221
 222         if (graph->listeners.source_output_port_removed) {
 223                 g_array_free(graph->listeners.source_output_port_removed, TRUE);
 224                 graph->listeners.source_output_port_removed = NULL;
 225         }
 226
 227         if (graph->listeners.filter_output_port_removed) {
 228                 g_array_free(graph->listeners.filter_output_port_removed, TRUE);
 229                 graph->listeners.filter_output_port_removed = NULL;
 230         }
 231
 232         if (graph->listeners.filter_input_port_removed) {
 233                 g_array_free(graph->listeners.filter_input_port_removed, TRUE);
 234                 graph->listeners.filter_input_port_removed = NULL;
 235         }
 236
 237         if (graph->listeners.sink_input_port_removed) {
 238                 g_array_free(graph->listeners.sink_input_port_removed, TRUE);
 239                 graph->listeners.sink_input_port_removed = NULL;
 240         }
 241
 242         if (graph->listeners.source_filter_ports_connected) {
 243                 g_array_free(graph->listeners.source_filter_ports_connected,
 244                         TRUE);
 245                 graph->listeners.source_filter_ports_connected = NULL;
 246         }
 247
 248         if (graph->listeners.source_sink_ports_connected) {
 249                 g_array_free(graph->listeners.source_sink_ports_connected,
 250                         TRUE);
 251                 graph->listeners.source_sink_ports_connected = NULL;
 252         }
 253
 254         if (graph->listeners.filter_sink_ports_connected) {
 255                 g_array_free(graph->listeners.filter_sink_ports_connected,
 256                         TRUE);
 257                 graph->listeners.filter_sink_ports_connected = NULL;
 258         }
 259
 260         if (graph->listeners.source_filter_ports_disconnected) {
 261                 g_array_free(graph->listeners.source_filter_ports_disconnected,
 262                         TRUE);
 263                 graph->listeners.source_filter_ports_disconnected = NULL;
 264         }
 265
 266         if (graph->listeners.source_sink_ports_disconnected) {
 267                 g_array_free(graph->listeners.source_sink_ports_disconnected,
 268                         TRUE);
 269                 graph->listeners.source_sink_ports_disconnected = NULL;
 270         }
 271
 272         if (graph->listeners.filter_sink_ports_disconnected) {
 273                 g_array_free(graph->listeners.filter_sink_ports_disconnected,
 274                         TRUE);
 275                 graph->listeners.filter_sink_ports_disconnected = NULL;
 276         }
 277
 278         bt_object_pool_finalize(&graph->event_msg_pool);
 279         bt_object_pool_finalize(&graph->packet_begin_msg_pool);
 280         bt_object_pool_finalize(&graph->packet_end_msg_pool);
 281         g_free(graph);
 282 }
 283
 284 static
 285 void destroy_message_event(struct bt_message *msg,
 286                 struct bt_graph *graph)
 287 {
 288         bt_message_event_destroy(msg);
 289 }
 290
 291 static
 292 void destroy_message_packet_begin(struct bt_message *msg,
 293                 struct bt_graph *graph)
 294 {
 295         bt_message_packet_beginning_destroy(msg);
 296 }
 297
 298 static
 299 void destroy_message_packet_end(struct bt_message *msg,
 300                 struct bt_graph *graph)
 301 {
 302         bt_message_packet_end_destroy(msg);
 303 }
 304
 305 static
 306 void notify_message_graph_is_destroyed(struct bt_message *msg)
 307 {
 308         bt_message_unlink_graph(msg);
 309 }
 310
 311 struct bt_graph *bt_graph_create(void)
 312 {
 313         struct bt_graph *graph;
 314         int ret;
 315
 316         BT_LOGD_STR("Creating graph object.");
 317         graph = g_new0(struct bt_graph, 1);
 318         if (!graph) {
 319                 BT_LOGE_STR("Failed to allocate one graph.");
 320                 goto end;
 321         }
 322
 323         bt_object_init_shared(&graph->base, destroy_graph);
 324         graph->connections = g_ptr_array_new_with_free_func(
 325                 (GDestroyNotify) bt_object_try_spec_release);
 326         if (!graph->connections) {
 327                 BT_LOGE_STR("Failed to allocate one GPtrArray.");
 328                 goto error;
 329         }
 330         graph->components = g_ptr_array_new_with_free_func(
 331                 (GDestroyNotify) bt_object_try_spec_release);
 332         if (!graph->components) {
 333                 BT_LOGE_STR("Failed to allocate one GPtrArray.");
 334                 goto error;
 335         }
 336         graph->sinks_to_consume = g_queue_new();
 337         if (!graph->sinks_to_consume) {
 338                 BT_LOGE_STR("Failed to allocate one GQueue.");
 339                 goto error;
 340         }
 341
 342         bt_graph_set_can_consume(graph, true);
 343         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
 344                 graph->listeners.source_output_port_added);
 345
 346         if (!graph->listeners.source_output_port_added) {
 347                 ret = -1;
 348                 goto error;
 349         }
 350
 351         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
 352                 graph->listeners.filter_output_port_added);
 353
 354         if (!graph->listeners.filter_output_port_added) {
 355                 ret = -1;
 356                 goto error;
 357         }
 358
 359         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
 360                 graph->listeners.filter_input_port_added);
 361
 362         if (!graph->listeners.filter_input_port_added) {
 363                 ret = -1;
 364                 goto error;
 365         }
 366
 367         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_added,
 368                 graph->listeners.sink_input_port_added);
 369
 370         if (!graph->listeners.sink_input_port_added) {
 371                 ret = -1;
 372                 goto error;
 373         }
 374
 375         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_removed,
 376                 graph->listeners.source_output_port_removed);
 377
 378         if (!graph->listeners.source_output_port_removed) {
 379                 ret = -1;
 380                 goto error;
 381         }
 382
 383         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_removed,
 384                 graph->listeners.filter_output_port_removed);
 385
 386         if (!graph->listeners.filter_output_port_removed) {
 387                 ret = -1;
 388                 goto error;
 389         }
 390
 391         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_removed,
 392                 graph->listeners.filter_input_port_removed);
 393
 394         if (!graph->listeners.filter_input_port_removed) {
 395                 ret = -1;
 396                 goto error;
 397         }
 398
 399         INIT_LISTENERS_ARRAY(struct bt_graph_listener_port_removed,
 400                 graph->listeners.sink_input_port_removed);
 401
 402         if (!graph->listeners.sink_input_port_removed) {
 403                 ret = -1;
 404                 goto error;
 405         }
 406
 407         INIT_LISTENERS_ARRAY(struct bt_graph_listener_ports_connected,
 408                 graph->listeners.source_filter_ports_connected);
 409
 410         if (!graph->listeners.source_filter_ports_connected) {
 411                 ret = -1;
 412                 goto error;
 413         }
 414
 415         INIT_LISTENERS_ARRAY(struct bt_graph_listener_ports_connected,
 416                 graph->listeners.source_sink_ports_connected);
 417
 418         if (!graph->listeners.source_sink_ports_connected) {
 419                 ret = -1;
 420                 goto error;
 421         }
 422
 423         INIT_LISTENERS_ARRAY(struct bt_graph_listener_ports_connected,
 424                 graph->listeners.filter_sink_ports_connected);
 425
 426         if (!graph->listeners.filter_sink_ports_connected) {
 427                 ret = -1;
 428                 goto error;
 429         }
 430
 431         INIT_LISTENERS_ARRAY(struct bt_graph_listener_ports_disconnected,
 432                 graph->listeners.source_filter_ports_disconnected);
 433
 434         if (!graph->listeners.source_filter_ports_disconnected) {
 435                 ret = -1;
 436                 goto error;
 437         }
 438
 439         INIT_LISTENERS_ARRAY(struct bt_graph_listener_ports_disconnected,
 440                 graph->listeners.source_sink_ports_disconnected);
 441
 442         if (!graph->listeners.source_sink_ports_disconnected) {
 443                 ret = -1;
 444                 goto error;
 445         }
 446
 447         INIT_LISTENERS_ARRAY(struct bt_graph_listener_ports_disconnected,
 448                 graph->listeners.filter_sink_ports_disconnected);
 449
 450         if (!graph->listeners.filter_sink_ports_disconnected) {
 451                 ret = -1;
 452                 goto error;
 453         }
 454
 455         ret = bt_object_pool_initialize(&graph->event_msg_pool,
 456                 (bt_object_pool_new_object_func) bt_message_event_new,
 457                 (bt_object_pool_destroy_object_func) destroy_message_event,
 458                 graph);
 459         if (ret) {
 460                 BT_LOGE("Failed to initialize event message pool: ret=%d",
 461                         ret);
 462                 goto error;
 463         }
 464
 465         ret = bt_object_pool_initialize(&graph->packet_begin_msg_pool,
 466                 (bt_object_pool_new_object_func) bt_message_packet_beginning_new,
 467                 (bt_object_pool_destroy_object_func) destroy_message_packet_begin,
 468                 graph);
 469         if (ret) {
 470                 BT_LOGE("Failed to initialize packet beginning message pool: ret=%d",
 471                         ret);
 472                 goto error;
 473         }
 474
 475         ret = bt_object_pool_initialize(&graph->packet_end_msg_pool,
 476                 (bt_object_pool_new_object_func) bt_message_packet_end_new,
 477                 (bt_object_pool_destroy_object_func) destroy_message_packet_end,
 478                 graph);
 479         if (ret) {
 480                 BT_LOGE("Failed to initialize packet end message pool: ret=%d",
 481                         ret);
 482                 goto error;
 483         }
 484
 485         graph->messages = g_ptr_array_new_with_free_func(
 486                 (GDestroyNotify) notify_message_graph_is_destroyed);
 487         BT_LIB_LOGD("Created graph object: %!+g", graph);
 488
 489 end:
 490         return (void *) graph;
 491
 492 error:
 493         BT_OBJECT_PUT_REF_AND_RESET(graph);
 494         goto end;
 495 }
 496
 497 enum bt_graph_status bt_graph_connect_ports(
 498                 struct bt_graph *graph,
 499                 const struct bt_port_output *upstream_port_out,
 500                 const struct bt_port_input *downstream_port_in,
 501                 const struct bt_connection **user_connection)
 502 {
 503         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 504         struct bt_connection *connection = NULL;
 505         struct bt_port *upstream_port = (void *) upstream_port_out;
 506         struct bt_port *downstream_port = (void *) downstream_port_in;
 507         struct bt_component *upstream_component = NULL;
 508         struct bt_component *downstream_component = NULL;
 509         enum bt_self_component_status component_status;
 510         bool init_can_consume;
 511
 512         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 513         BT_ASSERT_PRE_NON_NULL(upstream_port, "Upstream port");
 514         BT_ASSERT_PRE_NON_NULL(downstream_port, "Downstream port port");
 515         BT_ASSERT_PRE(!graph->canceled, "Graph is canceled: %!+g", graph);
 516         BT_ASSERT_PRE(!bt_port_is_connected(upstream_port),
 517                 "Upstream port is already connected: %!+p", upstream_port);
 518         BT_ASSERT_PRE(!bt_port_is_connected(downstream_port),
 519                 "Downstream port is already connected: %!+p", downstream_port);
 520         BT_ASSERT_PRE(bt_port_borrow_component_inline((void *) upstream_port),
 521                 "Upstream port does not belong to a component: %!+p",
 522                 upstream_port);
 523         BT_ASSERT_PRE(bt_port_borrow_component_inline((void *) downstream_port),
 524                 "Downstream port does not belong to a component: %!+p",
 525                 downstream_port);
 526         init_can_consume = graph->can_consume;
 527         BT_LIB_LOGD("Connecting component ports within graph: "
 528                 "%![graph-]+g, %![up-port-]+p, %![down-port-]+p",
 529                 graph, upstream_port, downstream_port);
 530         bt_graph_set_can_consume(graph, false);
 531         upstream_component = bt_port_borrow_component_inline(
 532                 (void *) upstream_port);
 533         downstream_component = bt_port_borrow_component_inline(
 534                 (void *) downstream_port);
 535
 536         /*
 537          * At this point the ports are not connected yet. Both
 538          * components need to accept an eventual connection to their
 539          * port by the other port before we continue.
 540          */
 541         BT_LIB_LOGD("Asking upstream component to accept the connection: "
 542                 "%![comp-]+c", upstream_component);
 543         component_status = bt_component_accept_port_connection(
 544                 upstream_component, (void *) upstream_port,
 545                 (void *) downstream_port);
 546         if (component_status != BT_SELF_COMPONENT_STATUS_OK) {
 547                 if (component_status == BT_SELF_COMPONENT_STATUS_REFUSE_PORT_CONNECTION) {
 548                         BT_LOGD_STR("Upstream component refused the connection.");
 549                 } else {
 550                         BT_LOGW("Cannot ask upstream component to accept the connection: "
 551                                 "status=%s", bt_self_component_status_string(component_status));
 552                 }
 553
 554                 status = (int) component_status;
 555                 goto end;
 556         }
 557
 558         BT_LIB_LOGD("Asking downstream component to accept the connection: "
 559                 "%![comp-]+c", downstream_component);
 560         component_status = bt_component_accept_port_connection(
 561                 downstream_component, (void *) downstream_port,
 562                 (void *) upstream_port);
 563         if (component_status != BT_SELF_COMPONENT_STATUS_OK) {
 564                 if (component_status == BT_SELF_COMPONENT_STATUS_REFUSE_PORT_CONNECTION) {
 565                         BT_LOGD_STR("Downstream component refused the connection.");
 566                 } else {
 567                         BT_LOGW("Cannot ask downstream component to accept the connection: "
 568                                 "status=%s", bt_self_component_status_string(component_status));
 569                 }
 570
 571                 status = (int) component_status;
 572                 goto end;
 573         }
 574
 575         BT_LOGD_STR("Creating connection.");
 576         connection = bt_connection_create(graph, (void *) upstream_port,
 577                 (void *) downstream_port);
 578         if (!connection) {
 579                 BT_LOGW("Cannot create connection object.");
 580                 status = BT_GRAPH_STATUS_NOMEM;
 581                 goto end;
 582         }
 583
 584         BT_LIB_LOGD("Connection object created: %!+x", connection);
 585
 586         /*
 587          * Ownership of upstream_component/downstream_component and of
 588          * the connection object is transferred to the graph.
 589          */
 590         g_ptr_array_add(graph->connections, connection);
 591
 592         /*
 593          * Notify both components that their port is connected.
 594          */
 595         BT_LIB_LOGD("Notifying upstream component that its port is connected: "
 596                 "%![comp-]+c, %![port-]+p", upstream_component, upstream_port);
 597         component_status = bt_component_port_connected(upstream_component,
 598                 (void *) upstream_port, (void *) downstream_port);
 599         if (component_status != BT_SELF_COMPONENT_STATUS_OK) {
 600                 BT_LIB_LOGW("Error while notifying upstream component that its port is connected: "
 601                         "status=%s, %![graph-]+g, %![up-comp-]+c, "
 602                         "%![down-comp-]+c, %![up-port-]+p, %![down-port-]+p",
 603                         bt_self_component_status_string(component_status),
 604                         graph, upstream_component, downstream_component,
 605                         upstream_port, downstream_port);
 606                 bt_connection_end(connection, true);
 607                 status = (int) component_status;
 608                 goto end;
 609         }
 610
 611         connection->msgied_upstream_port_connected = true;
 612         BT_LIB_LOGD("Notifying downstream component that its port is connected: "
 613                 "%![comp-]+c, %![port-]+p", downstream_component,
 614                 downstream_port);
 615         component_status = bt_component_port_connected(downstream_component,
 616                 (void *) downstream_port, (void *) upstream_port);
 617         if (component_status != BT_SELF_COMPONENT_STATUS_OK) {
 618                 BT_LIB_LOGW("Error while notifying downstream component that its port is connected: "
 619                         "status=%s, %![graph-]+g, %![up-comp-]+c, "
 620                         "%![down-comp-]+c, %![up-port-]+p, %![down-port-]+p",
 621                         bt_self_component_status_string(component_status),
 622                         graph, upstream_component, downstream_component,
 623                         upstream_port, downstream_port);
 624                 bt_connection_end(connection, true);
 625                 status = (int) component_status;
 626                 goto end;
 627         }
 628
 629         connection->msgied_downstream_port_connected = true;
 630
 631         /*
 632          * Notify the graph's creator that both ports are connected.
 633          */
 634         BT_LOGD_STR("Notifying graph's user that new component ports are connected.");
 635         bt_graph_notify_ports_connected(graph, upstream_port, downstream_port);
 636         connection->msgied_graph_ports_connected = true;
 637         BT_LIB_LOGD("Connected component ports within graph: "
 638                 "%![graph-]+g, %![up-comp-]+c, %![down-comp-]+c, "
 639                 "%![up-port-]+p, %![down-port-]+p",
 640                 graph, upstream_component, downstream_component,
 641                 upstream_port, downstream_port);
 642
 643         if (user_connection) {
 644                 /* Move reference to user */
 645                 *user_connection = connection;
 646                 connection = NULL;
 647         }
 648
 649 end:
 650         bt_object_put_ref(connection);
 651         (void) init_can_consume;
 652         bt_graph_set_can_consume(graph, init_can_consume);
 653         return status;
 654 }
 655
 656 static inline
 657 enum bt_graph_status consume_graph_sink(struct bt_component_sink *comp)
 658 {
 659         enum bt_self_component_status comp_status;
 660         struct bt_component_class_sink *sink_class = NULL;
 661
 662         BT_ASSERT(comp);
 663         sink_class = (void *) comp->parent.class;
 664         BT_ASSERT(sink_class->methods.consume);
 665         BT_LIB_LOGD("Calling user's consume method: %!+c", comp);
 666         comp_status = sink_class->methods.consume((void *) comp);
 667         BT_LOGD("User method returned: status=%s",
 668                 bt_self_component_status_string(comp_status));
 669         BT_ASSERT_PRE(comp_status == BT_SELF_COMPONENT_STATUS_OK ||
 670                 comp_status == BT_SELF_COMPONENT_STATUS_END ||
 671                 comp_status == BT_SELF_COMPONENT_STATUS_AGAIN ||
 672                 comp_status == BT_SELF_COMPONENT_STATUS_ERROR ||
 673                 comp_status == BT_SELF_COMPONENT_STATUS_NOMEM,
 674                 "Invalid component status returned by consuming method: "
 675                 "status=%s", bt_self_component_status_string(comp_status));
 676         if (comp_status < 0) {
 677                 BT_LOGW_STR("Consume method failed.");
 678                 goto end;
 679         }
 680
 681         BT_LIB_LOGV("Consumed from sink: %![comp-]+c, status=%s",
 682                 comp, bt_self_component_status_string(comp_status));
 683
 684 end:
 685         return (int) comp_status;
 686 }
 687
 688 /*
 689  * `node` is removed from the queue of sinks to consume when passed to
 690  * this function. This function adds it back to the queue if there's
 691  * still something to consume afterwards.
 692  */
 693 static inline
 694 enum bt_graph_status consume_sink_node(struct bt_graph *graph, GList *node)
 695 {
 696         enum bt_graph_status status;
 697         struct bt_component_sink *sink;
 698
 699         sink = node->data;
 700         status = consume_graph_sink(sink);
 701         if (unlikely(status != BT_GRAPH_STATUS_END)) {
 702                 g_queue_push_tail_link(graph->sinks_to_consume, node);
 703                 goto end;
 704         }
 705
 706         /* End reached, the node is not added back to the queue and free'd. */
 707         g_queue_delete_link(graph->sinks_to_consume, node);
 708
 709         /* Don't forward an END status if there are sinks left to consume. */
 710         if (!g_queue_is_empty(graph->sinks_to_consume)) {
 711                 status = BT_GRAPH_STATUS_OK;
 712                 goto end;
 713         }
 714
 715 end:
 716         BT_LIB_LOGV("Consumed sink node: %![comp-]+c, status=%s",
 717                 sink, bt_graph_status_string(status));
 718         return status;
 719 }
 720
 721 BT_HIDDEN
 722 enum bt_graph_status bt_graph_consume_sink_no_check(struct bt_graph *graph,
 723                 struct bt_component_sink *sink)
 724 {
 725         enum bt_graph_status status;
 726         GList *sink_node;
 727         int index;
 728
 729         BT_LIB_LOGV("Making specific sink consume: %![comp-]+c", sink);
 730         BT_ASSERT(bt_component_borrow_graph((void *) sink) == graph);
 731
 732         if (g_queue_is_empty(graph->sinks_to_consume)) {
 733                 BT_LOGV_STR("Graph's sink queue is empty: end of graph.");
 734                 status = BT_GRAPH_STATUS_END;
 735                 goto end;
 736         }
 737
 738         index = g_queue_index(graph->sinks_to_consume, sink);
 739         if (index < 0) {
 740                 BT_LOGV_STR("Sink is not marked as consumable: sink is ended.");
 741                 status = BT_GRAPH_STATUS_END;
 742                 goto end;
 743         }
 744
 745         sink_node = g_queue_pop_nth_link(graph->sinks_to_consume, index);
 746         BT_ASSERT(sink_node);
 747         status = consume_sink_node(graph, sink_node);
 748
 749 end:
 750         return status;
 751 }
 752
 753 static inline
 754 enum bt_graph_status consume_no_check(struct bt_graph *graph)
 755 {
 756         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 757         struct bt_component *sink;
 758         GList *current_node;
 759
 760         BT_ASSERT_PRE(graph->has_sink,
 761                 "Graph has no sink component: %!+g", graph);
 762         BT_LIB_LOGV("Making next sink consume: %![graph-]+g", graph);
 763
 764         if (unlikely(g_queue_is_empty(graph->sinks_to_consume))) {
 765                 BT_LOGV_STR("Graph's sink queue is empty: end of graph.");
 766                 status = BT_GRAPH_STATUS_END;
 767                 goto end;
 768         }
 769
 770         current_node = g_queue_pop_head_link(graph->sinks_to_consume);
 771         sink = current_node->data;
 772         BT_LIB_LOGV("Chose next sink to consume: %!+c", sink);
 773         status = consume_sink_node(graph, current_node);
 774
 775 end:
 776         return status;
 777 }
 778
 779 enum bt_graph_status bt_graph_consume(
 780                 struct bt_graph *graph)
 781 {
 782         enum bt_graph_status status;
 783
 784         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 785         BT_ASSERT_PRE(!graph->canceled, "Graph is canceled: %!+g", graph);
 786         BT_ASSERT_PRE(graph->can_consume,
 787                 "Cannot consume graph in its current state: %!+g", graph);
 788         bt_graph_set_can_consume(graph, BT_FALSE);
 789         status = consume_no_check(graph);
 790         bt_graph_set_can_consume(graph, BT_TRUE);
 791         return status;
 792 }
 793
 794 enum bt_graph_status bt_graph_run(struct bt_graph *graph)
 795 {
 796         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 797
 798         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 799         BT_ASSERT_PRE(!graph->canceled, "Graph is canceled: %!+g", graph);
 800         BT_ASSERT_PRE(graph->can_consume,
 801                 "Cannot consume graph in its current state: %!+g", graph);
 802         bt_graph_set_can_consume(graph, BT_FALSE);
 803         BT_LIB_LOGV("Running graph: %!+g", graph);
 804
 805         do {
 806                 /*
 807                  * Check if the graph is canceled at each iteration. If
 808                  * the graph was canceled by another thread or by a
 809                  * signal handler, this is not a warning nor an error,
 810                  * it was intentional: log with a DEBUG level only.
 811                  */
 812                 if (unlikely(graph->canceled)) {
 813                         BT_LIB_LOGD("Stopping the graph: graph is canceled: "
 814                                 "%!+g", graph);
 815                         status = BT_GRAPH_STATUS_CANCELED;
 816                         goto end;
 817                 }
 818
 819                 status = consume_no_check(graph);
 820                 if (unlikely(status == BT_GRAPH_STATUS_AGAIN)) {
 821                         /*
 822                          * If AGAIN is received and there are multiple
 823                          * sinks, go ahead and consume from the next
 824                          * sink.
 825                          *
 826                          * However, in the case where a single sink is
 827                          * left, the caller can decide to busy-wait and
 828                          * call bt_graph_run() continuously
 829                          * until the source is ready or it can decide to
 830                          * sleep for an arbitrary amount of time.
 831                          */
 832                         if (graph->sinks_to_consume->length > 1) {
 833                                 status = BT_GRAPH_STATUS_OK;
 834                         }
 835                 } else if (status == BT_GRAPH_STATUS_NO_SINK) {
 836                         goto end;
 837                 }
 838         } while (status == BT_GRAPH_STATUS_OK);
 839
 840         if (g_queue_is_empty(graph->sinks_to_consume)) {
 841                 status = BT_GRAPH_STATUS_END;
 842         }
 843
 844 end:
 845         BT_LIB_LOGV("Graph ran: %![graph-]+g, status=%s", graph,
 846                 bt_graph_status_string(status));
 847         bt_graph_set_can_consume(graph, BT_TRUE);
 848         return status;
 849 }
 850
 851 enum bt_graph_status
 852 bt_graph_add_source_component_output_port_added_listener(
 853                 struct bt_graph *graph,
 854                 bt_graph_source_component_output_port_added_listener_func func,
 855                 bt_graph_listener_removed_func listener_removed, void *data,
 856                 int *out_listener_id)
 857 {
 858         struct bt_graph_listener_port_added listener = {
 859                 .base = {
 860                         .removed = listener_removed,
 861                         .data = data,
 862                 },
 863                 .func = (port_added_func_t) func,
 864         };
 865         int listener_id;
 866
 867         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 868         BT_ASSERT_PRE_NON_NULL(func, "Listener");
 869         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
 870         BT_ASSERT_PRE(!graph->in_remove_listener,
 871                 "Graph currently executing a \"listener removed\" listener: "
 872                 "%!+g", graph);
 873         g_array_append_val(graph->listeners.source_output_port_added, listener);
 874         listener_id = graph->listeners.source_output_port_added->len - 1;
 875         BT_LIB_LOGV("Added \"source component output port added\" listener to graph: "
 876                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
 877                 listener_id);
 878
 879         if (listener_id) {
 880                 *out_listener_id = listener_id;
 881         }
 882
 883         return BT_GRAPH_STATUS_OK;
 884 }
 885
 886 enum bt_graph_status
 887 bt_graph_add_filter_component_output_port_added_listener(
 888                 struct bt_graph *graph,
 889                 bt_graph_filter_component_output_port_added_listener_func func,
 890                 bt_graph_listener_removed_func listener_removed, void *data,
 891                 int *out_listener_id)
 892 {
 893         struct bt_graph_listener_port_added listener = {
 894                 .base = {
 895                         .removed = listener_removed,
 896                         .data = data,
 897                 },
 898                 .func = (port_added_func_t) func,
 899         };
 900         int listener_id;
 901
 902         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 903         BT_ASSERT_PRE_NON_NULL(func, "Listener");
 904         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
 905         BT_ASSERT_PRE(!graph->in_remove_listener,
 906                 "Graph currently executing a \"listener removed\" listener: "
 907                 "%!+g", graph);
 908         g_array_append_val(graph->listeners.filter_output_port_added, listener);
 909         listener_id = graph->listeners.filter_output_port_added->len - 1;
 910         BT_LIB_LOGV("Added \"filter component output port added\" listener to graph: "
 911                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
 912                 listener_id);
 913
 914         if (listener_id) {
 915                 *out_listener_id = listener_id;
 916         }
 917
 918         return BT_GRAPH_STATUS_OK;
 919 }
 920
 921 enum bt_graph_status
 922 bt_graph_add_filter_component_input_port_added_listener(
 923                 struct bt_graph *graph,
 924                 bt_graph_filter_component_input_port_added_listener_func func,
 925                 bt_graph_listener_removed_func listener_removed, void *data,
 926                 int *out_listener_id)
 927 {
 928         struct bt_graph_listener_port_added listener = {
 929                 .base = {
 930                         .removed = listener_removed,
 931                         .data = data,
 932                 },
 933                 .func = (port_added_func_t) func,
 934         };
 935         int listener_id;
 936
 937         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 938         BT_ASSERT_PRE_NON_NULL(func, "Listener");
 939         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
 940         BT_ASSERT_PRE(!graph->in_remove_listener,
 941                 "Graph currently executing a \"listener removed\" listener: "
 942                 "%!+g", graph);
 943         g_array_append_val(graph->listeners.filter_input_port_added, listener);
 944         listener_id = graph->listeners.filter_input_port_added->len - 1;
 945         BT_LIB_LOGV("Added \"filter component input port added\" listener to graph: "
 946                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
 947                 listener_id);
 948
 949         if (listener_id) {
 950                 *out_listener_id = listener_id;
 951         }
 952
 953         return BT_GRAPH_STATUS_OK;
 954 }
 955
 956 enum bt_graph_status
 957 bt_graph_add_sink_component_input_port_added_listener(
 958                 struct bt_graph *graph,
 959                 bt_graph_sink_component_input_port_added_listener_func func,
 960                 bt_graph_listener_removed_func listener_removed, void *data,
 961                 int *out_listener_id)
 962 {
 963         struct bt_graph_listener_port_added listener = {
 964                 .base = {
 965                         .removed = listener_removed,
 966                         .data = data,
 967                 },
 968                 .func = (port_added_func_t) func,
 969         };
 970         int listener_id;
 971
 972         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
 973         BT_ASSERT_PRE_NON_NULL(func, "Listener");
 974         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
 975         BT_ASSERT_PRE(!graph->in_remove_listener,
 976                 "Graph currently executing a \"listener removed\" listener: "
 977                 "%!+g", graph);
 978         g_array_append_val(graph->listeners.sink_input_port_added, listener);
 979         listener_id = graph->listeners.sink_input_port_added->len - 1;
 980         BT_LIB_LOGV("Added \"sink component input port added\" listener to graph: "
 981                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
 982                 listener_id);
 983
 984         if (listener_id) {
 985                 *out_listener_id = listener_id;
 986         }
 987
 988         return BT_GRAPH_STATUS_OK;
 989 }
 990
 991 enum bt_graph_status
 992 bt_graph_add_source_component_output_port_removed_listener(
 993                 struct bt_graph *graph,
 994                 bt_graph_source_component_output_port_removed_listener_func func,
 995                 bt_graph_listener_removed_func listener_removed, void *data,
 996                 int *out_listener_id)
 997 {
 998         struct bt_graph_listener_port_removed listener = {
 999                 .base = {
1000                         .removed = listener_removed,
1001                         .data = data,
1002                 },
1003                 .func = (port_removed_func_t) func,
1004         };
1005         int listener_id;
1006
1007         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1008         BT_ASSERT_PRE_NON_NULL(func, "Listener");
1009         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
1010         BT_ASSERT_PRE(!graph->in_remove_listener,
1011                 "Graph currently executing a \"listener removed\" listener: "
1012                 "%!+g", graph);
1013         g_array_append_val(graph->listeners.source_output_port_removed, listener);
1014         listener_id = graph->listeners.source_output_port_removed->len - 1;
1015         BT_LIB_LOGV("Added \"source component output port removed\" listener to graph: "
1016                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
1017                 listener_id);
1018
1019         if (listener_id) {
1020                 *out_listener_id = listener_id;
1021         }
1022
1023         return BT_GRAPH_STATUS_OK;
1024 }
1025
1026 enum bt_graph_status
1027 bt_graph_add_filter_component_output_port_removed_listener(
1028                 struct bt_graph *graph,
1029                 bt_graph_filter_component_output_port_removed_listener_func func,
1030                 bt_graph_listener_removed_func listener_removed, void *data,
1031                 int *out_listener_id)
1032 {
1033         struct bt_graph_listener_port_removed listener = {
1034                 .base = {
1035                         .removed = listener_removed,
1036                         .data = data,
1037                 },
1038                 .func = (port_removed_func_t) func,
1039         };
1040         int listener_id;
1041
1042         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1043         BT_ASSERT_PRE_NON_NULL(func, "Listener");
1044         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
1045         BT_ASSERT_PRE(!graph->in_remove_listener,
1046                 "Graph currently executing a \"listener removed\" listener: "
1047                 "%!+g", graph);
1048         g_array_append_val(graph->listeners.filter_output_port_removed, listener);
1049         listener_id = graph->listeners.filter_output_port_removed->len - 1;
1050         BT_LIB_LOGV("Added \"filter component output port removed\" listener to graph: "
1051                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
1052                 listener_id);
1053
1054         if (listener_id) {
1055                 *out_listener_id = listener_id;
1056         }
1057
1058         return BT_GRAPH_STATUS_OK;
1059 }
1060
1061 enum bt_graph_status
1062 bt_graph_add_filter_component_input_port_removed_listener(
1063                 struct bt_graph *graph,
1064                 bt_graph_filter_component_input_port_removed_listener_func func,
1065                 bt_graph_listener_removed_func listener_removed, void *data,
1066                 int *out_listener_id)
1067 {
1068         struct bt_graph_listener_port_removed listener = {
1069                 .base = {
1070                         .removed = listener_removed,
1071                         .data = data,
1072                 },
1073                 .func = (port_removed_func_t) func,
1074         };
1075         int listener_id;
1076
1077         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1078         BT_ASSERT_PRE_NON_NULL(func, "Listener");
1079         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
1080         BT_ASSERT_PRE(!graph->in_remove_listener,
1081                 "Graph currently executing a \"listener removed\" listener: "
1082                 "%!+g", graph);
1083         g_array_append_val(graph->listeners.filter_input_port_removed, listener);
1084         listener_id = graph->listeners.filter_input_port_removed->len - 1;
1085         BT_LIB_LOGV("Added \"filter component input port removed\" listener to graph: "
1086                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
1087                 listener_id);
1088
1089         if (listener_id) {
1090                 *out_listener_id = listener_id;
1091         }
1092
1093         return BT_GRAPH_STATUS_OK;
1094 }
1095
1096 enum bt_graph_status
1097 bt_graph_add_sink_component_input_port_removed_listener(
1098                 struct bt_graph *graph,
1099                 bt_graph_sink_component_input_port_removed_listener_func func,
1100                 bt_graph_listener_removed_func listener_removed, void *data,
1101                 int *out_listener_id)
1102 {
1103         struct bt_graph_listener_port_removed listener = {
1104                 .base = {
1105                         .removed = listener_removed,
1106                         .data = data,
1107                 },
1108                 .func = (port_removed_func_t) func,
1109         };
1110         int listener_id;
1111
1112         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1113         BT_ASSERT_PRE_NON_NULL(func, "Listener");
1114         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
1115         BT_ASSERT_PRE(!graph->in_remove_listener,
1116                 "Graph currently executing a \"listener removed\" listener: "
1117                 "%!+g", graph);
1118         g_array_append_val(graph->listeners.sink_input_port_removed, listener);
1119         listener_id = graph->listeners.sink_input_port_removed->len - 1;
1120         BT_LIB_LOGV("Added \"sink component input port removed\" listener to graph: "
1121                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
1122                 listener_id);
1123
1124         if (listener_id) {
1125                 *out_listener_id = listener_id;
1126         }
1127
1128         return BT_GRAPH_STATUS_OK;
1129 }
1130
1131 enum bt_graph_status
1132 bt_graph_add_source_filter_component_ports_connected_listener(
1133                 struct bt_graph *graph,
1134                 bt_graph_source_filter_component_ports_connected_listener_func func,
1135                 bt_graph_listener_removed_func listener_removed, void *data,
1136                 int *out_listener_id)
1137 {
1138         struct bt_graph_listener_ports_connected listener = {
1139                 .base = {
1140                         .removed = listener_removed,
1141                         .data = data,
1142                 },
1143                 .func = (ports_connected_func_t) func,
1144         };
1145         int listener_id;
1146
1147         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1148         BT_ASSERT_PRE_NON_NULL(func, "Listener");
1149         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
1150         BT_ASSERT_PRE(!graph->in_remove_listener,
1151                 "Graph currently executing a \"listener removed\" listener: "
1152                 "%!+g", graph);
1153         g_array_append_val(graph->listeners.source_filter_ports_connected,
1154                 listener);
1155         listener_id = graph->listeners.source_filter_ports_connected->len - 1;
1156         BT_LIB_LOGV("Added \"source to filter component ports connected\" listener to graph: "
1157                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
1158                 listener_id);
1159
1160         if (listener_id) {
1161                 *out_listener_id = listener_id;
1162         }
1163
1164         return BT_GRAPH_STATUS_OK;
1165 }
1166
1167 enum bt_graph_status
1168 bt_graph_add_source_sink_component_ports_connected_listener(
1169                 struct bt_graph *graph,
1170                 bt_graph_source_sink_component_ports_connected_listener_func func,
1171                 bt_graph_listener_removed_func listener_removed, void *data,
1172                 int *out_listener_id)
1173 {
1174         struct bt_graph_listener_ports_connected listener = {
1175                 .base = {
1176                         .removed = listener_removed,
1177                         .data = data,
1178                 },
1179                 .func = (ports_connected_func_t) func,
1180         };
1181         int listener_id;
1182
1183         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1184         BT_ASSERT_PRE_NON_NULL(func, "Listener");
1185         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
1186         BT_ASSERT_PRE(!graph->in_remove_listener,
1187                 "Graph currently executing a \"listener removed\" listener: "
1188                 "%!+g", graph);
1189         g_array_append_val(graph->listeners.source_sink_ports_connected,
1190                 listener);
1191         listener_id = graph->listeners.source_sink_ports_connected->len - 1;
1192         BT_LIB_LOGV("Added \"source to sink component ports connected\" listener to graph: "
1193                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
1194                 listener_id);
1195
1196         if (listener_id) {
1197                 *out_listener_id = listener_id;
1198         }
1199
1200         return BT_GRAPH_STATUS_OK;
1201 }
1202
1203 enum bt_graph_status
1204 bt_graph_add_filter_sink_component_ports_connected_listener(
1205                 struct bt_graph *graph,
1206                 bt_graph_filter_sink_component_ports_connected_listener_func func,
1207                 bt_graph_listener_removed_func listener_removed, void *data,
1208                 int *out_listener_id)
1209 {
1210         struct bt_graph_listener_ports_connected listener = {
1211                 .base = {
1212                         .removed = listener_removed,
1213                         .data = data,
1214                 },
1215                 .func = (ports_connected_func_t) func,
1216         };
1217         int listener_id;
1218
1219         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1220         BT_ASSERT_PRE_NON_NULL(func, "Listener");
1221         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
1222         BT_ASSERT_PRE(!graph->in_remove_listener,
1223                 "Graph currently executing a \"listener removed\" listener: "
1224                 "%!+g", graph);
1225         g_array_append_val(graph->listeners.filter_sink_ports_connected,
1226                 listener);
1227         listener_id = graph->listeners.filter_sink_ports_connected->len - 1;
1228         BT_LIB_LOGV("Added \"filter to sink component ports connected\" listener to graph: "
1229                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
1230                 listener_id);
1231
1232         if (listener_id) {
1233                 *out_listener_id = listener_id;
1234         }
1235
1236         return BT_GRAPH_STATUS_OK;
1237 }
1238
1239 enum bt_graph_status
1240 bt_graph_add_source_filter_component_ports_disconnected_listener(
1241                 struct bt_graph *graph,
1242                 bt_graph_source_filter_component_ports_disconnected_listener_func func,
1243                 bt_graph_listener_removed_func listener_removed, void *data,
1244                 int *out_listener_id)
1245 {
1246         struct bt_graph_listener_ports_disconnected listener = {
1247                 .base = {
1248                         .removed = listener_removed,
1249                         .data = data,
1250                 },
1251                 .func = (ports_disconnected_func_t) func,
1252         };
1253         int listener_id;
1254
1255         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1256         BT_ASSERT_PRE_NON_NULL(func, "Listener");
1257         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
1258         BT_ASSERT_PRE(!graph->in_remove_listener,
1259                 "Graph currently executing a \"listener removed\" listener: "
1260                 "%!+g", graph);
1261         g_array_append_val(graph->listeners.source_filter_ports_disconnected,
1262                 listener);
1263         listener_id = graph->listeners.source_filter_ports_disconnected->len - 1;
1264         BT_LIB_LOGV("Added \"source to filter component ports disconnected\" listener to graph: "
1265                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
1266                 listener_id);
1267
1268         if (listener_id) {
1269                 *out_listener_id = listener_id;
1270         }
1271
1272         return BT_GRAPH_STATUS_OK;
1273 }
1274
1275 enum bt_graph_status
1276 bt_graph_add_source_sink_component_ports_disconnected_listener(
1277                 struct bt_graph *graph,
1278                 bt_graph_source_sink_component_ports_disconnected_listener_func func,
1279                 bt_graph_listener_removed_func listener_removed, void *data,
1280                 int *out_listener_id)
1281 {
1282         struct bt_graph_listener_ports_disconnected listener = {
1283                 .base = {
1284                         .removed = listener_removed,
1285                         .data = data,
1286                 },
1287                 .func = (ports_disconnected_func_t) func,
1288         };
1289         int listener_id;
1290
1291         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1292         BT_ASSERT_PRE_NON_NULL(func, "Listener");
1293         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
1294         BT_ASSERT_PRE(!graph->in_remove_listener,
1295                 "Graph currently executing a \"listener removed\" listener: "
1296                 "%!+g", graph);
1297         g_array_append_val(graph->listeners.source_sink_ports_disconnected,
1298                 listener);
1299         listener_id = graph->listeners.source_sink_ports_disconnected->len - 1;
1300         BT_LIB_LOGV("Added \"source to sink component ports disconnected\" listener to graph: "
1301                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
1302                 listener_id);
1303
1304         if (listener_id) {
1305                 *out_listener_id = listener_id;
1306         }
1307
1308         return BT_GRAPH_STATUS_OK;
1309 }
1310
1311 enum bt_graph_status
1312 bt_graph_add_filter_sink_component_ports_disconnected_listener(
1313                 struct bt_graph *graph,
1314                 bt_graph_filter_sink_component_ports_disconnected_listener_func func,
1315                 bt_graph_listener_removed_func listener_removed, void *data,
1316                 int *out_listener_id)
1317 {
1318         struct bt_graph_listener_ports_disconnected listener = {
1319                 .base = {
1320                         .removed = listener_removed,
1321                         .data = data,
1322                 },
1323                 .func = (ports_disconnected_func_t) func,
1324         };
1325         int listener_id;
1326
1327         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1328         BT_ASSERT_PRE_NON_NULL(func, "Listener");
1329         BT_ASSERT_PRE_NON_NULL(func, "\"Listener removed\" listener");
1330         BT_ASSERT_PRE(!graph->in_remove_listener,
1331                 "Graph currently executing a \"listener removed\" listener: "
1332                 "%!+g", graph);
1333         g_array_append_val(graph->listeners.filter_sink_ports_disconnected,
1334                 listener);
1335         listener_id = graph->listeners.filter_sink_ports_disconnected->len - 1;
1336         BT_LIB_LOGV("Added \"filter to sink component ports disconnected\" listener to graph: "
1337                 "%![graph-]+g, listener-addr=%p, id=%d", graph, listener,
1338                 listener_id);
1339
1340         if (listener_id) {
1341                 *out_listener_id = listener_id;
1342         }
1343
1344         return BT_GRAPH_STATUS_OK;
1345 }
1346
1347 BT_HIDDEN
1348 void bt_graph_notify_port_added(struct bt_graph *graph, struct bt_port *port)
1349 {
1350         uint64_t i;
1351         GArray *listeners;
1352         struct bt_component *comp;
1353
1354         BT_ASSERT(graph);
1355         BT_ASSERT(port);
1356         BT_LIB_LOGV("Notifying graph listeners that a port was added: "
1357                 "%![graph-]+g, %![port-]+p", graph, port);
1358         comp = bt_port_borrow_component_inline(port);
1359         BT_ASSERT(comp);
1360
1361         switch (comp->class->type) {
1362         case BT_COMPONENT_CLASS_TYPE_SOURCE:
1363         {
1364                 switch (port->type) {
1365                 case BT_PORT_TYPE_OUTPUT:
1366                         listeners = graph->listeners.source_output_port_added;
1367                         break;
1368                 default:
1369                         abort();
1370                 }
1371
1372                 break;
1373         }
1374         case BT_COMPONENT_CLASS_TYPE_FILTER:
1375         {
1376                 switch (port->type) {
1377                 case BT_PORT_TYPE_INPUT:
1378                         listeners = graph->listeners.filter_input_port_added;
1379                         break;
1380                 case BT_PORT_TYPE_OUTPUT:
1381                         listeners = graph->listeners.filter_output_port_added;
1382                         break;
1383                 default:
1384                         abort();
1385                 }
1386
1387                 break;
1388         }
1389         case BT_COMPONENT_CLASS_TYPE_SINK:
1390         {
1391                 switch (port->type) {
1392                 case BT_PORT_TYPE_INPUT:
1393                         listeners = graph->listeners.sink_input_port_added;
1394                         break;
1395                 default:
1396                         abort();
1397                 }
1398
1399                 break;
1400         }
1401         default:
1402                 abort();
1403         }
1404
1405         for (i = 0; i < listeners->len; i++) {
1406                 struct bt_graph_listener_port_added *listener =
1407                         &g_array_index(listeners,
1408                                 struct bt_graph_listener_port_added, i);
1409
1410                 BT_ASSERT(listener->func);
1411                 listener->func(comp, port, listener->base.data);
1412         }
1413 }
1414
1415 BT_HIDDEN
1416 void bt_graph_notify_port_removed(struct bt_graph *graph,
1417                 struct bt_component *comp, struct bt_port *port)
1418 {
1419         uint64_t i;
1420         GArray *listeners;
1421
1422         BT_ASSERT(graph);
1423         BT_ASSERT(port);
1424         BT_LIB_LOGV("Notifying graph listeners that a port was removed: "
1425                 "%![graph-]+g, %![comp-]+c, %![port-]+p", graph, comp, port);
1426
1427         switch (comp->class->type) {
1428         case BT_COMPONENT_CLASS_TYPE_SOURCE:
1429         {
1430                 switch (port->type) {
1431                 case BT_PORT_TYPE_OUTPUT:
1432                         listeners = graph->listeners.source_output_port_removed;
1433                         break;
1434                 default:
1435                         abort();
1436                 }
1437
1438                 break;
1439         }
1440         case BT_COMPONENT_CLASS_TYPE_FILTER:
1441         {
1442                 switch (port->type) {
1443                 case BT_PORT_TYPE_INPUT:
1444                         listeners = graph->listeners.filter_input_port_removed;
1445                         break;
1446                 case BT_PORT_TYPE_OUTPUT:
1447                         listeners = graph->listeners.filter_output_port_removed;
1448                         break;
1449                 default:
1450                         abort();
1451                 }
1452
1453                 break;
1454         }
1455         case BT_COMPONENT_CLASS_TYPE_SINK:
1456         {
1457                 switch (port->type) {
1458                 case BT_PORT_TYPE_INPUT:
1459                         listeners = graph->listeners.sink_input_port_removed;
1460                         break;
1461                 default:
1462                         abort();
1463                 }
1464
1465                 break;
1466         }
1467         default:
1468                 abort();
1469         }
1470
1471         for (i = 0; i < listeners->len; i++) {
1472                 struct bt_graph_listener_port_removed *listener =
1473                         &g_array_index(listeners,
1474                                 struct bt_graph_listener_port_removed, i);
1475
1476                 BT_ASSERT(listener->func);
1477                 listener->func(comp, port, listener->base.data);
1478         }
1479 }
1480
1481 BT_HIDDEN
1482 void bt_graph_notify_ports_connected(struct bt_graph *graph,
1483                 struct bt_port *upstream_port, struct bt_port *downstream_port)
1484 {
1485         uint64_t i;
1486         GArray *listeners;
1487         struct bt_component *upstream_comp;
1488         struct bt_component *downstream_comp;
1489
1490         BT_ASSERT(graph);
1491         BT_ASSERT(upstream_port);
1492         BT_ASSERT(downstream_port);
1493         BT_LIB_LOGV("Notifying graph listeners that ports were connected: "
1494                 "%![graph-]+g, %![up-port-]+p, %![down-port-]+p",
1495                 graph, upstream_port, downstream_port);
1496         upstream_comp = bt_port_borrow_component_inline(upstream_port);
1497         BT_ASSERT(upstream_comp);
1498         downstream_comp = bt_port_borrow_component_inline(downstream_port);
1499         BT_ASSERT(downstream_comp);
1500
1501         switch (upstream_comp->class->type) {
1502         case BT_COMPONENT_CLASS_TYPE_SOURCE:
1503         {
1504                 switch (downstream_comp->class->type) {
1505                 case BT_COMPONENT_CLASS_TYPE_FILTER:
1506                         listeners =
1507                                 graph->listeners.source_filter_ports_connected;
1508                         break;
1509                 case BT_COMPONENT_CLASS_TYPE_SINK:
1510                         listeners =
1511                                 graph->listeners.source_sink_ports_connected;
1512                         break;
1513                 default:
1514                         abort();
1515                 }
1516
1517                 break;
1518         }
1519         case BT_COMPONENT_CLASS_TYPE_FILTER:
1520         {
1521                 switch (downstream_comp->class->type) {
1522                 case BT_COMPONENT_CLASS_TYPE_SINK:
1523                         listeners =
1524                                 graph->listeners.filter_sink_ports_connected;
1525                         break;
1526                 default:
1527                         abort();
1528                 }
1529
1530                 break;
1531         }
1532         default:
1533                 abort();
1534         }
1535
1536         for (i = 0; i < listeners->len; i++) {
1537                 struct bt_graph_listener_ports_connected *listener =
1538                         &g_array_index(listeners,
1539                                 struct bt_graph_listener_ports_connected, i);
1540
1541                 BT_ASSERT(listener->func);
1542                 listener->func(upstream_comp, downstream_comp,
1543                         upstream_port, downstream_port, listener->base.data);
1544         }
1545 }
1546
1547 BT_HIDDEN
1548 void bt_graph_notify_ports_disconnected(struct bt_graph *graph,
1549                 struct bt_component *upstream_comp,
1550                 struct bt_component *downstream_comp,
1551                 struct bt_port *upstream_port,
1552                 struct bt_port *downstream_port)
1553 {
1554         uint64_t i;
1555         GArray *listeners;
1556
1557         BT_ASSERT(graph);
1558         BT_ASSERT(upstream_comp);
1559         BT_ASSERT(downstream_comp);
1560         BT_ASSERT(upstream_port);
1561         BT_ASSERT(downstream_port);
1562         BT_LIB_LOGV("Notifying graph listeners that ports were disconnected: "
1563                 "%![graph-]+g, %![up-port-]+p, %![down-port-]+p, "
1564                 "%![up-comp-]+c, %![down-comp-]+c",
1565                 graph, upstream_port, downstream_port, upstream_comp,
1566                 downstream_comp);
1567
1568         switch (upstream_comp->class->type) {
1569         case BT_COMPONENT_CLASS_TYPE_SOURCE:
1570         {
1571                 switch (downstream_comp->class->type) {
1572                 case BT_COMPONENT_CLASS_TYPE_FILTER:
1573                         listeners =
1574                                 graph->listeners.source_filter_ports_disconnected;
1575                         break;
1576                 case BT_COMPONENT_CLASS_TYPE_SINK:
1577                         listeners =
1578                                 graph->listeners.source_sink_ports_disconnected;
1579                         break;
1580                 default:
1581                         abort();
1582                 }
1583
1584                 break;
1585         }
1586         case BT_COMPONENT_CLASS_TYPE_FILTER:
1587         {
1588                 switch (downstream_comp->class->type) {
1589                 case BT_COMPONENT_CLASS_TYPE_SINK:
1590                         listeners =
1591                                 graph->listeners.filter_sink_ports_disconnected;
1592                         break;
1593                 default:
1594                         abort();
1595                 }
1596
1597                 break;
1598         }
1599         default:
1600                 abort();
1601         }
1602
1603         for (i = 0; i < listeners->len; i++) {
1604                 struct bt_graph_listener_ports_disconnected *listener =
1605                         &g_array_index(listeners,
1606                                 struct bt_graph_listener_ports_disconnected, i);
1607
1608                 BT_ASSERT(listener->func);
1609                 listener->func(upstream_comp, downstream_comp,
1610                         upstream_port, downstream_port, listener->base.data);
1611         }
1612 }
1613
1614 enum bt_graph_status bt_graph_cancel(
1615                 struct bt_graph *graph)
1616 {
1617
1618         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1619         graph->canceled = true;
1620         BT_LIB_LOGV("Canceled graph: %!+i", graph);
1621         return BT_GRAPH_STATUS_OK;
1622 }
1623
1624 bt_bool bt_graph_is_canceled(const struct bt_graph *graph)
1625 {
1626         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1627         return graph->canceled ? BT_TRUE : BT_FALSE;
1628 }
1629
1630 BT_HIDDEN
1631 void bt_graph_remove_connection(struct bt_graph *graph,
1632                 struct bt_connection *connection)
1633 {
1634         BT_ASSERT(graph);
1635         BT_ASSERT(connection);
1636         BT_LIB_LOGV("Removing graph's connection: %![graph-]+g, %![conn-]+x",
1637                 graph, connection);
1638         g_ptr_array_remove(graph->connections, connection);
1639 }
1640
1641 BT_ASSERT_PRE_FUNC
1642 static inline
1643 bool component_name_exists(struct bt_graph *graph, const char *name)
1644 {
1645         bool exists = false;
1646         uint64_t i;
1647
1648         for (i = 0; i < graph->components->len; i++) {
1649                 struct bt_component *other_comp = graph->components->pdata[i];
1650
1651                 if (strcmp(name, bt_component_get_name(other_comp)) == 0) {
1652                         BT_ASSERT_PRE_MSG("Another component with the same name already exists in the graph: "
1653                                 "%![other-comp-]+c, name=\"%s\"",
1654                                 other_comp, name);
1655                         exists = true;
1656                         goto end;
1657                 }
1658         }
1659
1660 end:
1661         return exists;
1662 }
1663
1664 static
1665 enum bt_graph_status add_component_with_init_method_data(
1666                 struct bt_graph *graph,
1667                 struct bt_component_class *comp_cls,
1668                 comp_init_method_t init_method,
1669                 const char *name, const struct bt_value *params,
1670                 void *init_method_data, struct bt_component **user_component)
1671 {
1672         enum bt_graph_status graph_status = BT_GRAPH_STATUS_OK;
1673         enum bt_self_component_status comp_status;
1674         struct bt_component *component = NULL;
1675         int ret;
1676         bool init_can_consume;
1677         struct bt_value *new_params = NULL;
1678
1679         BT_ASSERT(comp_cls);
1680         BT_ASSERT_PRE_NON_NULL(graph, "Graph");
1681         BT_ASSERT_PRE_NON_NULL(name, "Name");
1682         BT_ASSERT_PRE(!graph->canceled, "Graph is canceled: %!+g", graph);
1683         BT_ASSERT_PRE(!component_name_exists(graph, name),
1684                 "Duplicate component name: %!+g, name=\"%s\"", graph, name);
1685         BT_ASSERT_PRE(!params || bt_value_is_map(params),
1686                 "Parameter value is not a map value: %!+v", params);
1687         init_can_consume = graph->can_consume;
1688         bt_graph_set_can_consume(graph, false);
1689         BT_LIB_LOGD("Adding component to graph: "
1690                 "%![graph-]+g, %![cc-]+C, name=\"%s\", %![params-]+v, "
1691                 "init-method-data-addr=%p",
1692                 graph, comp_cls, name, params, init_method_data);
1693
1694         if (!params) {
1695                 new_params = bt_value_map_create();
1696                 if (!new_params) {
1697                         BT_LOGE_STR("Cannot create map value object.");
1698                         graph_status = BT_GRAPH_STATUS_NOMEM;
1699                         goto end;
1700                 }
1701
1702                 params = new_params;
1703         }
1704
1705         ret = bt_component_create(comp_cls, name, &component);
1706         if (ret) {
1707                 BT_LOGE("Cannot create empty component object: ret=%d",
1708                         ret);
1709                 graph_status = BT_GRAPH_STATUS_NOMEM;
1710                 goto end;
1711         }
1712
1713         /*
1714          * The user's initialization method needs to see that this
1715          * component is part of the graph. If the user method fails, we
1716          * immediately remove the component from the graph's components.
1717          */
1718         g_ptr_array_add(graph->components, component);
1719         bt_component_set_graph(component, graph);
1720
1721         if (init_method) {
1722                 BT_LOGD_STR("Calling user's initialization method.");
1723                 comp_status = init_method(component, params, init_method_data);
1724                 BT_LOGD("User method returned: status=%s",
1725                         bt_self_component_status_string(comp_status));
1726                 if (comp_status != BT_SELF_COMPONENT_STATUS_OK) {
1727                         BT_LOGW_STR("Initialization method failed.");
1728                         graph_status = (int) comp_status;
1729                         bt_component_set_graph(component, NULL);
1730                         g_ptr_array_remove_fast(graph->components, component);
1731                         goto end;
1732                 }
1733         }
1734
1735         /*
1736          * Mark the component as initialized so that its finalization
1737          * method is called when it is destroyed.
1738          */
1739         component->initialized = true;
1740
1741         /*
1742          * If it's a sink component, it needs to be part of the graph's
1743          * sink queue to be consumed by bt_graph_consume().
1744          */
1745         if (bt_component_is_sink(component)) {
1746                 graph->has_sink = true;
1747                 g_queue_push_tail(graph->sinks_to_consume, component);
1748         }
1749
1750         /*
1751          * Freeze the component class now that it's instantiated at
1752          * least once.
1753          */
1754         BT_LOGD_STR("Freezing component class.");
1755         bt_component_class_freeze(comp_cls);
1756         BT_LIB_LOGD("Added component to graph: "
1757                 "%![graph-]+g, %![cc-]+C, name=\"%s\", %![params-]+v, "
1758                 "init-method-data-addr=%p, %![comp-]+c",
1759                 graph, comp_cls, name, params, init_method_data, component);
1760
1761         if (user_component) {
1762                 /* Move reference to user */
1763                 *user_component = component;
1764                 component = NULL;
1765         }
1766
1767 end:
1768         bt_object_put_ref(component);
1769         bt_object_put_ref(new_params);
1770         (void) init_can_consume;
1771         bt_graph_set_can_consume(graph, init_can_consume);
1772         return graph_status;
1773 }
1774
1775 enum bt_graph_status
1776 bt_graph_add_source_component_with_init_method_data(
1777                 struct bt_graph *graph,
1778                 const struct bt_component_class_source *comp_cls,
1779                 const char *name, const struct bt_value *params,
1780                 void *init_method_data,
1781                 const struct bt_component_source **component)
1782 {
1783         BT_ASSERT_PRE_NON_NULL(comp_cls, "Component class");
1784         return add_component_with_init_method_data(graph,
1785                 (void *) comp_cls, (comp_init_method_t) comp_cls->methods.init,
1786                 name, params, init_method_data, (void *) component);
1787 }
1788
1789 enum bt_graph_status bt_graph_add_source_component(
1790                 struct bt_graph *graph,
1791                 const struct bt_component_class_source *comp_cls,
1792                 const char *name, const struct bt_value *params,
1793                 const struct bt_component_source **component)
1794 {
1795         return bt_graph_add_source_component_with_init_method_data(
1796                 graph, comp_cls, name, params, NULL, component);
1797 }
1798
1799 enum bt_graph_status
1800 bt_graph_add_filter_component_with_init_method_data(
1801                 struct bt_graph *graph,
1802                 const struct bt_component_class_filter *comp_cls,
1803                 const char *name, const struct bt_value *params,
1804                 void *init_method_data,
1805                 const struct bt_component_filter **component)
1806 {
1807         BT_ASSERT_PRE_NON_NULL(comp_cls, "Component class");
1808         return add_component_with_init_method_data(graph,
1809                 (void *) comp_cls, (comp_init_method_t) comp_cls->methods.init,
1810                 name, params, init_method_data, (void *) component);
1811 }
1812
1813 enum bt_graph_status bt_graph_add_filter_component(
1814                 struct bt_graph *graph,
1815                 const struct bt_component_class_filter *comp_cls,
1816                 const char *name, const struct bt_value *params,
1817                 const struct bt_component_filter **component)
1818 {
1819         return bt_graph_add_filter_component_with_init_method_data(
1820                 graph, comp_cls, name, params, NULL, component);
1821 }
1822
1823 enum bt_graph_status
1824 bt_graph_add_sink_component_with_init_method_data(
1825                 struct bt_graph *graph,
1826                 const struct bt_component_class_sink *comp_cls,
1827                 const char *name, const struct bt_value *params,
1828                 void *init_method_data,
1829                 const struct bt_component_sink **component)
1830 {
1831         BT_ASSERT_PRE_NON_NULL(comp_cls, "Component class");
1832         return add_component_with_init_method_data(graph,
1833                 (void *) comp_cls, (comp_init_method_t) comp_cls->methods.init,
1834                 name, params, init_method_data, (void *) component);
1835 }
1836
1837 enum bt_graph_status bt_graph_add_sink_component(
1838                 struct bt_graph *graph,
1839                 const struct bt_component_class_sink *comp_cls,
1840                 const char *name, const struct bt_value *params,
1841                 const struct bt_component_sink **component)
1842 {
1843         return bt_graph_add_sink_component_with_init_method_data(
1844                 graph, comp_cls, name, params, NULL, component);
1845 }
1846
1847 BT_HIDDEN
1848 int bt_graph_remove_unconnected_component(struct bt_graph *graph,
1849                 struct bt_component *component)
1850 {
1851         bool init_can_consume;
1852         uint64_t count;
1853         uint64_t i;
1854         int ret = 0;
1855
1856         BT_ASSERT(graph);
1857         BT_ASSERT(component);
1858         BT_ASSERT(component->base.ref_count == 0);
1859         BT_ASSERT(bt_component_borrow_graph(component) == graph);
1860
1861         init_can_consume = graph->can_consume;
1862         count = bt_component_get_input_port_count(component);
1863
1864         for (i = 0; i < count; i++) {
1865                 struct bt_port *port = (void *)
1866                         bt_component_borrow_input_port_by_index(component, i);
1867
1868                 BT_ASSERT(port);
1869
1870                 if (bt_port_is_connected(port)) {
1871                         BT_LIB_LOGW("Cannot remove component from graph: "
1872                                 "an input port is connected: "
1873                                 "%![graph-]+g, %![comp-]+c, %![port-]+p",
1874                                 graph, component, port);
1875                         goto error;
1876                 }
1877         }
1878
1879         count = bt_component_get_output_port_count(component);
1880
1881         for (i = 0; i < count; i++) {
1882                 struct bt_port *port = (void *)
1883                         bt_component_borrow_output_port_by_index(component, i);
1884
1885                 BT_ASSERT(port);
1886
1887                 if (bt_port_is_connected(port)) {
1888                         BT_LIB_LOGW("Cannot remove component from graph: "
1889                                 "an output port is connected: "
1890                                 "%![graph-]+g, %![comp-]+c, %![port-]+p",
1891                                 graph, component, port);
1892                         goto error;
1893                 }
1894         }
1895
1896         bt_graph_set_can_consume(graph, false);
1897
1898         /* Possibly remove from sinks to consume */
1899         (void) g_queue_remove(graph->sinks_to_consume, component);
1900
1901         if (graph->sinks_to_consume->length == 0) {
1902                 graph->has_sink = false;
1903         }
1904
1905         /*
1906          * This calls bt_object_try_spec_release() on the component, and
1907          * since its reference count is 0, its destructor is called. Its
1908          * destructor calls the user's finalization method (if set).
1909          */
1910         g_ptr_array_remove(graph->components, component);
1911         goto end;
1912
1913 error:
1914         ret = -1;
1915
1916 end:
1917         (void) init_can_consume;
1918         bt_graph_set_can_consume(graph, init_can_consume);
1919         return ret;
1920 }
1921
1922 BT_HIDDEN
1923 void bt_graph_add_message(struct bt_graph *graph,
1924                 struct bt_message *msg)
1925 {
1926         BT_ASSERT(graph);
1927         BT_ASSERT(msg);
1928
1929         /*
1930          * It's okay not to take a reference because, when a
1931          * message's reference count drops to 0, either:
1932          *
1933          * * It is recycled back to one of this graph's pool.
1934          * * It is destroyed because it doesn't have any link to any
1935          *   graph, which means the original graph is already destroyed.
1936          */
1937         g_ptr_array_add(graph->messages, msg);
1938 }
1939
1940 void bt_graph_get_ref(const struct bt_graph *graph)
1941 {
1942         bt_object_get_ref(graph);
1943 }
1944
1945 void bt_graph_put_ref(const struct bt_graph *graph)
1946 {
1947         bt_object_put_ref(graph);
1948 }