lib/component/graph.c

   1 /*
   2  * graph.c
   3  *
   4  * Babeltrace Plugin Component Graph
   5  *
   6  * Copyright 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   7  *
   8  * Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
   9  *
  10  * Permission is hereby granted, free of charge, to any person obtaining a copy
  11  * of this software and associated documentation files (the "Software"), to deal
  12  * in the Software without restriction, including without limitation the rights
  13  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  14  * copies of the Software, and to permit persons to whom the Software is
  15  * furnished to do so, subject to the following conditions:
  16  *
  17  * The above copyright notice and this permission notice shall be included in
  18  * all copies or substantial portions of the Software.
  19  *
  20  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  21  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  22  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  23  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  24  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  25  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  26  * SOFTWARE.
  27  */
  28
  29 #include <babeltrace/graph/component-internal.h>
  30 #include <babeltrace/graph/graph-internal.h>
  31 #include <babeltrace/graph/connection-internal.h>
  32 #include <babeltrace/graph/component-sink-internal.h>
  33 #include <babeltrace/graph/component-source.h>
  34 #include <babeltrace/graph/component-filter.h>
  35 #include <babeltrace/graph/port.h>
  36 #include <babeltrace/compiler.h>
  37 #include <unistd.h>
  38 #include <glib.h>
  39
  40 struct bt_graph_listener {
  41         void *func;
  42         void *data;
  43 };
  44
  45 static
  46 void bt_graph_destroy(struct bt_object *obj)
  47 {
  48         struct bt_graph *graph = container_of(obj,
  49                         struct bt_graph, base);
  50
  51         if (graph->components) {
  52                 g_ptr_array_free(graph->components, TRUE);
  53         }
  54         if (graph->connections) {
  55                 g_ptr_array_free(graph->connections, TRUE);
  56         }
  57         if (graph->sinks_to_consume) {
  58                 g_queue_free(graph->sinks_to_consume);
  59         }
  60
  61         if (graph->listeners.port_added) {
  62                 g_array_free(graph->listeners.port_added, TRUE);
  63         }
  64
  65         if (graph->listeners.port_removed) {
  66                 g_array_free(graph->listeners.port_removed, TRUE);
  67         }
  68
  69         if (graph->listeners.ports_connected) {
  70                 g_array_free(graph->listeners.ports_connected, TRUE);
  71         }
  72
  73         if (graph->listeners.ports_disconnected) {
  74                 g_array_free(graph->listeners.ports_disconnected, TRUE);
  75         }
  76
  77         g_free(graph);
  78 }
  79
  80 static
  81 int init_listeners_array(GArray **listeners)
  82 {
  83         int ret = 0;
  84
  85         assert(listeners);
  86         *listeners = g_array_new(FALSE, TRUE, sizeof(struct bt_graph_listener));
  87         if (!*listeners) {
  88                 ret = -1;
  89                 goto end;
  90         }
  91
  92 end:
  93         return ret;
  94 }
  95
  96 struct bt_graph *bt_graph_create(void)
  97 {
  98         struct bt_graph *graph;
  99         int ret;
 100
 101         graph = g_new0(struct bt_graph, 1);
 102         if (!graph) {
 103                 goto end;
 104         }
 105
 106         bt_object_init(graph, bt_graph_destroy);
 107
 108         graph->connections = g_ptr_array_new_with_free_func(bt_object_release);
 109         if (!graph->connections) {
 110                 goto error;
 111         }
 112         graph->components = g_ptr_array_new_with_free_func(bt_object_release);
 113         if (!graph->components) {
 114                 goto error;
 115         }
 116         graph->sinks_to_consume = g_queue_new();
 117         if (!graph->sinks_to_consume) {
 118                 goto error;
 119         }
 120
 121         ret = init_listeners_array(&graph->listeners.port_added);
 122         if (ret) {
 123                 goto error;
 124         }
 125
 126         ret = init_listeners_array(&graph->listeners.port_removed);
 127         if (ret) {
 128                 goto error;
 129         }
 130
 131         ret = init_listeners_array(&graph->listeners.ports_connected);
 132         if (ret) {
 133                 goto error;
 134         }
 135
 136         ret = init_listeners_array(&graph->listeners.ports_disconnected);
 137         if (ret) {
 138                 goto error;
 139         }
 140
 141 end:
 142         return graph;
 143 error:
 144         BT_PUT(graph);
 145         goto end;
 146 }
 147
 148 struct bt_connection *bt_graph_connect_ports(struct bt_graph *graph,
 149                 struct bt_port *upstream_port,
 150                 struct bt_port *downstream_port)
 151 {
 152         struct bt_connection *connection = NULL;
 153         struct bt_graph *upstream_graph = NULL;
 154         struct bt_graph *downstream_graph = NULL;
 155         struct bt_component *upstream_component = NULL;
 156         struct bt_component *downstream_component = NULL;
 157         struct bt_connection *existing_conn = NULL;
 158         enum bt_component_status component_status;
 159         bool upstream_was_already_in_graph;
 160         bool downstream_was_already_in_graph;
 161         int components_to_remove = 0;
 162         int i;
 163
 164         if (!graph || !upstream_port || !downstream_port) {
 165                 goto end;
 166         }
 167
 168         /* Ensure appropriate types for upstream and downstream ports. */
 169         if (bt_port_get_type(upstream_port) != BT_PORT_TYPE_OUTPUT) {
 170                 goto end;
 171         }
 172         if (bt_port_get_type(downstream_port) != BT_PORT_TYPE_INPUT) {
 173                 goto end;
 174         }
 175
 176         /* Ensure that both ports are currently unconnected. */
 177         existing_conn = bt_port_get_connection(upstream_port);
 178         bt_put(existing_conn);
 179         if (existing_conn) {
 180                 fprintf(stderr, "Upstream port is already connected\n");
 181                 goto end;
 182         }
 183
 184         existing_conn = bt_port_get_connection(downstream_port);
 185         bt_put(existing_conn);
 186         if (existing_conn) {
 187                 fprintf(stderr, "Downstream port is already connected\n");
 188                 goto end;
 189         }
 190
 191         /*
 192          * Ensure that both ports are still attached to their creating
 193          * component.
 194          */
 195         upstream_component = bt_port_get_component(upstream_port);
 196         if (!upstream_component) {
 197                 fprintf(stderr, "Upstream port does not belong to a component\n");
 198                 goto end;
 199         }
 200
 201         downstream_component = bt_port_get_component(downstream_port);
 202         if (!downstream_component) {
 203                 fprintf(stderr, "Downstream port does not belong to a component\n");
 204                 goto end;
 205         }
 206
 207         /* Ensure the components are not already part of another graph. */
 208         upstream_graph = bt_component_get_graph(upstream_component);
 209         if (upstream_graph && (graph != upstream_graph)) {
 210                 fprintf(stderr, "Upstream component is already part of another graph\n");
 211                 goto error;
 212         }
 213         upstream_was_already_in_graph = (graph == upstream_graph);
 214         downstream_graph = bt_component_get_graph(downstream_component);
 215         if (downstream_graph && (graph != downstream_graph)) {
 216                 fprintf(stderr, "Downstream component is already part of another graph\n");
 217                 goto error;
 218         }
 219         downstream_was_already_in_graph = (graph == downstream_graph);
 220
 221         connection = bt_connection_create(graph, upstream_port,
 222                         downstream_port);
 223         if (!connection) {
 224                 goto error;
 225         }
 226
 227         /*
 228          * Ownership of upstream_component/downstream_component and of
 229          * the connection object is transferred to the graph.
 230          */
 231         g_ptr_array_add(graph->connections, connection);
 232
 233         if (!upstream_was_already_in_graph) {
 234                 g_ptr_array_add(graph->components, upstream_component);
 235                 bt_component_set_graph(upstream_component, graph);
 236         }
 237         if (!downstream_was_already_in_graph) {
 238                 g_ptr_array_add(graph->components, downstream_component);
 239                 bt_component_set_graph(downstream_component, graph);
 240                 if (bt_component_get_class_type(downstream_component) ==
 241                                 BT_COMPONENT_CLASS_TYPE_SINK) {
 242                         g_queue_push_tail(graph->sinks_to_consume,
 243                                         downstream_component);
 244                 }
 245         }
 246
 247         /*
 248          * The graph is now the parent of these components which garantees their
 249          * existence for the duration of the graph's lifetime.
 250          */
 251
 252         /*
 253          * The components and connection are added to the graph before
 254          * invoking the `accept_port_connection` method in order to make
 255          * them visible to the components during the method's
 256          * invocation.
 257          */
 258         component_status = bt_component_accept_port_connection(
 259                 upstream_component, upstream_port, downstream_port);
 260         if (component_status != BT_COMPONENT_STATUS_OK) {
 261                 goto error_rollback;
 262         }
 263         component_status = bt_component_accept_port_connection(
 264                 downstream_component, downstream_port, upstream_port);
 265         if (component_status != BT_COMPONENT_STATUS_OK) {
 266                 goto error_rollback;
 267         }
 268
 269         /*
 270          * Both components accepted the connection. Notify the graph's
 271          * creator that both ports are connected.
 272          */
 273         bt_graph_notify_ports_connected(graph, upstream_port, downstream_port);
 274
 275 end:
 276         bt_put(upstream_graph);
 277         bt_put(downstream_graph);
 278         bt_put(upstream_component);
 279         bt_put(downstream_component);
 280         return connection;
 281 error_rollback:
 282         /*
 283          * Remove newly-added components from the graph, being careful
 284          * not to remove a component that was already present in the graph
 285          * and is connected to other components.
 286          */
 287         components_to_remove += upstream_was_already_in_graph ? 0 : 1;
 288         components_to_remove += downstream_was_already_in_graph ? 0 : 1;
 289
 290         if (!downstream_was_already_in_graph) {
 291                 if (bt_component_get_class_type(downstream_component) ==
 292                                 BT_COMPONENT_CLASS_TYPE_SINK) {
 293                         g_queue_pop_tail(graph->sinks_to_consume);
 294                 }
 295         }
 296         /* Remove newly created connection. */
 297         g_ptr_array_set_size(graph->connections,
 298                         graph->connections->len - 1);
 299
 300         /*
 301          * Remove newly added components.
 302          *
 303          * Note that this is a tricky situation. The graph, being the parent
 304          * of the components, does not hold a reference to them. Normally,
 305          * components are destroyed right away when the graph is released since
 306          * the graph, being their parent, bounds their lifetime
 307          * (see doc/ref-counting.md).
 308          *
 309          * In this particular case, we must take a number of steps:
 310          *   1) unset the components' parent to rollback the initial state of
 311          *      the components being connected.
 312          *      Note that the reference taken by the component on its graph is
 313          *      released by the set_parent call.
 314          *   2) set the pointer in the components array to NULL so that the
 315          *      destruction function called on the array's resize in invoked on
 316          *      NULL (no effect),
 317          *
 318          * NOTE: Point #1 assumes that *something* holds a reference to both
 319          *       components being connected. The fact that a reference is being
 320          *       held to a component means that it must hold a reference to its
 321          *       parent to prevent the parent from being destroyed (again, refer
 322          *       to doc/red-counting.md). This reference to a component is
 323          *       most likely being held *transitively* by the caller which holds
 324          *       a reference to both ports (a port has its component as a
 325          *       parent).
 326          *
 327          *       This assumes that a graph is not connecting components by
 328          *       itself while not holding a reference to the ports/components
 329          *       being connected (i.e. "cheating" by using internal APIs).
 330          */
 331         for (i = 0; i < components_to_remove; i++) {
 332                 struct bt_component *component = g_ptr_array_index(
 333                                 graph->components, graph->components->len - 1);
 334
 335                 bt_component_set_graph(component, NULL);
 336                 g_ptr_array_index(graph->components,
 337                                 graph->components->len - 1) = NULL;
 338                 g_ptr_array_set_size(graph->components,
 339                                 graph->components->len - 1);
 340         }
 341         /* NOTE: Resizing the ptr_arrays invokes the destruction of the elements. */
 342         goto end;
 343 error:
 344         BT_PUT(upstream_component);
 345         BT_PUT(downstream_component);
 346         goto end;
 347 }
 348
 349 static
 350 enum bt_component_status get_component_port_counts(
 351                 struct bt_component *component, uint64_t *input_count,
 352                 uint64_t *output_count)
 353 {
 354         enum bt_component_status ret;
 355
 356         switch (bt_component_get_class_type(component)) {
 357         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 358                 ret = bt_component_source_get_output_port_count(component,
 359                                 output_count);
 360                 if (ret != BT_COMPONENT_STATUS_OK) {
 361                         goto end;
 362                 }
 363                 break;
 364         case BT_COMPONENT_CLASS_TYPE_FILTER:
 365                 ret = bt_component_filter_get_output_port_count(component,
 366                                 output_count);
 367                 if (ret != BT_COMPONENT_STATUS_OK) {
 368                         goto end;
 369                 }
 370                 ret = bt_component_filter_get_input_port_count(component,
 371                                 input_count);
 372                 if (ret != BT_COMPONENT_STATUS_OK) {
 373                         goto end;
 374                 }
 375                 break;
 376         case BT_COMPONENT_CLASS_TYPE_SINK:
 377                 ret = bt_component_sink_get_input_port_count(component,
 378                                 input_count);
 379                 if (ret != BT_COMPONENT_STATUS_OK) {
 380                         goto end;
 381                 }
 382                 break;
 383         default:
 384                 assert(false);
 385                 break;
 386         }
 387         ret = BT_COMPONENT_STATUS_OK;
 388 end:
 389         return ret;
 390 }
 391
 392 enum bt_graph_status bt_graph_add_component_as_sibling(struct bt_graph *graph,
 393                 struct bt_component *origin,
 394                 struct bt_component *new_component)
 395 {
 396         uint64_t origin_input_port_count = 0;
 397         uint64_t origin_output_port_count = 0;
 398         uint64_t new_input_port_count = 0;
 399         uint64_t new_output_port_count = 0;
 400         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 401         struct bt_graph *origin_graph = NULL;
 402         struct bt_graph *new_graph = NULL;
 403         struct bt_port *origin_port = NULL;
 404         struct bt_port *new_port = NULL;
 405         struct bt_port *upstream_port = NULL;
 406         struct bt_port *downstream_port = NULL;
 407         struct bt_connection *origin_connection = NULL;
 408         struct bt_connection *new_connection = NULL;
 409         int port_index;
 410
 411         if (!graph || !origin || !new_component) {
 412                 status = BT_GRAPH_STATUS_INVALID;
 413                 goto end;
 414         }
 415
 416         if (bt_component_get_class_type(origin) !=
 417                         bt_component_get_class_type(new_component)) {
 418                 status = BT_GRAPH_STATUS_INVALID;
 419                 goto end;
 420         }
 421
 422         origin_graph = bt_component_get_graph(origin);
 423         if (!origin_graph || (origin_graph != graph)) {
 424                 status = BT_GRAPH_STATUS_INVALID;
 425                 goto end;
 426         }
 427
 428         new_graph = bt_component_get_graph(new_component);
 429         if (new_graph) {
 430                 status = BT_GRAPH_STATUS_ALREADY_IN_A_GRAPH;
 431                 goto end;
 432         }
 433
 434         if (get_component_port_counts(origin, &origin_input_port_count,
 435                         &origin_output_port_count) != BT_COMPONENT_STATUS_OK) {
 436                 status = BT_GRAPH_STATUS_INVALID;
 437                 goto end;
 438         }
 439         if (get_component_port_counts(new_component, &new_input_port_count,
 440                         &new_output_port_count) != BT_COMPONENT_STATUS_OK) {
 441                 status = BT_GRAPH_STATUS_INVALID;
 442                 goto end;
 443         }
 444
 445         if (origin_input_port_count != new_input_port_count ||
 446                         origin_output_port_count != new_output_port_count) {
 447                 status = BT_GRAPH_STATUS_INVALID;
 448                 goto end;
 449         }
 450
 451         /* Replicate input connections. */
 452         for (port_index = 0; port_index< origin_input_port_count; port_index++) {
 453                 origin_port = bt_component_get_input_port_at_index(origin,
 454                         port_index);
 455                 if (!origin_port) {
 456                         status = BT_GRAPH_STATUS_ERROR;
 457                         goto error_disconnect;
 458                 }
 459
 460                 new_port = bt_component_get_input_port_at_index(new_component,
 461                         port_index);
 462                 if (!new_port) {
 463                         status = BT_GRAPH_STATUS_ERROR;
 464                         goto error_disconnect;
 465                 }
 466
 467                 origin_connection = bt_port_get_connection(origin_port);
 468                 if (origin_connection) {
 469                         upstream_port = bt_connection_get_upstream_port(
 470                                 origin_connection);
 471                         if (!upstream_port) {
 472                                 goto error_disconnect;
 473                         }
 474
 475                         new_connection = bt_graph_connect_ports(graph,
 476                                 upstream_port, new_port);
 477                         if (!new_connection) {
 478                                 goto error_disconnect;
 479                         }
 480                 }
 481
 482                 BT_PUT(upstream_port);
 483                 BT_PUT(origin_connection);
 484                 BT_PUT(new_connection);
 485                 BT_PUT(origin_port);
 486                 BT_PUT(new_port);
 487         }
 488
 489         /* Replicate output connections. */
 490         for (port_index = 0; port_index < origin_output_port_count; port_index++) {
 491                 origin_port = bt_component_get_output_port_at_index(origin,
 492                         port_index);
 493                 if (!origin_port) {
 494                         status = BT_GRAPH_STATUS_ERROR;
 495                         goto error_disconnect;
 496                 }
 497                 new_port = bt_component_get_output_port_at_index(new_component,
 498                         port_index);
 499                 if (!new_port) {
 500                         status = BT_GRAPH_STATUS_ERROR;
 501                         goto error_disconnect;
 502                 }
 503
 504                 origin_connection = bt_port_get_connection(origin_port);
 505                 if (origin_connection) {
 506                         downstream_port = bt_connection_get_downstream_port(
 507                                         origin_connection);
 508                         if (!downstream_port) {
 509                                 goto error_disconnect;
 510                         }
 511
 512                         new_connection = bt_graph_connect_ports(graph,
 513                                 new_port, downstream_port);
 514                         if (!new_connection) {
 515                                 goto error_disconnect;
 516                         }
 517                 }
 518
 519                 BT_PUT(downstream_port);
 520                 BT_PUT(origin_connection);
 521                 BT_PUT(new_connection);
 522                 BT_PUT(origin_port);
 523                 BT_PUT(new_port);
 524         }
 525 end:
 526         bt_put(origin_graph);
 527         bt_put(new_graph);
 528         bt_put(origin_port);
 529         bt_put(new_port);
 530         bt_put(upstream_port);
 531         bt_put(downstream_port);
 532         bt_put(origin_connection);
 533         bt_put(new_connection);
 534         return status;
 535 error_disconnect:
 536         /* Destroy all connections of the new component. */
 537         /* FIXME. */
 538         goto end;
 539 }
 540
 541 enum bt_graph_status bt_graph_consume(struct bt_graph *graph)
 542 {
 543         struct bt_component *sink;
 544         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 545         enum bt_component_status comp_status;
 546         GList *current_node;
 547
 548         if (!graph) {
 549                 status = BT_GRAPH_STATUS_INVALID;
 550                 goto end;
 551         }
 552
 553         if (g_queue_is_empty(graph->sinks_to_consume)) {
 554                 status = BT_GRAPH_STATUS_END;
 555                 goto end;
 556         }
 557
 558         current_node = g_queue_pop_head_link(graph->sinks_to_consume);
 559         sink = current_node->data;
 560         comp_status = bt_component_sink_consume(sink);
 561         switch (comp_status) {
 562         case BT_COMPONENT_STATUS_OK:
 563                 break;
 564         case BT_COMPONENT_STATUS_END:
 565                 status = BT_GRAPH_STATUS_END;
 566                 break;
 567         case BT_COMPONENT_STATUS_AGAIN:
 568                 status = BT_GRAPH_STATUS_AGAIN;
 569                 break;
 570         case BT_COMPONENT_STATUS_INVALID:
 571                 status = BT_GRAPH_STATUS_INVALID;
 572                 break;
 573         default:
 574                 status = BT_GRAPH_STATUS_ERROR;
 575                 break;
 576         }
 577
 578         if (status != BT_GRAPH_STATUS_END) {
 579                 g_queue_push_tail_link(graph->sinks_to_consume, current_node);
 580                 goto end;
 581         }
 582
 583         /* End reached, the node is not added back to the queue and free'd. */
 584         g_queue_delete_link(graph->sinks_to_consume, current_node);
 585
 586         /* Don't forward an END status if there are sinks left to consume. */
 587         if (!g_queue_is_empty(graph->sinks_to_consume)) {
 588                 status = BT_GRAPH_STATUS_OK;
 589                 goto end;
 590         }
 591 end:
 592         return status;
 593 }
 594
 595 enum bt_graph_status bt_graph_run(struct bt_graph *graph)
 596 {
 597         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 598
 599         if (!graph) {
 600                 status = BT_GRAPH_STATUS_INVALID;
 601                 goto error;
 602         }
 603
 604         do {
 605                 status = bt_graph_consume(graph);
 606                 if (status == BT_GRAPH_STATUS_AGAIN) {
 607                         /*
 608                          * If AGAIN is received and there are multiple sinks,
 609                          * go ahead and consume from the next sink.
 610                          *
 611                          * However, in the case where a single sink is left,
 612                          * the caller can decide to busy-wait and call
 613                          * bt_graph_run continuously until the source is ready
 614                          * or it can decide to sleep for an arbitrary amount of
 615                          * time.
 616                          */
 617                         if (graph->sinks_to_consume->length > 1) {
 618                                 status = BT_GRAPH_STATUS_OK;
 619                         }
 620                 }
 621         } while (status == BT_GRAPH_STATUS_OK);
 622
 623         if (g_queue_is_empty(graph->sinks_to_consume)) {
 624                 status = BT_GRAPH_STATUS_END;
 625         }
 626 error:
 627         return status;
 628 }
 629
 630 static
 631 void add_listener(GArray *listeners, void *func, void *data)
 632 {
 633         struct bt_graph_listener listener = {
 634                 .func = func,
 635                 .data = data,
 636         };
 637
 638         g_array_append_val(listeners, listener);
 639 }
 640
 641 enum bt_graph_status bt_graph_add_port_added_listener(
 642                 struct bt_graph *graph,
 643                 bt_graph_port_added_listener listener, void *data)
 644 {
 645         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 646
 647         if (!graph || !listener) {
 648                 status = BT_GRAPH_STATUS_INVALID;
 649                 goto end;
 650         }
 651
 652         add_listener(graph->listeners.port_added, listener, data);
 653
 654 end:
 655         return status;
 656 }
 657
 658 enum bt_graph_status bt_graph_add_port_removed_listener(
 659                 struct bt_graph *graph,
 660                 bt_graph_port_removed_listener listener, void *data)
 661 {
 662         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 663
 664         if (!graph || !listener) {
 665                 status = BT_GRAPH_STATUS_INVALID;
 666                 goto end;
 667         }
 668
 669         add_listener(graph->listeners.port_removed, listener, data);
 670
 671 end:
 672         return status;
 673 }
 674
 675 enum bt_graph_status bt_graph_add_ports_connected_listener(
 676                 struct bt_graph *graph,
 677                 bt_graph_ports_connected_listener listener, void *data)
 678 {
 679         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 680
 681         if (!graph || !listener) {
 682                 status = BT_GRAPH_STATUS_INVALID;
 683                 goto end;
 684         }
 685
 686         add_listener(graph->listeners.ports_connected, listener, data);
 687
 688 end:
 689         return status;
 690 }
 691
 692 enum bt_graph_status bt_graph_add_ports_disconnected_listener(
 693                 struct bt_graph *graph,
 694                 bt_graph_ports_disconnected_listener listener, void *data)
 695 {
 696         enum bt_graph_status status = BT_GRAPH_STATUS_OK;
 697
 698         if (!graph || !listener) {
 699                 status = BT_GRAPH_STATUS_INVALID;
 700                 goto end;
 701         }
 702
 703         add_listener(graph->listeners.ports_disconnected, listener, data);
 704
 705 end:
 706         return status;
 707 }
 708
 709 BT_HIDDEN
 710 void bt_graph_notify_port_added(struct bt_graph *graph, struct bt_port *port)
 711 {
 712         size_t i;
 713
 714         for (i = 0; i < graph->listeners.port_added->len; i++) {
 715                 struct bt_graph_listener listener =
 716                         g_array_index(graph->listeners.port_added,
 717                                 struct bt_graph_listener, i);
 718                 bt_graph_port_added_listener func = listener.func;
 719
 720                 assert(func);
 721                 func(port, listener.data);
 722         }
 723 }
 724
 725 BT_HIDDEN
 726 void bt_graph_notify_port_removed(struct bt_graph *graph,
 727                 struct bt_component *comp, struct bt_port *port)
 728 {
 729         size_t i;
 730
 731         for (i = 0; i < graph->listeners.port_removed->len; i++) {
 732                 struct bt_graph_listener listener =
 733                         g_array_index(graph->listeners.port_removed,
 734                                 struct bt_graph_listener, i);
 735                 bt_graph_port_removed_listener func = listener.func;
 736
 737                 assert(func);
 738                 func(comp, port, listener.data);
 739         }
 740 }
 741
 742 BT_HIDDEN
 743 void bt_graph_notify_ports_connected(struct bt_graph *graph,
 744                 struct bt_port *upstream_port, struct bt_port *downstream_port)
 745 {
 746         size_t i;
 747
 748         for (i = 0; i < graph->listeners.ports_connected->len; i++) {
 749                 struct bt_graph_listener listener =
 750                         g_array_index(graph->listeners.ports_connected,
 751                                 struct bt_graph_listener, i);
 752                 bt_graph_ports_connected_listener func = listener.func;
 753
 754                 assert(func);
 755                 func(upstream_port, downstream_port, listener.data);
 756         }
 757 }
 758
 759 BT_HIDDEN
 760 void bt_graph_notify_ports_disconnected(struct bt_graph *graph,
 761                 struct bt_component *upstream_comp,
 762                 struct bt_component *downstream_comp,
 763                 struct bt_port *upstream_port, struct bt_port *downstream_port)
 764 {
 765         size_t i;
 766
 767         for (i = 0; i < graph->listeners.ports_disconnected->len; i++) {
 768                 struct bt_graph_listener listener =
 769                         g_array_index(graph->listeners.ports_disconnected,
 770                                 struct bt_graph_listener, i);
 771                 bt_graph_ports_disconnected_listener func = listener.func;
 772
 773                 assert(func);
 774                 func(upstream_comp, downstream_comp, upstream_port,
 775                         downstream_port, listener.data);
 776         }
 777 }