lib/graph/iterator.c

   1 /*
   2  * iterator.c
   3  *
   4  * Babeltrace Notification Iterator
   5  *
   6  * Copyright 2015 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   7  * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
   8  *
   9  * Permission is hereby granted, free of charge, to any person obtaining a copy
  10  * of this software and associated documentation files (the "Software"), to deal
  11  * in the Software without restriction, including without limitation the rights
  12  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  13  * copies of the Software, and to permit persons to whom the Software is
  14  * furnished to do so, subject to the following conditions:
  15  *
  16  * The above copyright notice and this permission notice shall be included in
  17  * all copies or substantial portions of the Software.
  18  *
  19  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  20  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  21  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  22  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  23  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  24  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  25  * SOFTWARE.
  26  */
  27
  28 #include <babeltrace/compiler-internal.h>
  29 #include <babeltrace/ref.h>
  30 #include <babeltrace/ctf-ir/event-internal.h>
  31 #include <babeltrace/ctf-ir/packet-internal.h>
  32 #include <babeltrace/ctf-ir/stream-internal.h>
  33 #include <babeltrace/graph/connection-internal.h>
  34 #include <babeltrace/graph/component.h>
  35 #include <babeltrace/graph/component-source-internal.h>
  36 #include <babeltrace/graph/component-class-internal.h>
  37 #include <babeltrace/graph/notification.h>
  38 #include <babeltrace/graph/notification-iterator.h>
  39 #include <babeltrace/graph/notification-iterator-internal.h>
  40 #include <babeltrace/graph/notification-internal.h>
  41 #include <babeltrace/graph/notification-event.h>
  42 #include <babeltrace/graph/notification-event-internal.h>
  43 #include <babeltrace/graph/notification-packet.h>
  44 #include <babeltrace/graph/notification-packet-internal.h>
  45 #include <babeltrace/graph/notification-stream.h>
  46 #include <babeltrace/graph/notification-stream-internal.h>
  47 #include <babeltrace/graph/port.h>
  48 #include <babeltrace/types.h>
  49 #include <stdint.h>
  50
  51 struct stream_state {
  52         struct bt_ctf_stream *stream; /* owned by this */
  53         struct bt_ctf_packet *cur_packet; /* owned by this */
  54         bt_bool is_ended;
  55 };
  56
  57 enum action_type {
  58         ACTION_TYPE_PUSH_NOTIF,
  59         ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM,
  60         ACTION_TYPE_ADD_STREAM_STATE,
  61         ACTION_TYPE_SET_STREAM_STATE_IS_ENDED,
  62         ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET,
  63 };
  64
  65 struct action {
  66         enum action_type type;
  67         union {
  68                 /* ACTION_TYPE_PUSH_NOTIF */
  69                 struct {
  70                         struct bt_notification *notif; /* owned by this */
  71                 } push_notif;
  72
  73                 /* ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM */
  74                 struct {
  75                         struct bt_ctf_stream *stream; /* owned by this */
  76                         struct bt_component *component; /* owned by this */
  77                         struct bt_port *port; /* owned by this */
  78                 } map_port_to_comp_in_stream;
  79
  80                 /* ACTION_TYPE_ADD_STREAM_STATE */
  81                 struct {
  82                         struct bt_ctf_stream *stream; /* owned by this */
  83                         struct stream_state *stream_state; /* owned by this */
  84                 } add_stream_state;
  85
  86                 /* ACTION_TYPE_SET_STREAM_STATE_IS_ENDED */
  87                 struct {
  88                         struct stream_state *stream_state; /* weak */
  89                 } set_stream_state_is_ended;
  90
  91                 /* ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET */
  92                 struct {
  93                         struct stream_state *stream_state; /* weak */
  94                         struct bt_ctf_packet *packet; /* owned by this */
  95                 } set_stream_state_cur_packet;
  96         } payload;
  97 };
  98
  99 static
 100 void stream_destroy_listener(struct bt_ctf_stream *stream, void *data)
 101 {
 102         struct bt_notification_iterator *iterator = data;
 103
 104         /* Remove associated stream state */
 105         g_hash_table_remove(iterator->stream_states, stream);
 106 }
 107
 108 static
 109 void destroy_stream_state(struct stream_state *stream_state)
 110 {
 111         if (!stream_state) {
 112                 return;
 113         }
 114
 115         bt_put(stream_state->cur_packet);
 116         bt_put(stream_state->stream);
 117         g_free(stream_state);
 118 }
 119
 120 static
 121 void destroy_action(struct action *action)
 122 {
 123         assert(action);
 124
 125         switch (action->type) {
 126         case ACTION_TYPE_PUSH_NOTIF:
 127                 BT_PUT(action->payload.push_notif.notif);
 128                 break;
 129         case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM:
 130                 BT_PUT(action->payload.map_port_to_comp_in_stream.stream);
 131                 BT_PUT(action->payload.map_port_to_comp_in_stream.component);
 132                 BT_PUT(action->payload.map_port_to_comp_in_stream.port);
 133                 break;
 134         case ACTION_TYPE_ADD_STREAM_STATE:
 135                 BT_PUT(action->payload.add_stream_state.stream);
 136                 destroy_stream_state(
 137                         action->payload.add_stream_state.stream_state);
 138                 action->payload.add_stream_state.stream_state = NULL;
 139                 break;
 140         case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET:
 141                 BT_PUT(action->payload.set_stream_state_cur_packet.packet);
 142                 break;
 143         case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED:
 144                 break;
 145         default:
 146                 assert(BT_FALSE);
 147         }
 148 }
 149
 150 static
 151 void add_action(struct bt_notification_iterator *iterator,
 152                 struct action *action)
 153 {
 154         g_array_append_val(iterator->actions, *action);
 155 }
 156
 157 static
 158 void clear_actions(struct bt_notification_iterator *iterator)
 159 {
 160         size_t i;
 161
 162         for (i = 0; i < iterator->actions->len; i++) {
 163                 struct action *action = &g_array_index(iterator->actions,
 164                         struct action, i);
 165
 166                 destroy_action(action);
 167         }
 168
 169         g_array_set_size(iterator->actions, 0);
 170 }
 171
 172 static
 173 void apply_actions(struct bt_notification_iterator *iterator)
 174 {
 175         size_t i;
 176
 177         for (i = 0; i < iterator->actions->len; i++) {
 178                 struct action *action = &g_array_index(iterator->actions,
 179                         struct action, i);
 180
 181                 switch (action->type) {
 182                 case ACTION_TYPE_PUSH_NOTIF:
 183                         /* Move notification to queue */
 184                         g_queue_push_head(iterator->queue,
 185                                 action->payload.push_notif.notif);
 186                         bt_notification_freeze(
 187                                 action->payload.push_notif.notif);
 188                         action->payload.push_notif.notif = NULL;
 189                         break;
 190                 case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM:
 191                         bt_ctf_stream_map_component_to_port(
 192                                 action->payload.map_port_to_comp_in_stream.stream,
 193                                 action->payload.map_port_to_comp_in_stream.component,
 194                                 action->payload.map_port_to_comp_in_stream.port);
 195                         break;
 196                 case ACTION_TYPE_ADD_STREAM_STATE:
 197                         /* Move stream state to hash table */
 198                         g_hash_table_insert(iterator->stream_states,
 199                                 action->payload.add_stream_state.stream,
 200                                 action->payload.add_stream_state.stream_state);
 201
 202                         action->payload.add_stream_state.stream_state = NULL;
 203                         break;
 204                 case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED:
 205                         /*
 206                          * We know that this stream is ended. We need to
 207                          * remember this as long as the stream exists to
 208                          * enforce that the same stream does not end
 209                          * twice.
 210                          *
 211                          * Here we add a destroy listener to the stream
 212                          * which we put after (becomes weak as the hash
 213                          * table key). If we were the last object to own
 214                          * this stream, the destroy listener is called
 215                          * when we call bt_put() which removes this
 216                          * stream state completely. This is important
 217                          * because the memory used by this stream object
 218                          * could be reused for another stream, and they
 219                          * must have different states.
 220                          */
 221                         bt_ctf_stream_add_destroy_listener(
 222                                 action->payload.set_stream_state_is_ended.stream_state->stream,
 223                                 stream_destroy_listener, iterator);
 224                         action->payload.set_stream_state_is_ended.stream_state->is_ended = BT_TRUE;
 225                         BT_PUT(action->payload.set_stream_state_is_ended.stream_state->stream);
 226                         break;
 227                 case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET:
 228                         /* Move packet to stream state's current packet */
 229                         BT_MOVE(action->payload.set_stream_state_cur_packet.stream_state->cur_packet,
 230                                 action->payload.set_stream_state_cur_packet.packet);
 231                         break;
 232                 default:
 233                         assert(BT_FALSE);
 234                 }
 235         }
 236
 237         clear_actions(iterator);
 238 }
 239
 240 static
 241 struct stream_state *create_stream_state(struct bt_ctf_stream *stream)
 242 {
 243         struct stream_state *stream_state = g_new0(struct stream_state, 1);
 244
 245         if (!stream_state) {
 246                 goto end;
 247         }
 248
 249         /*
 250          * We keep a reference to the stream until we know it's ended
 251          * because we need to be able to create an automatic "stream
 252          * end" notification when the user's "next" method returns
 253          * BT_NOTIFICATION_ITERATOR_STATUS_END.
 254          *
 255          * We put this reference when the stream is marked as ended.
 256          */
 257         stream_state->stream = bt_get(stream);
 258
 259 end:
 260         return stream_state;
 261 }
 262
 263 static
 264 void bt_notification_iterator_destroy(struct bt_object *obj)
 265 {
 266         struct bt_notification_iterator *iterator;
 267
 268         assert(obj);
 269
 270         /*
 271          * The notification iterator's reference count is 0 if we're
 272          * here. Increment it to avoid a double-destroy (possibly
 273          * infinitely recursive). This could happen for example if the
 274          * notification iterator's finalization function does bt_get()
 275          * (or anything that causes bt_get() to be called) on itself
 276          * (ref. count goes from 0 to 1), and then bt_put(): the
 277          * reference count would go from 1 to 0 again and this function
 278          * would be called again.
 279          */
 280         obj->ref_count.count++;
 281         iterator = container_of(obj, struct bt_notification_iterator, base);
 282         bt_notification_iterator_finalize(iterator);
 283
 284         if (iterator->queue) {
 285                 struct bt_notification *notif;
 286
 287                 while ((notif = g_queue_pop_tail(iterator->queue))) {
 288                         bt_put(notif);
 289                 }
 290
 291                 g_queue_free(iterator->queue);
 292         }
 293
 294         if (iterator->stream_states) {
 295                 /*
 296                  * Remove our destroy listener from each stream which
 297                  * has a state in this iterator. Otherwise the destroy
 298                  * listener would be called with an invalid/other
 299                  * notification iterator object.
 300                  */
 301                 GHashTableIter ht_iter;
 302                 gpointer stream_gptr, stream_state_gptr;
 303
 304                 g_hash_table_iter_init(&ht_iter, iterator->stream_states);
 305
 306                 while (g_hash_table_iter_next(&ht_iter, &stream_gptr, &stream_state_gptr)) {
 307                         assert(stream_gptr);
 308                         bt_ctf_stream_remove_destroy_listener(
 309                                 (void *) stream_gptr, stream_destroy_listener,
 310                                 iterator);
 311                 }
 312
 313                 g_hash_table_destroy(iterator->stream_states);
 314         }
 315
 316         if (iterator->actions) {
 317                 g_array_free(iterator->actions, TRUE);
 318         }
 319
 320         if (iterator->connection) {
 321                 /*
 322                  * Remove ourself from the originating connection so
 323                  * that it does not try to finalize a dangling pointer
 324                  * later.
 325                  */
 326                 bt_connection_remove_iterator(iterator->connection, iterator);
 327         }
 328
 329         bt_put(iterator->current_notification);
 330         g_free(iterator);
 331 }
 332
 333 BT_HIDDEN
 334 void bt_notification_iterator_finalize(
 335                 struct bt_notification_iterator *iterator)
 336 {
 337         struct bt_component_class *comp_class = NULL;
 338         bt_component_class_notification_iterator_finalize_method
 339                 finalize_method = NULL;
 340
 341         assert(iterator);
 342
 343         switch (iterator->state) {
 344         case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED:
 345         case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED:
 346                 /* Already finalized */
 347                 return;
 348         default:
 349                 break;
 350         }
 351
 352         if (iterator->state == BT_NOTIFICATION_ITERATOR_STATE_ENDED) {
 353                 iterator->state = BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED;
 354         } else {
 355                 iterator->state = BT_NOTIFICATION_ITERATOR_STATE_FINALIZED;
 356         }
 357
 358         assert(iterator->upstream_component);
 359         comp_class = iterator->upstream_component->class;
 360
 361         /* Call user-defined destroy method */
 362         switch (comp_class->type) {
 363         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 364         {
 365                 struct bt_component_class_source *source_class;
 366
 367                 source_class = container_of(comp_class, struct bt_component_class_source, parent);
 368                 finalize_method = source_class->methods.iterator.finalize;
 369                 break;
 370         }
 371         case BT_COMPONENT_CLASS_TYPE_FILTER:
 372         {
 373                 struct bt_component_class_filter *filter_class;
 374
 375                 filter_class = container_of(comp_class, struct bt_component_class_filter, parent);
 376                 finalize_method = filter_class->methods.iterator.finalize;
 377                 break;
 378         }
 379         default:
 380                 /* Unreachable */
 381                 assert(0);
 382         }
 383
 384         if (finalize_method) {
 385                 finalize_method(
 386                         bt_private_notification_iterator_from_notification_iterator(iterator));
 387         }
 388
 389         iterator->upstream_component = NULL;
 390         iterator->upstream_port = NULL;
 391 }
 392
 393 BT_HIDDEN
 394 void bt_notification_iterator_set_connection(
 395                 struct bt_notification_iterator *iterator,
 396                 struct bt_connection *connection)
 397 {
 398         assert(iterator);
 399         iterator->connection = connection;
 400 }
 401
 402 static
 403 int create_subscription_mask_from_notification_types(
 404                 struct bt_notification_iterator *iterator,
 405                 const enum bt_notification_type *notif_types)
 406 {
 407         const enum bt_notification_type *notif_type;
 408         int ret = 0;
 409
 410         assert(notif_types);
 411         iterator->subscription_mask = 0;
 412
 413         for (notif_type = notif_types;
 414                         *notif_type != BT_NOTIFICATION_TYPE_SENTINEL;
 415                         notif_type++) {
 416                 switch (*notif_type) {
 417                 case BT_NOTIFICATION_TYPE_ALL:
 418                         iterator->subscription_mask |=
 419                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT |
 420                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY |
 421                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN |
 422                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END |
 423                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN |
 424                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END;
 425                         break;
 426                 case BT_NOTIFICATION_TYPE_EVENT:
 427                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT;
 428                         break;
 429                 case BT_NOTIFICATION_TYPE_INACTIVITY:
 430                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY;
 431                         break;
 432                 case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 433                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN;
 434                         break;
 435                 case BT_NOTIFICATION_TYPE_STREAM_END:
 436                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END;
 437                         break;
 438                 case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 439                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN;
 440                         break;
 441                 case BT_NOTIFICATION_TYPE_PACKET_END:
 442                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END;
 443                         break;
 444                 default:
 445                         ret = -1;
 446                         goto end;
 447                 }
 448         }
 449
 450 end:
 451         return ret;
 452 }
 453
 454 BT_HIDDEN
 455 struct bt_notification_iterator *bt_notification_iterator_create(
 456                 struct bt_component *upstream_comp,
 457                 struct bt_port *upstream_port,
 458                 const enum bt_notification_type *notification_types,
 459                 struct bt_connection *connection)
 460 {
 461         enum bt_component_class_type type;
 462         struct bt_notification_iterator *iterator = NULL;
 463
 464         assert(upstream_comp);
 465         assert(upstream_port);
 466         assert(notification_types);
 467         assert(bt_port_is_connected(upstream_port));
 468
 469         type = bt_component_get_class_type(upstream_comp);
 470         assert(type == BT_COMPONENT_CLASS_TYPE_SOURCE ||
 471                 type == BT_COMPONENT_CLASS_TYPE_FILTER);
 472         iterator = g_new0(struct bt_notification_iterator, 1);
 473         if (!iterator) {
 474                 goto error;
 475         }
 476
 477         bt_object_init(iterator, bt_notification_iterator_destroy);
 478
 479         if (create_subscription_mask_from_notification_types(iterator,
 480                         notification_types)) {
 481                 goto error;
 482         }
 483
 484         iterator->stream_states = g_hash_table_new_full(g_direct_hash,
 485                 g_direct_equal, NULL, (GDestroyNotify) destroy_stream_state);
 486         if (!iterator->stream_states) {
 487                 goto error;
 488         }
 489
 490         iterator->queue = g_queue_new();
 491         if (!iterator->queue) {
 492                 goto error;
 493         }
 494
 495         iterator->actions = g_array_new(FALSE, FALSE, sizeof(struct action));
 496         if (!iterator->actions) {
 497                 goto error;
 498         }
 499
 500         iterator->upstream_component = upstream_comp;
 501         iterator->upstream_port = upstream_port;
 502         iterator->connection = connection;
 503         iterator->state = BT_NOTIFICATION_ITERATOR_STATE_ACTIVE;
 504         goto end;
 505
 506 error:
 507         BT_PUT(iterator);
 508
 509 end:
 510         return iterator;
 511 }
 512
 513 void *bt_private_notification_iterator_get_user_data(
 514                 struct bt_private_notification_iterator *private_iterator)
 515 {
 516         struct bt_notification_iterator *iterator =
 517                 bt_notification_iterator_from_private(private_iterator);
 518
 519         return iterator ? iterator->user_data : NULL;
 520 }
 521
 522 enum bt_notification_iterator_status
 523 bt_private_notification_iterator_set_user_data(
 524                 struct bt_private_notification_iterator *private_iterator,
 525                 void *data)
 526 {
 527         enum bt_notification_iterator_status ret =
 528                         BT_NOTIFICATION_ITERATOR_STATUS_OK;
 529         struct bt_notification_iterator *iterator =
 530                 bt_notification_iterator_from_private(private_iterator);
 531
 532         if (!iterator) {
 533                 ret = BT_NOTIFICATION_ITERATOR_STATUS_INVALID;
 534                 goto end;
 535         }
 536
 537         iterator->user_data = data;
 538 end:
 539         return ret;
 540 }
 541
 542 struct bt_notification *bt_notification_iterator_get_notification(
 543                 struct bt_notification_iterator *iterator)
 544 {
 545         struct bt_notification *notification = NULL;
 546
 547         if (!iterator) {
 548                 goto end;
 549         }
 550
 551         notification = bt_get(iterator->current_notification);
 552
 553 end:
 554         return notification;
 555 }
 556
 557 static
 558 enum bt_notification_iterator_notif_type
 559 bt_notification_iterator_notif_type_from_notif_type(
 560                 enum bt_notification_type notif_type)
 561 {
 562         enum bt_notification_iterator_notif_type iter_notif_type;
 563
 564         switch (notif_type) {
 565         case BT_NOTIFICATION_TYPE_EVENT:
 566                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT;
 567                 break;
 568         case BT_NOTIFICATION_TYPE_INACTIVITY:
 569                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY;
 570                 break;
 571         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 572                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN;
 573                 break;
 574         case BT_NOTIFICATION_TYPE_STREAM_END:
 575                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END;
 576                 break;
 577         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 578                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN;
 579                 break;
 580         case BT_NOTIFICATION_TYPE_PACKET_END:
 581                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END;
 582                 break;
 583         default:
 584                 assert(BT_FALSE);
 585         }
 586
 587         return iter_notif_type;
 588 }
 589
 590 static
 591 bt_bool validate_notification(struct bt_notification_iterator *iterator,
 592                 struct bt_notification *notif,
 593                 struct bt_ctf_stream *notif_stream,
 594                 struct bt_ctf_packet *notif_packet)
 595 {
 596         bt_bool is_valid = BT_TRUE;
 597         struct stream_state *stream_state;
 598         struct bt_port *stream_comp_cur_port;
 599
 600         assert(notif_stream);
 601         stream_comp_cur_port =
 602                 bt_ctf_stream_port_for_component(notif_stream,
 603                         iterator->upstream_component);
 604         if (!stream_comp_cur_port) {
 605                 /*
 606                  * This is the first time this notification iterator
 607                  * bumps into this stream. Add an action to map the
 608                  * iterator's upstream component to the iterator's
 609                  * upstream port in this stream.
 610                  */
 611                 struct action action = {
 612                         .type = ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM,
 613                         .payload.map_port_to_comp_in_stream = {
 614                                 .stream = bt_get(notif_stream),
 615                                 .component = bt_get(iterator->upstream_component),
 616                                 .port = bt_get(iterator->upstream_port),
 617                         },
 618                 };
 619
 620                 add_action(iterator, &action);
 621         } else {
 622                 if (stream_comp_cur_port != iterator->upstream_port) {
 623                         /*
 624                          * It looks like two different ports of the same
 625                          * component are emitting notifications which
 626                          * have references to the same stream. This is
 627                          * bad: the API guarantees that it can never
 628                          * happen.
 629                          */
 630                         is_valid = BT_FALSE;
 631                         goto end;
 632                 }
 633
 634         }
 635
 636         stream_state = g_hash_table_lookup(iterator->stream_states,
 637                 notif_stream);
 638         if (stream_state) {
 639                 if (stream_state->is_ended) {
 640                         /*
 641                          * There's a new notification which has a
 642                          * reference to a stream which, from this
 643                          * iterator's point of view, is ended ("end of
 644                          * stream" notification was returned). This is
 645                          * bad: the API guarantees that it can never
 646                          * happen.
 647                          */
 648                         is_valid = BT_FALSE;
 649                         goto end;
 650                 }
 651
 652                 switch (notif->type) {
 653                 case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 654                         /*
 655                          * We already have a stream state, which means
 656                          * we already returned a "stream begin"
 657                          * notification: this is an invalid duplicate.
 658                          */
 659                         is_valid = BT_FALSE;
 660                         goto end;
 661                 case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 662                         if (notif_packet == stream_state->cur_packet) {
 663                                 /* Duplicate "packet begin" notification */
 664                                 is_valid = BT_FALSE;
 665                                 goto end;
 666                         }
 667                         break;
 668                 default:
 669                         break;
 670                 }
 671         }
 672
 673 end:
 674         return is_valid;
 675 }
 676
 677 static
 678 bt_bool is_subscribed_to_notification_type(struct bt_notification_iterator *iterator,
 679                 enum bt_notification_type notif_type)
 680 {
 681         uint32_t iter_notif_type =
 682                 (uint32_t) bt_notification_iterator_notif_type_from_notif_type(
 683                         notif_type);
 684
 685         return (iter_notif_type & iterator->subscription_mask) ? BT_TRUE : BT_FALSE;
 686 }
 687
 688 static
 689 void add_action_push_notif(struct bt_notification_iterator *iterator,
 690                 struct bt_notification *notif)
 691 {
 692         struct action action = {
 693                 .type = ACTION_TYPE_PUSH_NOTIF,
 694         };
 695
 696         assert(notif);
 697
 698         if (!is_subscribed_to_notification_type(iterator, notif->type)) {
 699                 return;
 700         }
 701
 702         action.payload.push_notif.notif = bt_get(notif);
 703         add_action(iterator, &action);
 704 }
 705
 706 static
 707 int add_action_push_notif_stream_begin(
 708                 struct bt_notification_iterator *iterator,
 709                 struct bt_ctf_stream *stream)
 710 {
 711         int ret = 0;
 712         struct bt_notification *stream_begin_notif = NULL;
 713
 714         if (!is_subscribed_to_notification_type(iterator,
 715                         BT_NOTIFICATION_TYPE_STREAM_BEGIN)) {
 716                 goto end;
 717         }
 718
 719         assert(stream);
 720         stream_begin_notif = bt_notification_stream_begin_create(stream);
 721         if (!stream_begin_notif) {
 722                 goto error;
 723         }
 724
 725         add_action_push_notif(iterator, stream_begin_notif);
 726         goto end;
 727
 728 error:
 729         ret = -1;
 730
 731 end:
 732         bt_put(stream_begin_notif);
 733         return ret;
 734 }
 735
 736 static
 737 int add_action_push_notif_stream_end(
 738                 struct bt_notification_iterator *iterator,
 739                 struct bt_ctf_stream *stream)
 740 {
 741         int ret = 0;
 742         struct bt_notification *stream_end_notif = NULL;
 743
 744         if (!is_subscribed_to_notification_type(iterator,
 745                         BT_NOTIFICATION_TYPE_STREAM_END)) {
 746                 goto end;
 747         }
 748
 749         assert(stream);
 750         stream_end_notif = bt_notification_stream_end_create(stream);
 751         if (!stream_end_notif) {
 752                 goto error;
 753         }
 754
 755         add_action_push_notif(iterator, stream_end_notif);
 756         goto end;
 757
 758 error:
 759         ret = -1;
 760
 761 end:
 762         bt_put(stream_end_notif);
 763         return ret;
 764 }
 765
 766 static
 767 int add_action_push_notif_packet_begin(
 768                 struct bt_notification_iterator *iterator,
 769                 struct bt_ctf_packet *packet)
 770 {
 771         int ret = 0;
 772         struct bt_notification *packet_begin_notif = NULL;
 773
 774         if (!is_subscribed_to_notification_type(iterator,
 775                         BT_NOTIFICATION_TYPE_PACKET_BEGIN)) {
 776                 goto end;
 777         }
 778
 779         assert(packet);
 780         packet_begin_notif = bt_notification_packet_begin_create(packet);
 781         if (!packet_begin_notif) {
 782                 goto error;
 783         }
 784
 785         add_action_push_notif(iterator, packet_begin_notif);
 786         goto end;
 787
 788 error:
 789         ret = -1;
 790
 791 end:
 792         bt_put(packet_begin_notif);
 793         return ret;
 794 }
 795
 796 static
 797 int add_action_push_notif_packet_end(
 798                 struct bt_notification_iterator *iterator,
 799                 struct bt_ctf_packet *packet)
 800 {
 801         int ret = 0;
 802         struct bt_notification *packet_end_notif = NULL;
 803
 804         if (!is_subscribed_to_notification_type(iterator,
 805                         BT_NOTIFICATION_TYPE_PACKET_END)) {
 806                 goto end;
 807         }
 808
 809         assert(packet);
 810         packet_end_notif = bt_notification_packet_end_create(packet);
 811         if (!packet_end_notif) {
 812                 goto error;
 813         }
 814
 815         add_action_push_notif(iterator, packet_end_notif);
 816         goto end;
 817
 818 error:
 819         ret = -1;
 820
 821 end:
 822         bt_put(packet_end_notif);
 823         return ret;
 824 }
 825
 826 static
 827 void add_action_set_stream_state_is_ended(
 828                 struct bt_notification_iterator *iterator,
 829                 struct stream_state *stream_state)
 830 {
 831         struct action action = {
 832                 .type = ACTION_TYPE_SET_STREAM_STATE_IS_ENDED,
 833                 .payload.set_stream_state_is_ended = {
 834                         .stream_state = stream_state,
 835                 },
 836         };
 837
 838         assert(stream_state);
 839         add_action(iterator, &action);
 840 }
 841
 842 static
 843 void add_action_set_stream_state_cur_packet(
 844                 struct bt_notification_iterator *iterator,
 845                 struct stream_state *stream_state,
 846                 struct bt_ctf_packet *packet)
 847 {
 848         struct action action = {
 849                 .type = ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET,
 850                 .payload.set_stream_state_cur_packet = {
 851                         .stream_state = stream_state,
 852                         .packet = bt_get(packet),
 853                 },
 854         };
 855
 856         assert(stream_state);
 857         add_action(iterator, &action);
 858 }
 859
 860 static
 861 int ensure_stream_state_exists(struct bt_notification_iterator *iterator,
 862                 struct bt_notification *stream_begin_notif,
 863                 struct bt_ctf_stream *notif_stream,
 864                 struct stream_state **stream_state)
 865 {
 866         int ret = 0;
 867
 868         if (!notif_stream) {
 869                 /*
 870                  * The notification does not reference any stream: no
 871                  * need to get or create a stream state.
 872                  */
 873                 goto end;
 874         }
 875
 876         *stream_state = g_hash_table_lookup(iterator->stream_states,
 877                 notif_stream);
 878         if (!*stream_state) {
 879                 /*
 880                  * This iterator did not bump into this stream yet:
 881                  * create a stream state and a "stream begin"
 882                  * notification.
 883                  */
 884                 struct action action = {
 885                         .type = ACTION_TYPE_ADD_STREAM_STATE,
 886                         .payload.add_stream_state = {
 887                                 .stream = bt_get(notif_stream),
 888                                 .stream_state = NULL,
 889                         },
 890                 };
 891
 892                 *stream_state = create_stream_state(notif_stream);
 893                 if (!stream_state) {
 894                         goto error;
 895                 }
 896
 897                 action.payload.add_stream_state.stream_state =
 898                         *stream_state;
 899                 add_action(iterator, &action);
 900
 901                 if (stream_begin_notif) {
 902                         add_action_push_notif(iterator, stream_begin_notif);
 903                 } else {
 904                         ret = add_action_push_notif_stream_begin(iterator,
 905                                 notif_stream);
 906                         if (ret) {
 907                                 goto error;
 908                         }
 909                 }
 910         }
 911
 912         goto end;
 913
 914 error:
 915         destroy_stream_state(*stream_state);
 916         ret = -1;
 917
 918 end:
 919         return ret;
 920 }
 921
 922 static
 923 int handle_packet_switch(struct bt_notification_iterator *iterator,
 924                 struct bt_notification *packet_begin_notif,
 925                 struct bt_ctf_packet *new_packet,
 926                 struct stream_state *stream_state)
 927 {
 928         int ret = 0;
 929
 930         if (stream_state->cur_packet == new_packet) {
 931                 goto end;
 932         }
 933
 934         if (stream_state->cur_packet) {
 935                 /* End of the current packet */
 936                 ret = add_action_push_notif_packet_end(iterator,
 937                         stream_state->cur_packet);
 938                 if (ret) {
 939                         goto error;
 940                 }
 941         }
 942
 943         /* Beginning of the new packet */
 944         if (packet_begin_notif) {
 945                 add_action_push_notif(iterator, packet_begin_notif);
 946         } else if (new_packet) {
 947                 ret = add_action_push_notif_packet_begin(iterator,
 948                         new_packet);
 949                 if (ret) {
 950                         goto error;
 951                 }
 952         }
 953
 954         add_action_set_stream_state_cur_packet(iterator, stream_state,
 955                 new_packet);
 956         goto end;
 957
 958 error:
 959         ret = -1;
 960
 961 end:
 962         return ret;
 963 }
 964
 965 static
 966 int handle_notif_stream_begin(
 967                 struct bt_notification_iterator *iterator,
 968                 struct bt_notification *notif,
 969                 struct bt_ctf_stream *notif_stream)
 970 {
 971         int ret = 0;
 972         struct stream_state *stream_state;
 973
 974         assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_BEGIN);
 975         assert(notif_stream);
 976         ret = ensure_stream_state_exists(iterator, notif, notif_stream,
 977                 &stream_state);
 978         if (ret) {
 979                 goto error;
 980         }
 981
 982         goto end;
 983
 984 error:
 985         ret = -1;
 986
 987 end:
 988         return ret;
 989 }
 990
 991 static
 992 int handle_notif_stream_end(
 993                 struct bt_notification_iterator *iterator,
 994                 struct bt_notification *notif,
 995                 struct bt_ctf_stream *notif_stream)
 996 {
 997         int ret = 0;
 998         struct stream_state *stream_state;
 999
1000         assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_END);
1001         assert(notif_stream);
1002         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1003                 &stream_state);
1004         if (ret) {
1005                 goto error;
1006         }
1007
1008         ret = handle_packet_switch(iterator, NULL, NULL, stream_state);
1009         if (ret) {
1010                 goto error;
1011         }
1012
1013         add_action_push_notif(iterator, notif);
1014         add_action_set_stream_state_is_ended(iterator, stream_state);
1015         goto end;
1016
1017 error:
1018         ret = -1;
1019
1020 end:
1021         return ret;
1022 }
1023
1024 static
1025 int handle_notif_packet_begin(
1026                 struct bt_notification_iterator *iterator,
1027                 struct bt_notification *notif,
1028                 struct bt_ctf_stream *notif_stream,
1029                 struct bt_ctf_packet *notif_packet)
1030 {
1031         int ret = 0;
1032         struct stream_state *stream_state;
1033
1034         assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_BEGIN);
1035         assert(notif_packet);
1036         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1037                 &stream_state);
1038         if (ret) {
1039                 goto error;
1040         }
1041
1042         ret = handle_packet_switch(iterator, notif, notif_packet, stream_state);
1043         if (ret) {
1044                 goto error;
1045         }
1046
1047         goto end;
1048
1049 error:
1050         ret = -1;
1051
1052 end:
1053         return ret;
1054 }
1055
1056 static
1057 int handle_notif_packet_end(
1058                 struct bt_notification_iterator *iterator,
1059                 struct bt_notification *notif,
1060                 struct bt_ctf_stream *notif_stream,
1061                 struct bt_ctf_packet *notif_packet)
1062 {
1063         int ret = 0;
1064         struct stream_state *stream_state;
1065
1066         assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_END);
1067         assert(notif_packet);
1068         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1069                 &stream_state);
1070         if (ret) {
1071                 goto error;
1072         }
1073
1074         ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state);
1075         if (ret) {
1076                 goto error;
1077         }
1078
1079         /* End of the current packet */
1080         add_action_push_notif(iterator, notif);
1081         add_action_set_stream_state_cur_packet(iterator, stream_state, NULL);
1082         goto end;
1083
1084 error:
1085         ret = -1;
1086
1087 end:
1088         return ret;
1089 }
1090
1091 static
1092 int handle_notif_event(
1093                 struct bt_notification_iterator *iterator,
1094                 struct bt_notification *notif,
1095                 struct bt_ctf_stream *notif_stream,
1096                 struct bt_ctf_packet *notif_packet)
1097 {
1098         int ret = 0;
1099         struct stream_state *stream_state;
1100
1101         assert(notif->type == BT_NOTIFICATION_TYPE_EVENT);
1102         assert(notif_packet);
1103         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1104                 &stream_state);
1105         if (ret) {
1106                 goto error;
1107         }
1108
1109         ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state);
1110         if (ret) {
1111                 goto error;
1112         }
1113
1114         add_action_push_notif(iterator, notif);
1115         goto end;
1116
1117 error:
1118         ret = -1;
1119
1120 end:
1121         return ret;
1122 }
1123
1124 static
1125 int enqueue_notification_and_automatic(
1126                 struct bt_notification_iterator *iterator,
1127                 struct bt_notification *notif)
1128 {
1129         int ret = 0;
1130         struct bt_ctf_event *notif_event = NULL;
1131         struct bt_ctf_stream *notif_stream = NULL;
1132         struct bt_ctf_packet *notif_packet = NULL;
1133
1134         assert(notif);
1135
1136         // TODO: Skip most of this if the iterator is only subscribed
1137         //       to event/inactivity notifications.
1138
1139         /* Get the stream and packet referred by the notification */
1140         switch (notif->type) {
1141         case BT_NOTIFICATION_TYPE_EVENT:
1142                 notif_event = bt_notification_event_borrow_event(notif);
1143                 assert(notif_event);
1144                 notif_packet = bt_ctf_event_borrow_packet(notif_event);
1145                 assert(notif_packet);
1146                 break;
1147         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
1148                 notif_stream =
1149                         bt_notification_stream_begin_borrow_stream(notif);
1150                 assert(notif_stream);
1151                 break;
1152         case BT_NOTIFICATION_TYPE_STREAM_END:
1153                 notif_stream = bt_notification_stream_end_borrow_stream(notif);
1154                 assert(notif_stream);
1155                 break;
1156         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
1157                 notif_packet =
1158                         bt_notification_packet_begin_borrow_packet(notif);
1159                 assert(notif_packet);
1160                 break;
1161         case BT_NOTIFICATION_TYPE_PACKET_END:
1162                 notif_packet = bt_notification_packet_end_borrow_packet(notif);
1163                 assert(notif_packet);
1164                 break;
1165         case BT_NOTIFICATION_TYPE_INACTIVITY:
1166                 /* Always valid */
1167                 goto handle_notif;
1168         default:
1169                 /*
1170                  * Invalid type of notification. Only the notification
1171                  * types above are allowed to be returned by a user
1172                  * component.
1173                  */
1174                 goto error;
1175         }
1176
1177         if (notif_packet) {
1178                 notif_stream = bt_ctf_packet_borrow_stream(notif_packet);
1179                 assert(notif_stream);
1180         }
1181
1182         if (!notif_stream) {
1183                 /*
1184                  * The notification has no reference to a stream: it
1185                  * cannot cause the creation of automatic notifications.
1186                  */
1187                 goto end;
1188         }
1189
1190         if (!validate_notification(iterator, notif, notif_stream,
1191                         notif_packet)) {
1192                 goto error;
1193         }
1194
1195 handle_notif:
1196         switch (notif->type) {
1197         case BT_NOTIFICATION_TYPE_EVENT:
1198                 ret = handle_notif_event(iterator, notif, notif_stream,
1199                         notif_packet);
1200                 break;
1201         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
1202                 ret = handle_notif_stream_begin(iterator, notif, notif_stream);
1203                 break;
1204         case BT_NOTIFICATION_TYPE_STREAM_END:
1205                 ret = handle_notif_stream_end(iterator, notif, notif_stream);
1206                 break;
1207         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
1208                 ret = handle_notif_packet_begin(iterator, notif, notif_stream,
1209                         notif_packet);
1210                 break;
1211         case BT_NOTIFICATION_TYPE_PACKET_END:
1212                 ret = handle_notif_packet_end(iterator, notif, notif_stream,
1213                         notif_packet);
1214                 break;
1215         case BT_NOTIFICATION_TYPE_INACTIVITY:
1216                 add_action_push_notif(iterator, notif);
1217                 break;
1218         default:
1219                 break;
1220         }
1221
1222         if (ret) {
1223                 goto error;
1224         }
1225
1226         apply_actions(iterator);
1227         goto end;
1228
1229 error:
1230         ret = -1;
1231
1232 end:
1233         return ret;
1234 }
1235
1236 static
1237 int handle_end(struct bt_notification_iterator *iterator)
1238 {
1239         GHashTableIter stream_state_iter;
1240         gpointer stream_gptr, stream_state_gptr;
1241         int ret = 0;
1242
1243         /*
1244          * Emit a "stream end" notification for each non-ended stream
1245          * known by this iterator and mark them as ended.
1246          */
1247         g_hash_table_iter_init(&stream_state_iter, iterator->stream_states);
1248
1249         while (g_hash_table_iter_next(&stream_state_iter, &stream_gptr,
1250                         &stream_state_gptr)) {
1251                 struct stream_state *stream_state = stream_state_gptr;
1252
1253                 assert(stream_state_gptr);
1254
1255                 if (stream_state->is_ended) {
1256                         continue;
1257                 }
1258
1259                 ret = handle_packet_switch(iterator, NULL, NULL, stream_state);
1260                 if (ret) {
1261                         goto error;
1262                 }
1263
1264                 ret = add_action_push_notif_stream_end(iterator, stream_gptr);
1265                 if (ret) {
1266                         goto error;
1267                 }
1268
1269                 add_action_set_stream_state_is_ended(iterator, stream_state);
1270         }
1271
1272         apply_actions(iterator);
1273         goto end;
1274
1275 error:
1276         ret = -1;
1277
1278 end:
1279         return ret;
1280 }
1281
1282 static
1283 enum bt_notification_iterator_status ensure_queue_has_notifications(
1284                 struct bt_notification_iterator *iterator)
1285 {
1286         struct bt_private_notification_iterator *priv_iterator =
1287                 bt_private_notification_iterator_from_notification_iterator(iterator);
1288         bt_component_class_notification_iterator_next_method next_method = NULL;
1289         struct bt_notification_iterator_next_return next_return = {
1290                 .status = BT_NOTIFICATION_ITERATOR_STATUS_OK,
1291                 .notification = NULL,
1292         };
1293         enum bt_notification_iterator_status status =
1294                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
1295         int ret;
1296
1297         assert(iterator);
1298
1299         if (iterator->queue->length > 0) {
1300                 /* We already have enough */
1301                 goto end;
1302         }
1303
1304         switch (iterator->state) {
1305         case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED:
1306                 status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
1307                 goto end;
1308         case BT_NOTIFICATION_ITERATOR_STATE_ENDED:
1309                 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
1310                 goto end;
1311         default:
1312                 break;
1313         }
1314
1315         assert(iterator->upstream_component);
1316         assert(iterator->upstream_component->class);
1317
1318         /* Pick the appropriate "next" method */
1319         switch (iterator->upstream_component->class->type) {
1320         case BT_COMPONENT_CLASS_TYPE_SOURCE:
1321         {
1322                 struct bt_component_class_source *source_class =
1323                         container_of(iterator->upstream_component->class,
1324                                 struct bt_component_class_source, parent);
1325
1326                 assert(source_class->methods.iterator.next);
1327                 next_method = source_class->methods.iterator.next;
1328                 break;
1329         }
1330         case BT_COMPONENT_CLASS_TYPE_FILTER:
1331         {
1332                 struct bt_component_class_filter *filter_class =
1333                         container_of(iterator->upstream_component->class,
1334                                 struct bt_component_class_filter, parent);
1335
1336                 assert(filter_class->methods.iterator.next);
1337                 next_method = filter_class->methods.iterator.next;
1338                 break;
1339         }
1340         default:
1341                 assert(BT_FALSE);
1342                 break;
1343         }
1344
1345         /*
1346          * Call the user's "next" method to get the next notification
1347          * and status.
1348          */
1349         assert(next_method);
1350
1351         while (iterator->queue->length == 0) {
1352                 next_return = next_method(priv_iterator);
1353                 if (next_return.status < 0) {
1354                         status = next_return.status;
1355                         goto end;
1356                 }
1357
1358                 switch (next_return.status) {
1359                 case BT_NOTIFICATION_ITERATOR_STATUS_END:
1360                         ret = handle_end(iterator);
1361                         if (ret) {
1362                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
1363                                 goto end;
1364                         }
1365
1366                         if (iterator->state == BT_NOTIFICATION_ITERATOR_STATE_FINALIZED) {
1367                                 iterator->state =
1368                                         BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED;
1369
1370                                 if (iterator->queue->length == 0) {
1371                                         status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
1372                                 }
1373                         } else {
1374                                 iterator->state =
1375                                         BT_NOTIFICATION_ITERATOR_STATE_ENDED;
1376
1377                                 if (iterator->queue->length == 0) {
1378                                         status = BT_NOTIFICATION_ITERATOR_STATUS_END;
1379                                 }
1380                         }
1381                         goto end;
1382                 case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
1383                         status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN;
1384                         goto end;
1385                 case BT_NOTIFICATION_ITERATOR_STATUS_OK:
1386                         if (!next_return.notification) {
1387                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
1388                                 goto end;
1389                         }
1390
1391                         /*
1392                          * We know the notification is valid. Before we
1393                          * push it to the head of the queue, push the
1394                          * appropriate automatic notifications if any.
1395                          */
1396                         ret = enqueue_notification_and_automatic(iterator,
1397                                 next_return.notification);
1398                         BT_PUT(next_return.notification);
1399                         if (ret) {
1400                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
1401                                 goto end;
1402                         }
1403                         break;
1404                 default:
1405                         /* Unknown non-error status */
1406                         assert(BT_FALSE);
1407                 }
1408         }
1409
1410 end:
1411         return status;
1412 }
1413
1414 enum bt_notification_iterator_status
1415 bt_notification_iterator_next(struct bt_notification_iterator *iterator)
1416 {
1417         enum bt_notification_iterator_status status;
1418
1419         if (!iterator) {
1420                 status = BT_NOTIFICATION_ITERATOR_STATUS_INVALID;
1421                 goto end;
1422         }
1423
1424         /*
1425          * Make sure that the iterator's queue contains at least one
1426          * notification.
1427          */
1428         status = ensure_queue_has_notifications(iterator);
1429         if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
1430                 goto end;
1431         }
1432
1433         /*
1434          * Move the notification at the tail of the queue to the
1435          * iterator's current notification.
1436          */
1437         assert(iterator->queue->length > 0);
1438         bt_put(iterator->current_notification);
1439         iterator->current_notification = g_queue_pop_tail(iterator->queue);
1440         assert(iterator->current_notification);
1441
1442 end:
1443         return status;
1444 }
1445
1446 struct bt_component *bt_notification_iterator_get_component(
1447                 struct bt_notification_iterator *iterator)
1448 {
1449         return bt_get(iterator->upstream_component);
1450 }
1451
1452 struct bt_private_component *
1453 bt_private_notification_iterator_get_private_component(
1454                 struct bt_private_notification_iterator *private_iterator)
1455 {
1456         return bt_private_component_from_component(
1457                 bt_notification_iterator_get_component(
1458                         bt_notification_iterator_from_private(private_iterator)));
1459 }
1460
1461 enum bt_notification_iterator_status bt_notification_iterator_seek_time(
1462                 struct bt_notification_iterator *iterator,
1463                 enum bt_notification_iterator_seek_origin seek_origin,
1464                 int64_t time)
1465 {
1466         enum bt_notification_iterator_status ret =
1467                         BT_NOTIFICATION_ITERATOR_STATUS_UNSUPPORTED;
1468         return ret;
1469 }