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 #define BT_LOG_TAG "NOTIF-ITER"
  29 #include <babeltrace/lib-logging-internal.h>
  30
  31 #include <babeltrace/compiler-internal.h>
  32 #include <babeltrace/ref.h>
  33 #include <babeltrace/ctf-ir/fields.h>
  34 #include <babeltrace/ctf-ir/field-types.h>
  35 #include <babeltrace/ctf-ir/field-types-internal.h>
  36 #include <babeltrace/ctf-ir/event-internal.h>
  37 #include <babeltrace/ctf-ir/packet-internal.h>
  38 #include <babeltrace/ctf-ir/stream-internal.h>
  39 #include <babeltrace/graph/connection.h>
  40 #include <babeltrace/graph/connection-internal.h>
  41 #include <babeltrace/graph/component.h>
  42 #include <babeltrace/graph/component-source-internal.h>
  43 #include <babeltrace/graph/component-class-internal.h>
  44 #include <babeltrace/graph/notification.h>
  45 #include <babeltrace/graph/notification-iterator.h>
  46 #include <babeltrace/graph/notification-iterator-internal.h>
  47 #include <babeltrace/graph/notification-internal.h>
  48 #include <babeltrace/graph/notification-event.h>
  49 #include <babeltrace/graph/notification-event-internal.h>
  50 #include <babeltrace/graph/notification-packet.h>
  51 #include <babeltrace/graph/notification-packet-internal.h>
  52 #include <babeltrace/graph/notification-stream.h>
  53 #include <babeltrace/graph/notification-stream-internal.h>
  54 #include <babeltrace/graph/notification-discarded-elements-internal.h>
  55 #include <babeltrace/graph/port.h>
  56 #include <babeltrace/types.h>
  57 #include <stdint.h>
  58 #include <inttypes.h>
  59 #include <stdlib.h>
  60
  61 struct discarded_elements_state {
  62         struct bt_ctf_clock_value *cur_begin;
  63         uint64_t cur_count;
  64 };
  65
  66 struct stream_state {
  67         struct bt_ctf_stream *stream; /* owned by this */
  68         struct bt_ctf_packet *cur_packet; /* owned by this */
  69         struct discarded_elements_state discarded_packets_state;
  70         struct discarded_elements_state discarded_events_state;
  71         bt_bool is_ended;
  72 };
  73
  74 enum action_type {
  75         ACTION_TYPE_PUSH_NOTIF,
  76         ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM,
  77         ACTION_TYPE_ADD_STREAM_STATE,
  78         ACTION_TYPE_SET_STREAM_STATE_IS_ENDED,
  79         ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET,
  80         ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS,
  81         ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS,
  82 };
  83
  84 struct action {
  85         enum action_type type;
  86         union {
  87                 /* ACTION_TYPE_PUSH_NOTIF */
  88                 struct {
  89                         struct bt_notification *notif; /* owned by this */
  90                 } push_notif;
  91
  92                 /* ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM */
  93                 struct {
  94                         struct bt_ctf_stream *stream; /* owned by this */
  95                         struct bt_component *component; /* owned by this */
  96                         struct bt_port *port; /* owned by this */
  97                 } map_port_to_comp_in_stream;
  98
  99                 /* ACTION_TYPE_ADD_STREAM_STATE */
 100                 struct {
 101                         struct bt_ctf_stream *stream; /* owned by this */
 102                         struct stream_state *stream_state; /* owned by this */
 103                 } add_stream_state;
 104
 105                 /* ACTION_TYPE_SET_STREAM_STATE_IS_ENDED */
 106                 struct {
 107                         struct stream_state *stream_state; /* weak */
 108                 } set_stream_state_is_ended;
 109
 110                 /* ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET */
 111                 struct {
 112                         struct stream_state *stream_state; /* weak */
 113                         struct bt_ctf_packet *packet; /* owned by this */
 114                 } set_stream_state_cur_packet;
 115
 116                 /* ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS */
 117                 /* ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS */
 118                 struct {
 119                         struct stream_state *stream_state; /* weak */
 120                         struct bt_ctf_clock_value *cur_begin; /* owned by this */
 121                         uint64_t cur_count;
 122                 } update_stream_state_discarded_elements;
 123         } payload;
 124 };
 125
 126 static
 127 void stream_destroy_listener(struct bt_ctf_stream *stream, void *data)
 128 {
 129         struct bt_notification_iterator *iterator = data;
 130
 131         /* Remove associated stream state */
 132         g_hash_table_remove(iterator->stream_states, stream);
 133 }
 134
 135 static
 136 void destroy_stream_state(struct stream_state *stream_state)
 137 {
 138         if (!stream_state) {
 139                 return;
 140         }
 141
 142         BT_LOGV("Destroying stream state: stream-state-addr=%p", stream_state);
 143         BT_LOGV_STR("Putting stream state's current packet.");
 144         bt_put(stream_state->cur_packet);
 145         BT_LOGV_STR("Putting stream state's stream.");
 146         bt_put(stream_state->stream);
 147         bt_put(stream_state->discarded_packets_state.cur_begin);
 148         bt_put(stream_state->discarded_events_state.cur_begin);
 149         g_free(stream_state);
 150 }
 151
 152 static
 153 void destroy_action(struct action *action)
 154 {
 155         assert(action);
 156
 157         switch (action->type) {
 158         case ACTION_TYPE_PUSH_NOTIF:
 159                 BT_PUT(action->payload.push_notif.notif);
 160                 break;
 161         case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM:
 162                 BT_PUT(action->payload.map_port_to_comp_in_stream.stream);
 163                 BT_PUT(action->payload.map_port_to_comp_in_stream.component);
 164                 BT_PUT(action->payload.map_port_to_comp_in_stream.port);
 165                 break;
 166         case ACTION_TYPE_ADD_STREAM_STATE:
 167                 BT_PUT(action->payload.add_stream_state.stream);
 168                 destroy_stream_state(
 169                         action->payload.add_stream_state.stream_state);
 170                 action->payload.add_stream_state.stream_state = NULL;
 171                 break;
 172         case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET:
 173                 BT_PUT(action->payload.set_stream_state_cur_packet.packet);
 174                 break;
 175         case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED:
 176                 break;
 177         case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS:
 178         case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS:
 179                 BT_PUT(action->payload.update_stream_state_discarded_elements.cur_begin);
 180                 break;
 181         default:
 182                 BT_LOGF("Unexpected action's type: type=%d", action->type);
 183                 abort();
 184         }
 185 }
 186
 187 static
 188 void add_action(struct bt_notification_iterator *iterator,
 189                 struct action *action)
 190 {
 191         g_array_append_val(iterator->actions, *action);
 192 }
 193
 194 static
 195 void clear_actions(struct bt_notification_iterator *iterator)
 196 {
 197         size_t i;
 198
 199         for (i = 0; i < iterator->actions->len; i++) {
 200                 struct action *action = &g_array_index(iterator->actions,
 201                         struct action, i);
 202
 203                 destroy_action(action);
 204         }
 205
 206         g_array_set_size(iterator->actions, 0);
 207 }
 208
 209 static inline
 210 const char *action_type_string(enum action_type type)
 211 {
 212         switch (type) {
 213         case ACTION_TYPE_PUSH_NOTIF:
 214                 return "ACTION_TYPE_PUSH_NOTIF";
 215         case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM:
 216                 return "ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM";
 217         case ACTION_TYPE_ADD_STREAM_STATE:
 218                 return "ACTION_TYPE_ADD_STREAM_STATE";
 219         case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED:
 220                 return "ACTION_TYPE_SET_STREAM_STATE_IS_ENDED";
 221         case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET:
 222                 return "ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET";
 223         case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS:
 224                 return "ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS";
 225         case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS:
 226                 return "ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS";
 227         default:
 228                 return "(unknown)";
 229         }
 230 }
 231
 232 static
 233 void apply_actions(struct bt_notification_iterator *iterator)
 234 {
 235         size_t i;
 236
 237         BT_LOGV("Applying notification's iterator current actions: "
 238                 "count=%u", iterator->actions->len);
 239
 240         for (i = 0; i < iterator->actions->len; i++) {
 241                 struct action *action = &g_array_index(iterator->actions,
 242                         struct action, i);
 243
 244                 BT_LOGV("Applying action: index=%zu, type=%s",
 245                         i, action_type_string(action->type));
 246
 247                 switch (action->type) {
 248                 case ACTION_TYPE_PUSH_NOTIF:
 249                         /* Move notification to queue */
 250                         g_queue_push_head(iterator->queue,
 251                                 action->payload.push_notif.notif);
 252                         bt_notification_freeze(
 253                                 action->payload.push_notif.notif);
 254                         action->payload.push_notif.notif = NULL;
 255                         break;
 256                 case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM:
 257                         bt_ctf_stream_map_component_to_port(
 258                                 action->payload.map_port_to_comp_in_stream.stream,
 259                                 action->payload.map_port_to_comp_in_stream.component,
 260                                 action->payload.map_port_to_comp_in_stream.port);
 261                         break;
 262                 case ACTION_TYPE_ADD_STREAM_STATE:
 263                         /* Move stream state to hash table */
 264                         g_hash_table_insert(iterator->stream_states,
 265                                 action->payload.add_stream_state.stream,
 266                                 action->payload.add_stream_state.stream_state);
 267
 268                         action->payload.add_stream_state.stream_state = NULL;
 269                         break;
 270                 case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED:
 271                         /*
 272                          * We know that this stream is ended. We need to
 273                          * remember this as long as the stream exists to
 274                          * enforce that the same stream does not end
 275                          * twice.
 276                          *
 277                          * Here we add a destroy listener to the stream
 278                          * which we put after (becomes weak as the hash
 279                          * table key). If we were the last object to own
 280                          * this stream, the destroy listener is called
 281                          * when we call bt_put() which removes this
 282                          * stream state completely. This is important
 283                          * because the memory used by this stream object
 284                          * could be reused for another stream, and they
 285                          * must have different states.
 286                          */
 287                         bt_ctf_stream_add_destroy_listener(
 288                                 action->payload.set_stream_state_is_ended.stream_state->stream,
 289                                 stream_destroy_listener, iterator);
 290                         action->payload.set_stream_state_is_ended.stream_state->is_ended = BT_TRUE;
 291                         BT_PUT(action->payload.set_stream_state_is_ended.stream_state->stream);
 292                         break;
 293                 case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET:
 294                         /* Move packet to stream state's current packet */
 295                         BT_MOVE(action->payload.set_stream_state_cur_packet.stream_state->cur_packet,
 296                                 action->payload.set_stream_state_cur_packet.packet);
 297                         break;
 298                 case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS:
 299                 case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS:
 300                 {
 301                         struct discarded_elements_state *state;
 302
 303                         if (action->type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS) {
 304                                 state = &action->payload.update_stream_state_discarded_elements.stream_state->discarded_packets_state;
 305                         } else {
 306                                 state = &action->payload.update_stream_state_discarded_elements.stream_state->discarded_events_state;
 307                         }
 308
 309                         BT_MOVE(state->cur_begin,
 310                                 action->payload.update_stream_state_discarded_elements.cur_begin);
 311                         state->cur_count = action->payload.update_stream_state_discarded_elements.cur_count;
 312                         break;
 313                 }
 314                 default:
 315                         BT_LOGF("Unexpected action's type: type=%d",
 316                                 action->type);
 317                         abort();
 318                 }
 319         }
 320
 321         clear_actions(iterator);
 322 }
 323
 324 static
 325 struct stream_state *create_stream_state(struct bt_ctf_stream *stream)
 326 {
 327         struct stream_state *stream_state = g_new0(struct stream_state, 1);
 328
 329         if (!stream_state) {
 330                 BT_LOGE_STR("Failed to allocate one stream state.");
 331                 goto end;
 332         }
 333
 334         /*
 335          * We keep a reference to the stream until we know it's ended
 336          * because we need to be able to create an automatic "stream
 337          * end" notification when the user's "next" method returns
 338          * BT_NOTIFICATION_ITERATOR_STATUS_END.
 339          *
 340          * We put this reference when the stream is marked as ended.
 341          */
 342         stream_state->stream = bt_get(stream);
 343         BT_LOGV("Created stream state: stream-addr=%p, stream-name=\"%s\", "
 344                 "stream-state-addr=%p",
 345                 stream, bt_ctf_stream_get_name(stream), stream_state);
 346
 347 end:
 348         return stream_state;
 349 }
 350
 351 static
 352 void bt_notification_iterator_destroy(struct bt_object *obj)
 353 {
 354         struct bt_notification_iterator *iterator;
 355
 356         assert(obj);
 357
 358         /*
 359          * The notification iterator's reference count is 0 if we're
 360          * here. Increment it to avoid a double-destroy (possibly
 361          * infinitely recursive). This could happen for example if the
 362          * notification iterator's finalization function does bt_get()
 363          * (or anything that causes bt_get() to be called) on itself
 364          * (ref. count goes from 0 to 1), and then bt_put(): the
 365          * reference count would go from 1 to 0 again and this function
 366          * would be called again.
 367          */
 368         obj->ref_count.count++;
 369         iterator = container_of(obj, struct bt_notification_iterator, base);
 370         BT_LOGD("Destroying notification iterator object: addr=%p",
 371                 iterator);
 372         bt_notification_iterator_finalize(iterator);
 373
 374         if (iterator->queue) {
 375                 struct bt_notification *notif;
 376
 377                 BT_LOGD("Putting notifications in queue.");
 378
 379                 while ((notif = g_queue_pop_tail(iterator->queue))) {
 380                         bt_put(notif);
 381                 }
 382
 383                 g_queue_free(iterator->queue);
 384         }
 385
 386         if (iterator->stream_states) {
 387                 /*
 388                  * Remove our destroy listener from each stream which
 389                  * has a state in this iterator. Otherwise the destroy
 390                  * listener would be called with an invalid/other
 391                  * notification iterator object.
 392                  */
 393                 GHashTableIter ht_iter;
 394                 gpointer stream_gptr, stream_state_gptr;
 395
 396                 g_hash_table_iter_init(&ht_iter, iterator->stream_states);
 397
 398                 while (g_hash_table_iter_next(&ht_iter, &stream_gptr, &stream_state_gptr)) {
 399                         assert(stream_gptr);
 400
 401                         BT_LOGD_STR("Removing stream's destroy listener for notification iterator.");
 402                         bt_ctf_stream_remove_destroy_listener(
 403                                 (void *) stream_gptr, stream_destroy_listener,
 404                                 iterator);
 405                 }
 406
 407                 g_hash_table_destroy(iterator->stream_states);
 408         }
 409
 410         if (iterator->actions) {
 411                 g_array_free(iterator->actions, TRUE);
 412         }
 413
 414         if (iterator->connection) {
 415                 /*
 416                  * Remove ourself from the originating connection so
 417                  * that it does not try to finalize a dangling pointer
 418                  * later.
 419                  */
 420                 bt_connection_remove_iterator(iterator->connection, iterator);
 421         }
 422
 423         BT_LOGD_STR("Putting current notification.");
 424         bt_put(iterator->current_notification);
 425         g_free(iterator);
 426 }
 427
 428 BT_HIDDEN
 429 void bt_notification_iterator_finalize(
 430                 struct bt_notification_iterator *iterator)
 431 {
 432         struct bt_component_class *comp_class = NULL;
 433         bt_component_class_notification_iterator_finalize_method
 434                 finalize_method = NULL;
 435
 436         assert(iterator);
 437
 438         switch (iterator->state) {
 439         case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED:
 440         case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED:
 441                 /* Already finalized */
 442                 BT_LOGD("Not finalizing notification iterator: already finalized: "
 443                         "addr=%p", iterator);
 444                 return;
 445         default:
 446                 break;
 447         }
 448
 449         BT_LOGD("Finalizing notification iterator: addr=%p", iterator);
 450
 451         if (iterator->state == BT_NOTIFICATION_ITERATOR_STATE_ENDED) {
 452                 BT_LOGD("Updating notification iterator's state: "
 453                         "new-state=BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED");
 454                 iterator->state = BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED;
 455         } else {
 456                 BT_LOGD("Updating notification iterator's state: "
 457                         "new-state=BT_NOTIFICATION_ITERATOR_STATE_FINALIZED");
 458                 iterator->state = BT_NOTIFICATION_ITERATOR_STATE_FINALIZED;
 459         }
 460
 461         assert(iterator->upstream_component);
 462         comp_class = iterator->upstream_component->class;
 463
 464         /* Call user-defined destroy method */
 465         switch (comp_class->type) {
 466         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 467         {
 468                 struct bt_component_class_source *source_class;
 469
 470                 source_class = container_of(comp_class, struct bt_component_class_source, parent);
 471                 finalize_method = source_class->methods.iterator.finalize;
 472                 break;
 473         }
 474         case BT_COMPONENT_CLASS_TYPE_FILTER:
 475         {
 476                 struct bt_component_class_filter *filter_class;
 477
 478                 filter_class = container_of(comp_class, struct bt_component_class_filter, parent);
 479                 finalize_method = filter_class->methods.iterator.finalize;
 480                 break;
 481         }
 482         default:
 483                 /* Unreachable */
 484                 abort();
 485         }
 486
 487         if (finalize_method) {
 488                 BT_LOGD("Calling user's finalization method: addr=%p",
 489                         iterator);
 490                 finalize_method(
 491                         bt_private_notification_iterator_from_notification_iterator(iterator));
 492         }
 493
 494         iterator->upstream_component = NULL;
 495         iterator->upstream_port = NULL;
 496         BT_LOGD("Finalized notification iterator: addr=%p", iterator);
 497 }
 498
 499 BT_HIDDEN
 500 void bt_notification_iterator_set_connection(
 501                 struct bt_notification_iterator *iterator,
 502                 struct bt_connection *connection)
 503 {
 504         assert(iterator);
 505         iterator->connection = connection;
 506         BT_LOGV("Set notification iterator's connection: "
 507                 "iter-addr=%p, conn-addr=%p", iterator, connection);
 508 }
 509
 510 static
 511 int create_subscription_mask_from_notification_types(
 512                 struct bt_notification_iterator *iterator,
 513                 const enum bt_notification_type *notif_types)
 514 {
 515         const enum bt_notification_type *notif_type;
 516         int ret = 0;
 517
 518         assert(notif_types);
 519         iterator->subscription_mask = 0;
 520
 521         for (notif_type = notif_types;
 522                         *notif_type != BT_NOTIFICATION_TYPE_SENTINEL;
 523                         notif_type++) {
 524                 switch (*notif_type) {
 525                 case BT_NOTIFICATION_TYPE_ALL:
 526                         iterator->subscription_mask |=
 527                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT |
 528                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY |
 529                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN |
 530                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END |
 531                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN |
 532                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END |
 533                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_EVENTS |
 534                                 BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_PACKETS;
 535                         break;
 536                 case BT_NOTIFICATION_TYPE_EVENT:
 537                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT;
 538                         break;
 539                 case BT_NOTIFICATION_TYPE_INACTIVITY:
 540                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY;
 541                         break;
 542                 case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 543                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN;
 544                         break;
 545                 case BT_NOTIFICATION_TYPE_STREAM_END:
 546                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END;
 547                         break;
 548                 case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 549                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN;
 550                         break;
 551                 case BT_NOTIFICATION_TYPE_PACKET_END:
 552                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END;
 553                         break;
 554                 case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS:
 555                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_EVENTS;
 556                         break;
 557                 case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS:
 558                         iterator->subscription_mask |= BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_PACKETS;
 559                         break;
 560                 default:
 561                         ret = -1;
 562                         goto end;
 563                 }
 564
 565                 BT_LOGV("Added notification type to subscription mask: "
 566                         "type=%s, mask=%x",
 567                         bt_notification_type_string(*notif_type),
 568                         iterator->subscription_mask);
 569         }
 570
 571 end:
 572         return ret;
 573 }
 574
 575 BT_HIDDEN
 576 enum bt_connection_status bt_notification_iterator_create(
 577                 struct bt_component *upstream_comp,
 578                 struct bt_port *upstream_port,
 579                 const enum bt_notification_type *notification_types,
 580                 struct bt_connection *connection,
 581                 struct bt_notification_iterator **user_iterator)
 582 {
 583         enum bt_connection_status status = BT_CONNECTION_STATUS_OK;
 584         enum bt_component_class_type type;
 585         struct bt_notification_iterator *iterator = NULL;
 586
 587         assert(upstream_comp);
 588         assert(upstream_port);
 589         assert(notification_types);
 590         assert(bt_port_is_connected(upstream_port));
 591         assert(user_iterator);
 592         BT_LOGD("Creating notification iterator: "
 593                 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
 594                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 595                 "conn-addr=%p",
 596                 upstream_comp, bt_component_get_name(upstream_comp),
 597                 upstream_port, bt_port_get_name(upstream_port),
 598                 connection);
 599         type = bt_component_get_class_type(upstream_comp);
 600         assert(type == BT_COMPONENT_CLASS_TYPE_SOURCE ||
 601                 type == BT_COMPONENT_CLASS_TYPE_FILTER);
 602         iterator = g_new0(struct bt_notification_iterator, 1);
 603         if (!iterator) {
 604                 BT_LOGE_STR("Failed to allocate one notification iterator.");
 605                 status = BT_CONNECTION_STATUS_NOMEM;
 606                 goto end;
 607         }
 608
 609         bt_object_init(iterator, bt_notification_iterator_destroy);
 610
 611         if (create_subscription_mask_from_notification_types(iterator,
 612                         notification_types)) {
 613                 BT_LOGW_STR("Cannot create subscription mask from notification types.");
 614                 status = BT_CONNECTION_STATUS_INVALID;
 615                 goto end;
 616         }
 617
 618         iterator->stream_states = g_hash_table_new_full(g_direct_hash,
 619                 g_direct_equal, NULL, (GDestroyNotify) destroy_stream_state);
 620         if (!iterator->stream_states) {
 621                 BT_LOGE_STR("Failed to allocate a GHashTable.");
 622                 status = BT_CONNECTION_STATUS_NOMEM;
 623                 goto end;
 624         }
 625
 626         iterator->queue = g_queue_new();
 627         if (!iterator->queue) {
 628                 BT_LOGE_STR("Failed to allocate a GQueue.");
 629                 status = BT_CONNECTION_STATUS_NOMEM;
 630                 goto end;
 631         }
 632
 633         iterator->actions = g_array_new(FALSE, FALSE, sizeof(struct action));
 634         if (!iterator->actions) {
 635                 BT_LOGE_STR("Failed to allocate a GArray.");
 636                 status = BT_CONNECTION_STATUS_NOMEM;
 637                 goto end;
 638         }
 639
 640         iterator->upstream_component = upstream_comp;
 641         iterator->upstream_port = upstream_port;
 642         iterator->connection = connection;
 643         iterator->state = BT_NOTIFICATION_ITERATOR_STATE_ACTIVE;
 644         BT_LOGD("Created notification iterator: "
 645                 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
 646                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 647                 "conn-addr=%p, iter-addr=%p",
 648                 upstream_comp, bt_component_get_name(upstream_comp),
 649                 upstream_port, bt_port_get_name(upstream_port),
 650                 connection, iterator);
 651
 652         /* Move reference to user */
 653         *user_iterator = iterator;
 654         iterator = NULL;
 655
 656 end:
 657         bt_put(iterator);
 658         return status;
 659 }
 660
 661 void *bt_private_notification_iterator_get_user_data(
 662                 struct bt_private_notification_iterator *private_iterator)
 663 {
 664         struct bt_notification_iterator *iterator =
 665                 bt_notification_iterator_from_private(private_iterator);
 666
 667         return iterator ? iterator->user_data : NULL;
 668 }
 669
 670 enum bt_notification_iterator_status
 671 bt_private_notification_iterator_set_user_data(
 672                 struct bt_private_notification_iterator *private_iterator,
 673                 void *data)
 674 {
 675         enum bt_notification_iterator_status ret =
 676                         BT_NOTIFICATION_ITERATOR_STATUS_OK;
 677         struct bt_notification_iterator *iterator =
 678                 bt_notification_iterator_from_private(private_iterator);
 679
 680         if (!iterator) {
 681                 BT_LOGW_STR("Invalid parameter: notification iterator is NULL.");
 682                 ret = BT_NOTIFICATION_ITERATOR_STATUS_INVALID;
 683                 goto end;
 684         }
 685
 686         iterator->user_data = data;
 687         BT_LOGV("Set notification iterator's user data: "
 688                 "iter-addr=%p, user-data-addr=%p", iterator, data);
 689
 690 end:
 691         return ret;
 692 }
 693
 694 struct bt_notification *bt_notification_iterator_get_notification(
 695                 struct bt_notification_iterator *iterator)
 696 {
 697         struct bt_notification *notification = NULL;
 698
 699         if (!iterator) {
 700                 BT_LOGW_STR("Invalid parameter: notification iterator is NULL.");
 701                 goto end;
 702         }
 703
 704         notification = bt_get(iterator->current_notification);
 705
 706 end:
 707         return notification;
 708 }
 709
 710 static
 711 enum bt_notification_iterator_notif_type
 712 bt_notification_iterator_notif_type_from_notif_type(
 713                 enum bt_notification_type notif_type)
 714 {
 715         enum bt_notification_iterator_notif_type iter_notif_type;
 716
 717         switch (notif_type) {
 718         case BT_NOTIFICATION_TYPE_EVENT:
 719                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT;
 720                 break;
 721         case BT_NOTIFICATION_TYPE_INACTIVITY:
 722                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY;
 723                 break;
 724         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 725                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN;
 726                 break;
 727         case BT_NOTIFICATION_TYPE_STREAM_END:
 728                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END;
 729                 break;
 730         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 731                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN;
 732                 break;
 733         case BT_NOTIFICATION_TYPE_PACKET_END:
 734                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END;
 735                 break;
 736         case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS:
 737                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_EVENTS;
 738                 break;
 739         case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS:
 740                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_PACKETS;
 741                 break;
 742         default:
 743                 abort();
 744         }
 745
 746         return iter_notif_type;
 747 }
 748
 749 static
 750 bt_bool validate_notification(struct bt_notification_iterator *iterator,
 751                 struct bt_notification *notif,
 752                 struct bt_ctf_stream *notif_stream,
 753                 struct bt_ctf_packet *notif_packet)
 754 {
 755         bt_bool is_valid = BT_TRUE;
 756         struct stream_state *stream_state;
 757         struct bt_port *stream_comp_cur_port;
 758
 759         assert(notif_stream);
 760         stream_comp_cur_port =
 761                 bt_ctf_stream_port_for_component(notif_stream,
 762                         iterator->upstream_component);
 763         if (!stream_comp_cur_port) {
 764                 /*
 765                  * This is the first time this notification iterator
 766                  * bumps into this stream. Add an action to map the
 767                  * iterator's upstream component to the iterator's
 768                  * upstream port in this stream.
 769                  */
 770                 struct action action = {
 771                         .type = ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM,
 772                         .payload.map_port_to_comp_in_stream = {
 773                                 .stream = bt_get(notif_stream),
 774                                 .component = bt_get(iterator->upstream_component),
 775                                 .port = bt_get(iterator->upstream_port),
 776                         },
 777                 };
 778
 779                 add_action(iterator, &action);
 780         } else {
 781                 if (stream_comp_cur_port != iterator->upstream_port) {
 782                         /*
 783                          * It looks like two different ports of the same
 784                          * component are emitting notifications which
 785                          * have references to the same stream. This is
 786                          * bad: the API guarantees that it can never
 787                          * happen.
 788                          */
 789                         BT_LOGW("Two different ports of the same component are emitting notifications which refer to the same stream: "
 790                                 "stream-addr=%p, stream-name=\"%s\", "
 791                                 "stream-comp-cur-port-addr=%p, "
 792                                 "stream-comp-cur-port-name=%p, "
 793                                 "iter-upstream-port-addr=%p, "
 794                                 "iter-upstream-port-name=%s",
 795                                 notif_stream,
 796                                 bt_ctf_stream_get_name(notif_stream),
 797                                 stream_comp_cur_port,
 798                                 bt_port_get_name(stream_comp_cur_port),
 799                                 iterator->upstream_port,
 800                                 bt_port_get_name(iterator->upstream_port));
 801                         is_valid = BT_FALSE;
 802                         goto end;
 803                 }
 804
 805         }
 806
 807         stream_state = g_hash_table_lookup(iterator->stream_states,
 808                 notif_stream);
 809         if (stream_state) {
 810                 BT_LOGV("Stream state already exists: "
 811                         "stream-addr=%p, stream-name=\"%s\", "
 812                         "stream-state-addr=%p",
 813                         notif_stream,
 814                         bt_ctf_stream_get_name(notif_stream), stream_state);
 815
 816                 if (stream_state->is_ended) {
 817                         /*
 818                          * There's a new notification which has a
 819                          * reference to a stream which, from this
 820                          * iterator's point of view, is ended ("end of
 821                          * stream" notification was returned). This is
 822                          * bad: the API guarantees that it can never
 823                          * happen.
 824                          */
 825                         BT_LOGW("Stream is already ended: "
 826                                 "stream-addr=%p, stream-name=\"%s\"",
 827                                 notif_stream,
 828                                 bt_ctf_stream_get_name(notif_stream));
 829                         is_valid = BT_FALSE;
 830                         goto end;
 831                 }
 832
 833                 switch (notif->type) {
 834                 case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 835                         /*
 836                          * We already have a stream state, which means
 837                          * we already returned a "stream begin"
 838                          * notification: this is an invalid duplicate.
 839                          */
 840                         BT_LOGW("Duplicate stream beginning notification: "
 841                                 "stream-addr=%p, stream-name=\"%s\"",
 842                                 notif_stream,
 843                                 bt_ctf_stream_get_name(notif_stream));
 844                         is_valid = BT_FALSE;
 845                         goto end;
 846                 case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 847                         if (notif_packet == stream_state->cur_packet) {
 848                                 /* Duplicate "packet begin" notification */
 849                                 BT_LOGW("Duplicate stream beginning notification: "
 850                                         "stream-addr=%p, stream-name=\"%s\", "
 851                                         "packet-addr=%p",
 852                                         notif_stream,
 853                                         bt_ctf_stream_get_name(notif_stream),
 854                                         notif_packet);
 855                                 is_valid = BT_FALSE;
 856                                 goto end;
 857                         }
 858                         break;
 859                 default:
 860                         break;
 861                 }
 862         }
 863
 864 end:
 865         return is_valid;
 866 }
 867
 868 static
 869 bt_bool is_subscribed_to_notification_type(struct bt_notification_iterator *iterator,
 870                 enum bt_notification_type notif_type)
 871 {
 872         uint32_t iter_notif_type =
 873                 (uint32_t) bt_notification_iterator_notif_type_from_notif_type(
 874                         notif_type);
 875
 876         return (iter_notif_type & iterator->subscription_mask) ? BT_TRUE : BT_FALSE;
 877 }
 878
 879 static
 880 void add_action_push_notif(struct bt_notification_iterator *iterator,
 881                 struct bt_notification *notif)
 882 {
 883         struct action action = {
 884                 .type = ACTION_TYPE_PUSH_NOTIF,
 885         };
 886
 887         assert(notif);
 888
 889         if (!is_subscribed_to_notification_type(iterator, notif->type)) {
 890                 return;
 891         }
 892
 893         action.payload.push_notif.notif = bt_get(notif);
 894         add_action(iterator, &action);
 895         BT_LOGV("Added \"push notification\" action: notif-addr=%p", notif);
 896 }
 897
 898 static
 899 int add_action_push_notif_stream_begin(
 900                 struct bt_notification_iterator *iterator,
 901                 struct bt_ctf_stream *stream)
 902 {
 903         int ret = 0;
 904         struct bt_notification *stream_begin_notif = NULL;
 905
 906         if (!is_subscribed_to_notification_type(iterator,
 907                         BT_NOTIFICATION_TYPE_STREAM_BEGIN)) {
 908                 BT_LOGV("Not adding \"push stream beginning notification\" action: "
 909                         "notification iterator is not subscribed: addr=%p",
 910                         iterator);
 911                 goto end;
 912         }
 913
 914         assert(stream);
 915         stream_begin_notif = bt_notification_stream_begin_create(stream);
 916         if (!stream_begin_notif) {
 917                 BT_LOGE_STR("Cannot create stream beginning notification.");
 918                 goto error;
 919         }
 920
 921         add_action_push_notif(iterator, stream_begin_notif);
 922         BT_LOGV("Added \"push stream beginning notification\" action: "
 923                 "stream-addr=%p, stream-name=\"%s\"",
 924                 stream, bt_ctf_stream_get_name(stream));
 925         goto end;
 926
 927 error:
 928         ret = -1;
 929
 930 end:
 931         bt_put(stream_begin_notif);
 932         return ret;
 933 }
 934
 935 static
 936 int add_action_push_notif_stream_end(
 937                 struct bt_notification_iterator *iterator,
 938                 struct bt_ctf_stream *stream)
 939 {
 940         int ret = 0;
 941         struct bt_notification *stream_end_notif = NULL;
 942
 943         if (!is_subscribed_to_notification_type(iterator,
 944                         BT_NOTIFICATION_TYPE_STREAM_END)) {
 945                 BT_LOGV("Not adding \"push stream end notification\" action: "
 946                         "notification iterator is not subscribed: addr=%p",
 947                         iterator);
 948                 goto end;
 949         }
 950
 951         assert(stream);
 952         stream_end_notif = bt_notification_stream_end_create(stream);
 953         if (!stream_end_notif) {
 954                 BT_LOGE_STR("Cannot create stream end notification.");
 955                 goto error;
 956         }
 957
 958         add_action_push_notif(iterator, stream_end_notif);
 959         BT_LOGV("Added \"push stream end notification\" action: "
 960                 "stream-addr=%p, stream-name=\"%s\"",
 961                 stream, bt_ctf_stream_get_name(stream));
 962         goto end;
 963
 964 error:
 965         ret = -1;
 966
 967 end:
 968         bt_put(stream_end_notif);
 969         return ret;
 970 }
 971
 972 static
 973 int add_action_push_notif_packet_begin(
 974                 struct bt_notification_iterator *iterator,
 975                 struct bt_ctf_packet *packet)
 976 {
 977         int ret = 0;
 978         struct bt_notification *packet_begin_notif = NULL;
 979
 980         if (!is_subscribed_to_notification_type(iterator,
 981                         BT_NOTIFICATION_TYPE_PACKET_BEGIN)) {
 982                 BT_LOGV("Not adding \"push packet beginning notification\" action: "
 983                         "notification iterator is not subscribed: addr=%p",
 984                         iterator);
 985                 goto end;
 986         }
 987
 988         assert(packet);
 989         packet_begin_notif = bt_notification_packet_begin_create(packet);
 990         if (!packet_begin_notif) {
 991                 BT_LOGE_STR("Cannot create packet beginning notification.");
 992                 goto error;
 993         }
 994
 995         add_action_push_notif(iterator, packet_begin_notif);
 996         BT_LOGV("Added \"push packet beginning notification\" action: "
 997                 "packet-addr=%p", packet);
 998         goto end;
 999
1000 error:
1001         ret = -1;
1002
1003 end:
1004         bt_put(packet_begin_notif);
1005         return ret;
1006 }
1007
1008 static
1009 int add_action_push_notif_packet_end(
1010                 struct bt_notification_iterator *iterator,
1011                 struct bt_ctf_packet *packet)
1012 {
1013         int ret = 0;
1014         struct bt_notification *packet_end_notif = NULL;
1015
1016         if (!is_subscribed_to_notification_type(iterator,
1017                         BT_NOTIFICATION_TYPE_PACKET_END)) {
1018                 BT_LOGV("Not adding \"push packet end notification\" action: "
1019                         "notification iterator is not subscribed: addr=%p",
1020                         iterator);
1021                 goto end;
1022         }
1023
1024         assert(packet);
1025         packet_end_notif = bt_notification_packet_end_create(packet);
1026         if (!packet_end_notif) {
1027                 BT_LOGE_STR("Cannot create packet end notification.");
1028                 goto error;
1029         }
1030
1031         add_action_push_notif(iterator, packet_end_notif);
1032         BT_LOGV("Added \"push packet end notification\" action: "
1033                 "packet-addr=%p", packet);
1034         goto end;
1035
1036 error:
1037         ret = -1;
1038
1039 end:
1040         bt_put(packet_end_notif);
1041         return ret;
1042 }
1043
1044 static
1045 void add_action_set_stream_state_is_ended(
1046                 struct bt_notification_iterator *iterator,
1047                 struct stream_state *stream_state)
1048 {
1049         struct action action = {
1050                 .type = ACTION_TYPE_SET_STREAM_STATE_IS_ENDED,
1051                 .payload.set_stream_state_is_ended = {
1052                         .stream_state = stream_state,
1053                 },
1054         };
1055
1056         assert(stream_state);
1057         add_action(iterator, &action);
1058         BT_LOGV("Added \"set stream state's ended\" action: "
1059                 "stream-state-addr=%p", stream_state);
1060 }
1061
1062 static
1063 void add_action_set_stream_state_cur_packet(
1064                 struct bt_notification_iterator *iterator,
1065                 struct stream_state *stream_state,
1066                 struct bt_ctf_packet *packet)
1067 {
1068         struct action action = {
1069                 .type = ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET,
1070                 .payload.set_stream_state_cur_packet = {
1071                         .stream_state = stream_state,
1072                         .packet = bt_get(packet),
1073                 },
1074         };
1075
1076         assert(stream_state);
1077         add_action(iterator, &action);
1078         BT_LOGV("Added \"set stream state's current packet\" action: "
1079                 "stream-state-addr=%p, packet-addr=%p",
1080                 stream_state, packet);
1081 }
1082
1083 static
1084 void add_action_update_stream_state_discarded_elements(
1085                 struct bt_notification_iterator *iterator,
1086                 enum action_type type,
1087                 struct stream_state *stream_state,
1088                 struct bt_ctf_clock_value *cur_begin,
1089                 uint64_t cur_count)
1090 {
1091         struct action action = {
1092                 .type = type,
1093                 .payload.update_stream_state_discarded_elements = {
1094                         .stream_state = stream_state,
1095                         .cur_begin = bt_get(cur_begin),
1096                         .cur_count = cur_count,
1097                 },
1098         };
1099
1100         assert(stream_state);
1101         assert(type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS ||
1102                         type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS);
1103         add_action(iterator, &action);
1104         if (type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS) {
1105                 BT_LOGV("Added \"update stream state's discarded packets\" action: "
1106                         "stream-state-addr=%p, cur-begin-addr=%p, cur-count=%" PRIu64,
1107                         stream_state, cur_begin, cur_count);
1108         } else if (type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS) {
1109                 BT_LOGV("Added \"update stream state's discarded events\" action: "
1110                         "stream-state-addr=%p, cur-begin-addr=%p, cur-count=%" PRIu64,
1111                         stream_state, cur_begin, cur_count);
1112         }
1113 }
1114
1115 static
1116 int ensure_stream_state_exists(struct bt_notification_iterator *iterator,
1117                 struct bt_notification *stream_begin_notif,
1118                 struct bt_ctf_stream *notif_stream,
1119                 struct stream_state **stream_state)
1120 {
1121         int ret = 0;
1122
1123         if (!notif_stream) {
1124                 /*
1125                  * The notification does not reference any stream: no
1126                  * need to get or create a stream state.
1127                  */
1128                 goto end;
1129         }
1130
1131         *stream_state = g_hash_table_lookup(iterator->stream_states,
1132                 notif_stream);
1133         if (!*stream_state) {
1134                 /*
1135                  * This iterator did not bump into this stream yet:
1136                  * create a stream state and a "stream begin"
1137                  * notification.
1138                  */
1139                 struct action action = {
1140                         .type = ACTION_TYPE_ADD_STREAM_STATE,
1141                         .payload.add_stream_state = {
1142                                 .stream = bt_get(notif_stream),
1143                                 .stream_state = NULL,
1144                         },
1145                 };
1146
1147                 *stream_state = create_stream_state(notif_stream);
1148                 if (!stream_state) {
1149                         BT_LOGE_STR("Cannot create stream state.");
1150                         goto error;
1151                 }
1152
1153                 action.payload.add_stream_state.stream_state =
1154                         *stream_state;
1155                 add_action(iterator, &action);
1156
1157                 if (stream_begin_notif) {
1158                         add_action_push_notif(iterator, stream_begin_notif);
1159                 } else {
1160                         ret = add_action_push_notif_stream_begin(iterator,
1161                                 notif_stream);
1162                         if (ret) {
1163                                 BT_LOGE_STR("Cannot add \"push stream beginning notification\" action.");
1164                                 goto error;
1165                         }
1166                 }
1167         }
1168
1169         goto end;
1170
1171 error:
1172         destroy_stream_state(*stream_state);
1173         ret = -1;
1174
1175 end:
1176         return ret;
1177 }
1178
1179 static
1180 struct bt_ctf_field *get_struct_field_uint(struct bt_ctf_field *struct_field,
1181                 const char *field_name)
1182 {
1183         struct bt_ctf_field *field = NULL;
1184         struct bt_ctf_field_type *ft = NULL;
1185
1186         field = bt_ctf_field_structure_get_field_by_name(struct_field,
1187                 field_name);
1188         if (!field) {
1189                 BT_LOGV_STR("`%s` field does not exist.");
1190                 goto end;
1191         }
1192
1193         if (!bt_ctf_field_is_integer(field)) {
1194                 BT_LOGV("Skipping `%s` field because its type is not an integer field type: "
1195                         "field-addr=%p, ft-addr=%p, ft-id=%s", field_name,
1196                         field, ft, bt_ctf_field_type_id_string(
1197                                 bt_ctf_field_type_get_type_id(ft)));
1198                 BT_PUT(field);
1199                 goto end;
1200         }
1201
1202         ft = bt_ctf_field_get_type(field);
1203         assert(ft);
1204
1205         if (bt_ctf_field_type_integer_is_signed(ft)) {
1206                 BT_LOGV("Skipping `%s` integer field because its type is signed: "
1207                         "field-addr=%p, ft-addr=%p", field_name, field, ft);
1208                 BT_PUT(field);
1209                 goto end;
1210         }
1211
1212 end:
1213         bt_put(ft);
1214         return field;
1215 }
1216
1217 static
1218 uint64_t get_packet_context_events_discarded(struct bt_ctf_packet *packet)
1219 {
1220         struct bt_ctf_field *packet_context = NULL;
1221         struct bt_ctf_field *field = NULL;
1222         uint64_t retval = -1ULL;
1223         int ret;
1224
1225         packet_context = bt_ctf_packet_get_context(packet);
1226         if (!packet_context) {
1227                 goto end;
1228         }
1229
1230         field = get_struct_field_uint(packet_context, "events_discarded");
1231         if (!field) {
1232                 BT_LOGV("`events_discarded` field does not exist in packet's context field: "
1233                         "packet-addr=%p, packet-context-field-addr=%p",
1234                         packet, packet_context);
1235                 goto end;
1236         }
1237
1238         assert(bt_ctf_field_is_integer(field));
1239         ret = bt_ctf_field_unsigned_integer_get_value(field, &retval);
1240         if (ret) {
1241                 BT_LOGV("Cannot get raw value of packet's context field's `events_discarded` integer field: "
1242                         "packet-addr=%p, field-addr=%p",
1243                         packet, field);
1244                 retval = -1ULL;
1245                 goto end;
1246         }
1247
1248 end:
1249         bt_put(packet_context);
1250         bt_put(field);
1251         return retval;
1252 }
1253
1254 static
1255 uint64_t get_packet_header_packet_seq_num(struct bt_ctf_packet *packet)
1256 {
1257         struct bt_ctf_field *packet_header = NULL;
1258         struct bt_ctf_field *field = NULL;
1259         uint64_t retval = -1ULL;
1260         int ret;
1261
1262         packet_header = bt_ctf_packet_get_header(packet);
1263         if (!packet_header) {
1264                 goto end;
1265         }
1266
1267         field = get_struct_field_uint(packet_header, "packet_seq_num");
1268         if (!field) {
1269                 BT_LOGV("`packet_seq_num` field does not exist in packet's header field: "
1270                         "packet-addr=%p, packet-header-field-addr=%p",
1271                         packet, packet_header);
1272                 goto end;
1273         }
1274
1275         assert(bt_ctf_field_is_integer(field));
1276         ret = bt_ctf_field_unsigned_integer_get_value(field, &retval);
1277         if (ret) {
1278                 BT_LOGV("Cannot get raw value of packet's header field's `packet_seq_num` integer field: "
1279                         "packet-addr=%p, field-addr=%p",
1280                         packet, field);
1281                 retval = -1ULL;
1282                 goto end;
1283         }
1284
1285 end:
1286         bt_put(packet_header);
1287         bt_put(field);
1288         return retval;
1289 }
1290
1291 static
1292 int handle_discarded_packets(struct bt_notification_iterator *iterator,
1293                 struct bt_ctf_packet *packet,
1294                 struct bt_ctf_clock_value *ts_begin,
1295                 struct bt_ctf_clock_value *ts_end,
1296                 struct stream_state *stream_state)
1297 {
1298         struct bt_notification *notif = NULL;
1299         uint64_t diff;
1300         uint64_t next_count;
1301         int ret = 0;
1302
1303         next_count = get_packet_header_packet_seq_num(packet);
1304         if (next_count == -1ULL) {
1305                 next_count = stream_state->discarded_packets_state.cur_count;
1306                 goto update_state;
1307         }
1308
1309         if (next_count < stream_state->discarded_packets_state.cur_count) {
1310                 BT_LOGW("Current value of packet's header field's `packet_seq_num` field is lesser than the previous value for the same stream: "
1311                         "not updating the stream state's current value: "
1312                         "packet-addr=%p, prev-count=%" PRIu64 ", "
1313                         "cur-count=%" PRIu64,
1314                         packet, stream_state->discarded_packets_state.cur_count,
1315                         next_count);
1316                 goto end;
1317         }
1318
1319         diff = next_count - stream_state->discarded_packets_state.cur_count;
1320         if (diff > 0) {
1321                 /*
1322                  * Add a discarded packets notification. The packets
1323                  * are considered to be lost betweem the state's last time
1324                  * and the current packet's beginning time.
1325                  */
1326                 notif = bt_notification_discarded_elements_create(
1327                         BT_NOTIFICATION_TYPE_DISCARDED_PACKETS,
1328                         stream_state->stream,
1329                         stream_state->discarded_packets_state.cur_begin,
1330                         ts_begin, diff);
1331                 if (!notif) {
1332                         BT_LOGE_STR("Cannot create discarded packets notification.");
1333                         ret = -1;
1334                         goto end;
1335                 }
1336
1337                 add_action_push_notif(iterator, notif);
1338         }
1339
1340 update_state:
1341         add_action_update_stream_state_discarded_elements(iterator,
1342                 ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS,
1343                 stream_state, ts_end, next_count);
1344
1345 end:
1346         bt_put(notif);
1347         return ret;
1348 }
1349
1350 static
1351 int handle_discarded_events(struct bt_notification_iterator *iterator,
1352                 struct bt_ctf_packet *packet,
1353                 struct bt_ctf_clock_value *ts_begin,
1354                 struct bt_ctf_clock_value *ts_end,
1355                 struct stream_state *stream_state)
1356 {
1357         struct bt_notification *notif = NULL;
1358         uint64_t diff;
1359         uint64_t next_count;
1360         int ret = 0;
1361
1362         next_count = get_packet_context_events_discarded(packet);
1363         if (next_count == -1ULL) {
1364                 next_count = stream_state->discarded_events_state.cur_count;
1365                 goto update_state;
1366         }
1367
1368         if (next_count < stream_state->discarded_events_state.cur_count) {
1369                 BT_LOGW("Current value of packet's context field's `events_discarded` field is lesser than the previous value for the same stream: "
1370                         "not updating the stream state's current value: "
1371                         "packet-addr=%p, prev-count=%" PRIu64 ", "
1372                         "cur-count=%" PRIu64,
1373                         packet, stream_state->discarded_events_state.cur_count,
1374                         next_count);
1375                 goto end;
1376         }
1377
1378         diff = next_count - stream_state->discarded_events_state.cur_count;
1379         if (diff > 0) {
1380                 /*
1381                  * Add a discarded events notification. The events are
1382                  * considered to be lost betweem the state's last time
1383                  * and the current packet's end time.
1384                  */
1385                 notif = bt_notification_discarded_elements_create(
1386                         BT_NOTIFICATION_TYPE_DISCARDED_EVENTS,
1387                         stream_state->stream,
1388                         stream_state->discarded_events_state.cur_begin,
1389                         ts_end, diff);
1390                 if (!notif) {
1391                         BT_LOGE_STR("Cannot create discarded events notification.");
1392                         ret = -1;
1393                         goto end;
1394                 }
1395
1396                 add_action_push_notif(iterator, notif);
1397         }
1398
1399 update_state:
1400         add_action_update_stream_state_discarded_elements(iterator,
1401                 ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS,
1402                 stream_state, ts_end, next_count);
1403
1404 end:
1405         bt_put(notif);
1406         return ret;
1407 }
1408
1409 static
1410 int get_field_clock_value(struct bt_ctf_field *root_field,
1411                 const char *field_name,
1412                 struct bt_ctf_clock_value **user_clock_val)
1413 {
1414         struct bt_ctf_field *field;
1415         struct bt_ctf_field_type *ft = NULL;
1416         struct bt_ctf_clock_class *clock_class = NULL;
1417         struct bt_ctf_clock_value *clock_value = NULL;
1418         uint64_t val;
1419         int ret = 0;
1420
1421         field = get_struct_field_uint(root_field, field_name);
1422         if (!field) {
1423                 /* Not an error: skip this */
1424                 goto end;
1425         }
1426
1427         ft = bt_ctf_field_get_type(field);
1428         assert(ft);
1429         clock_class = bt_ctf_field_type_integer_get_mapped_clock_class(ft);
1430         if (!clock_class) {
1431                 BT_LOGW("Integer field type has no mapped clock class but it's expected to have one: "
1432                         "ft-addr=%p", ft);
1433                 ret = -1;
1434                 goto end;
1435         }
1436
1437         ret = bt_ctf_field_unsigned_integer_get_value(field, &val);
1438         if (ret) {
1439                 BT_LOGW("Cannot get integer field's raw value: "
1440                         "field-addr=%p", field);
1441                 ret = -1;
1442                 goto end;
1443         }
1444
1445         clock_value = bt_ctf_clock_value_create(clock_class, val);
1446         if (!clock_value) {
1447                 BT_LOGE_STR("Cannot create clock value from clock class.");
1448                 ret = -1;
1449                 goto end;
1450         }
1451
1452         /* Move clock value to user */
1453         *user_clock_val = clock_value;
1454         clock_value = NULL;
1455
1456 end:
1457         bt_put(field);
1458         bt_put(ft);
1459         bt_put(clock_class);
1460         bt_put(clock_value);
1461         return ret;
1462 }
1463
1464 static
1465 int get_ts_begin_ts_end_from_packet(struct bt_ctf_packet *packet,
1466                 struct bt_ctf_clock_value **user_ts_begin,
1467                 struct bt_ctf_clock_value **user_ts_end)
1468 {
1469         struct bt_ctf_field *packet_context = NULL;
1470         struct bt_ctf_clock_value *ts_begin = NULL;
1471         struct bt_ctf_clock_value *ts_end = NULL;
1472         int ret = 0;
1473
1474         packet_context = bt_ctf_packet_get_context(packet);
1475         if (!packet_context) {
1476                 goto end;
1477         }
1478
1479         ret = get_field_clock_value(packet_context, "timestamp_begin",
1480                 &ts_begin);
1481         if (ret) {
1482                 BT_LOGW("Cannot create clock value for packet context's `timestamp_begin` field: "
1483                         "packet-addr=%p, packet-context-field-addr=%p",
1484                         packet, packet_context);
1485                 goto end;
1486         }
1487
1488         ret = get_field_clock_value(packet_context, "timestamp_end",
1489                 &ts_end);
1490         if (ret) {
1491                 BT_LOGW("Cannot create clock value for packet context's `timestamp_begin` field: "
1492                         "packet-addr=%p, packet-context-field-addr=%p",
1493                         packet, packet_context);
1494                 goto end;
1495         }
1496
1497         /* Move clock values to user */
1498         *user_ts_begin = ts_begin;
1499         ts_begin = NULL;
1500         *user_ts_end = ts_end;
1501         ts_end = NULL;
1502
1503 end:
1504         bt_put(packet_context);
1505         bt_put(ts_begin);
1506         bt_put(ts_end);
1507         return ret;
1508 }
1509
1510 static
1511 int handle_discarded_elements(struct bt_notification_iterator *iterator,
1512                 struct bt_ctf_packet *packet, struct stream_state *stream_state)
1513 {
1514         struct bt_ctf_clock_value *ts_begin = NULL;
1515         struct bt_ctf_clock_value *ts_end = NULL;
1516         int ret;
1517
1518         ret = get_ts_begin_ts_end_from_packet(packet, &ts_begin, &ts_end);
1519         if (ret) {
1520                 BT_LOGW("Cannot get packet's beginning or end clock values: "
1521                         "packet-addr=%p, ret=%d", packet, ret);
1522                 ret = -1;
1523                 goto end;
1524         }
1525
1526         ret = handle_discarded_packets(iterator, packet, ts_begin, ts_end,
1527                 stream_state);
1528         if (ret) {
1529                 BT_LOGW("Cannot handle discarded packets for packet: "
1530                         "packet-addr=%p, ret=%d", packet, ret);
1531                 ret = -1;
1532                 goto end;
1533         }
1534
1535         ret = handle_discarded_events(iterator, packet, ts_begin, ts_end,
1536                 stream_state);
1537         if (ret) {
1538                 BT_LOGW("Cannot handle discarded events for packet: "
1539                         "packet-addr=%p, ret=%d", packet, ret);
1540                 ret = -1;
1541                 goto end;
1542         }
1543
1544 end:
1545         bt_put(ts_begin);
1546         bt_put(ts_end);
1547         return ret;
1548 }
1549
1550 static
1551 int handle_packet_switch(struct bt_notification_iterator *iterator,
1552                 struct bt_notification *packet_begin_notif,
1553                 struct bt_ctf_packet *new_packet,
1554                 struct stream_state *stream_state)
1555 {
1556         int ret = 0;
1557
1558         if (stream_state->cur_packet == new_packet) {
1559                 goto end;
1560         }
1561
1562         BT_LOGV("Handling packet switch: "
1563                 "cur-packet-addr=%p, new-packet-addr=%p",
1564                 stream_state->cur_packet, new_packet);
1565
1566         if (stream_state->cur_packet) {
1567                 /* End of the current packet */
1568                 ret = add_action_push_notif_packet_end(iterator,
1569                         stream_state->cur_packet);
1570                 if (ret) {
1571                         BT_LOGE_STR("Cannot add \"push packet end notification\" action.");
1572                         goto error;
1573                 }
1574         }
1575
1576         /*
1577          * Check the new packet's header and context fields for
1578          * discarded packets and events to emit those automatic
1579          * notifications.
1580          */
1581         ret = handle_discarded_elements(iterator, new_packet, stream_state);
1582         if (ret) {
1583                 BT_LOGE_STR("Cannot handle discarded elements for new packet.");
1584                 goto error;
1585         }
1586
1587         /* Beginning of the new packet */
1588         if (packet_begin_notif) {
1589                 add_action_push_notif(iterator, packet_begin_notif);
1590         } else if (new_packet) {
1591                 ret = add_action_push_notif_packet_begin(iterator,
1592                         new_packet);
1593                 if (ret) {
1594                         BT_LOGE_STR("Cannot add \"push packet beginning notification\" action.");
1595                         goto error;
1596                 }
1597         }
1598
1599         add_action_set_stream_state_cur_packet(iterator, stream_state,
1600                 new_packet);
1601         goto end;
1602
1603 error:
1604         ret = -1;
1605
1606 end:
1607         return ret;
1608 }
1609
1610 static
1611 int handle_notif_stream_begin(
1612                 struct bt_notification_iterator *iterator,
1613                 struct bt_notification *notif,
1614                 struct bt_ctf_stream *notif_stream)
1615 {
1616         int ret = 0;
1617         struct stream_state *stream_state;
1618
1619         assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_BEGIN);
1620         assert(notif_stream);
1621         ret = ensure_stream_state_exists(iterator, notif, notif_stream,
1622                 &stream_state);
1623         if (ret) {
1624                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1625                 goto error;
1626         }
1627
1628         goto end;
1629
1630 error:
1631         ret = -1;
1632
1633 end:
1634         return ret;
1635 }
1636
1637 static
1638 int handle_notif_stream_end(
1639                 struct bt_notification_iterator *iterator,
1640                 struct bt_notification *notif,
1641                 struct bt_ctf_stream *notif_stream)
1642 {
1643         int ret = 0;
1644         struct stream_state *stream_state;
1645
1646         assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_END);
1647         assert(notif_stream);
1648         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1649                 &stream_state);
1650         if (ret) {
1651                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1652                 goto error;
1653         }
1654
1655         ret = handle_packet_switch(iterator, NULL, NULL, stream_state);
1656         if (ret) {
1657                 BT_LOGE_STR("Cannot handle packet switch.");
1658                 goto error;
1659         }
1660
1661         add_action_push_notif(iterator, notif);
1662         add_action_set_stream_state_is_ended(iterator, stream_state);
1663         goto end;
1664
1665 error:
1666         ret = -1;
1667
1668 end:
1669         return ret;
1670 }
1671
1672 static
1673 int handle_notif_discarded_elements(
1674                 struct bt_notification_iterator *iterator,
1675                 struct bt_notification *notif,
1676                 struct bt_ctf_stream *notif_stream)
1677 {
1678         int ret = 0;
1679         struct stream_state *stream_state;
1680
1681         assert(notif->type == BT_NOTIFICATION_TYPE_DISCARDED_EVENTS ||
1682                         notif->type == BT_NOTIFICATION_TYPE_DISCARDED_PACKETS);
1683         assert(notif_stream);
1684         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1685                 &stream_state);
1686         if (ret) {
1687                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1688                 goto error;
1689         }
1690
1691         add_action_push_notif(iterator, notif);
1692         goto end;
1693
1694 error:
1695         ret = -1;
1696
1697 end:
1698         return ret;
1699 }
1700
1701 static
1702 int handle_notif_packet_begin(
1703                 struct bt_notification_iterator *iterator,
1704                 struct bt_notification *notif,
1705                 struct bt_ctf_stream *notif_stream,
1706                 struct bt_ctf_packet *notif_packet)
1707 {
1708         int ret = 0;
1709         struct stream_state *stream_state;
1710
1711         assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_BEGIN);
1712         assert(notif_packet);
1713         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1714                 &stream_state);
1715         if (ret) {
1716                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1717                 goto error;
1718         }
1719
1720         ret = handle_packet_switch(iterator, notif, notif_packet, stream_state);
1721         if (ret) {
1722                 BT_LOGE_STR("Cannot handle packet switch.");
1723                 goto error;
1724         }
1725
1726         goto end;
1727
1728 error:
1729         ret = -1;
1730
1731 end:
1732         return ret;
1733 }
1734
1735 static
1736 int handle_notif_packet_end(
1737                 struct bt_notification_iterator *iterator,
1738                 struct bt_notification *notif,
1739                 struct bt_ctf_stream *notif_stream,
1740                 struct bt_ctf_packet *notif_packet)
1741 {
1742         int ret = 0;
1743         struct stream_state *stream_state;
1744
1745         assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_END);
1746         assert(notif_packet);
1747         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1748                 &stream_state);
1749         if (ret) {
1750                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1751                 goto error;
1752         }
1753
1754         ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state);
1755         if (ret) {
1756                 BT_LOGE_STR("Cannot handle packet switch.");
1757                 goto error;
1758         }
1759
1760         /* End of the current packet */
1761         add_action_push_notif(iterator, notif);
1762         add_action_set_stream_state_cur_packet(iterator, stream_state, NULL);
1763         goto end;
1764
1765 error:
1766         ret = -1;
1767
1768 end:
1769         return ret;
1770 }
1771
1772 static
1773 int handle_notif_event(
1774                 struct bt_notification_iterator *iterator,
1775                 struct bt_notification *notif,
1776                 struct bt_ctf_stream *notif_stream,
1777                 struct bt_ctf_packet *notif_packet)
1778 {
1779         int ret = 0;
1780         struct stream_state *stream_state;
1781
1782         assert(notif->type == BT_NOTIFICATION_TYPE_EVENT);
1783         assert(notif_packet);
1784         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1785                 &stream_state);
1786         if (ret) {
1787                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1788                 goto error;
1789         }
1790
1791         ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state);
1792         if (ret) {
1793                 BT_LOGE_STR("Cannot handle packet switch.");
1794                 goto error;
1795         }
1796
1797         add_action_push_notif(iterator, notif);
1798         goto end;
1799
1800 error:
1801         ret = -1;
1802
1803 end:
1804         return ret;
1805 }
1806
1807 static
1808 int enqueue_notification_and_automatic(
1809                 struct bt_notification_iterator *iterator,
1810                 struct bt_notification *notif)
1811 {
1812         int ret = 0;
1813         struct bt_ctf_event *notif_event = NULL;
1814         struct bt_ctf_stream *notif_stream = NULL;
1815         struct bt_ctf_packet *notif_packet = NULL;
1816
1817         assert(notif);
1818
1819         BT_LOGV("Enqueuing user notification and automatic notifications: "
1820                 "iter-addr=%p, notif-addr=%p", iterator, notif);
1821
1822         // TODO: Skip most of this if the iterator is only subscribed
1823         //       to event/inactivity notifications.
1824
1825         /* Get the stream and packet referred by the notification */
1826         switch (notif->type) {
1827         case BT_NOTIFICATION_TYPE_EVENT:
1828                 notif_event = bt_notification_event_borrow_event(notif);
1829                 assert(notif_event);
1830                 notif_packet = bt_ctf_event_borrow_packet(notif_event);
1831                 assert(notif_packet);
1832                 break;
1833         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
1834                 notif_stream =
1835                         bt_notification_stream_begin_borrow_stream(notif);
1836                 assert(notif_stream);
1837                 break;
1838         case BT_NOTIFICATION_TYPE_STREAM_END:
1839                 notif_stream = bt_notification_stream_end_borrow_stream(notif);
1840                 assert(notif_stream);
1841                 break;
1842         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
1843                 notif_packet =
1844                         bt_notification_packet_begin_borrow_packet(notif);
1845                 assert(notif_packet);
1846                 break;
1847         case BT_NOTIFICATION_TYPE_PACKET_END:
1848                 notif_packet = bt_notification_packet_end_borrow_packet(notif);
1849                 assert(notif_packet);
1850                 break;
1851         case BT_NOTIFICATION_TYPE_INACTIVITY:
1852                 /* Always valid */
1853                 goto handle_notif;
1854         default:
1855                 /*
1856                  * Invalid type of notification. Only the notification
1857                  * types above are allowed to be returned by a user
1858                  * component.
1859                  */
1860                 BT_LOGF("Unexpected notification type at this point: "
1861                         "notif-addr=%p, notif-type=%s", notif,
1862                         bt_notification_type_string(notif->type));
1863                 abort();
1864         }
1865
1866         if (notif_packet) {
1867                 notif_stream = bt_ctf_packet_borrow_stream(notif_packet);
1868                 assert(notif_stream);
1869         }
1870
1871         if (!notif_stream) {
1872                 /*
1873                  * The notification has no reference to a stream: it
1874                  * cannot cause the creation of automatic notifications.
1875                  */
1876                 BT_LOGV_STR("Notification has no reference to any stream: skipping automatic notification generation.");
1877                 goto end;
1878         }
1879
1880         if (!validate_notification(iterator, notif, notif_stream,
1881                         notif_packet)) {
1882                 BT_LOGW_STR("Invalid notification.");
1883                 goto error;
1884         }
1885
1886 handle_notif:
1887         switch (notif->type) {
1888         case BT_NOTIFICATION_TYPE_EVENT:
1889                 ret = handle_notif_event(iterator, notif, notif_stream,
1890                         notif_packet);
1891                 break;
1892         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
1893                 ret = handle_notif_stream_begin(iterator, notif, notif_stream);
1894                 break;
1895         case BT_NOTIFICATION_TYPE_STREAM_END:
1896                 ret = handle_notif_stream_end(iterator, notif, notif_stream);
1897                 break;
1898         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
1899                 ret = handle_notif_packet_begin(iterator, notif, notif_stream,
1900                         notif_packet);
1901                 break;
1902         case BT_NOTIFICATION_TYPE_PACKET_END:
1903                 ret = handle_notif_packet_end(iterator, notif, notif_stream,
1904                         notif_packet);
1905                 break;
1906         case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS:
1907         case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS:
1908                 ret = handle_notif_discarded_elements(iterator, notif,
1909                         notif_stream);
1910                 break;
1911         case BT_NOTIFICATION_TYPE_INACTIVITY:
1912                 add_action_push_notif(iterator, notif);
1913                 break;
1914         default:
1915                 break;
1916         }
1917
1918         if (ret) {
1919                 BT_LOGW_STR("Failed to handle notification for automatic notification generation.");
1920                 goto error;
1921         }
1922
1923         apply_actions(iterator);
1924         BT_LOGV("Enqueued user notification and automatic notifications: "
1925                 "iter-addr=%p, notif-addr=%p", iterator, notif);
1926         goto end;
1927
1928 error:
1929         ret = -1;
1930
1931 end:
1932         return ret;
1933 }
1934
1935 static
1936 int handle_end(struct bt_notification_iterator *iterator)
1937 {
1938         GHashTableIter stream_state_iter;
1939         gpointer stream_gptr, stream_state_gptr;
1940         int ret = 0;
1941
1942         BT_LOGV("Handling end of iteration: addr=%p", iterator);
1943
1944         /*
1945          * Emit a "stream end" notification for each non-ended stream
1946          * known by this iterator and mark them as ended.
1947          */
1948         g_hash_table_iter_init(&stream_state_iter, iterator->stream_states);
1949
1950         while (g_hash_table_iter_next(&stream_state_iter, &stream_gptr,
1951                         &stream_state_gptr)) {
1952                 struct stream_state *stream_state = stream_state_gptr;
1953
1954                 assert(stream_state_gptr);
1955
1956                 if (stream_state->is_ended) {
1957                         continue;
1958                 }
1959
1960                 ret = handle_packet_switch(iterator, NULL, NULL, stream_state);
1961                 if (ret) {
1962                         BT_LOGE_STR("Cannot handle packet switch.");
1963                         goto error;
1964                 }
1965
1966                 ret = add_action_push_notif_stream_end(iterator, stream_gptr);
1967                 if (ret) {
1968                         BT_LOGE_STR("Cannot add \"push stream end notification\" action.");
1969                         goto error;
1970                 }
1971
1972                 add_action_set_stream_state_is_ended(iterator, stream_state);
1973         }
1974
1975         apply_actions(iterator);
1976         BT_LOGV("Handled end of iteration: addr=%p", iterator);
1977         goto end;
1978
1979 error:
1980         ret = -1;
1981
1982 end:
1983         return ret;
1984 }
1985
1986 static
1987 enum bt_notification_iterator_status ensure_queue_has_notifications(
1988                 struct bt_notification_iterator *iterator)
1989 {
1990         struct bt_private_notification_iterator *priv_iterator =
1991                 bt_private_notification_iterator_from_notification_iterator(iterator);
1992         bt_component_class_notification_iterator_next_method next_method = NULL;
1993         struct bt_notification_iterator_next_return next_return = {
1994                 .status = BT_NOTIFICATION_ITERATOR_STATUS_OK,
1995                 .notification = NULL,
1996         };
1997         enum bt_notification_iterator_status status =
1998                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
1999         int ret;
2000
2001         assert(iterator);
2002         BT_LOGD("Ensuring that notification iterator's queue has at least one notification: "
2003                 "iter-addr=%p, queue-size=%u, iter-state=%s",
2004                 iterator, iterator->queue->length,
2005                 bt_notification_iterator_state_string(iterator->state));
2006
2007         if (iterator->queue->length > 0) {
2008                 /*
2009                  * We already have enough. Even if this notification
2010                  * iterator is finalized, its user can still flush its
2011                  * current queue's content by calling its "next" method
2012                  * since this content is local and has no impact on what
2013                  * used to be the iterator's upstream component.
2014                  */
2015                 BT_LOGD_STR("Queue already has at least one notification.");
2016                 goto end;
2017         }
2018
2019         switch (iterator->state) {
2020         case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED:
2021         case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED:
2022                 BT_LOGD_STR("Notification iterator's \"next\" called, but it is finalized.");
2023                 status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
2024                 goto end;
2025         case BT_NOTIFICATION_ITERATOR_STATE_ENDED:
2026                 BT_LOGD_STR("Notification iterator is ended.");
2027                 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
2028                 goto end;
2029         default:
2030                 break;
2031         }
2032
2033         assert(iterator->upstream_component);
2034         assert(iterator->upstream_component->class);
2035
2036         /* Pick the appropriate "next" method */
2037         switch (iterator->upstream_component->class->type) {
2038         case BT_COMPONENT_CLASS_TYPE_SOURCE:
2039         {
2040                 struct bt_component_class_source *source_class =
2041                         container_of(iterator->upstream_component->class,
2042                                 struct bt_component_class_source, parent);
2043
2044                 assert(source_class->methods.iterator.next);
2045                 next_method = source_class->methods.iterator.next;
2046                 break;
2047         }
2048         case BT_COMPONENT_CLASS_TYPE_FILTER:
2049         {
2050                 struct bt_component_class_filter *filter_class =
2051                         container_of(iterator->upstream_component->class,
2052                                 struct bt_component_class_filter, parent);
2053
2054                 assert(filter_class->methods.iterator.next);
2055                 next_method = filter_class->methods.iterator.next;
2056                 break;
2057         }
2058         default:
2059                 abort();
2060         }
2061
2062         /*
2063          * Call the user's "next" method to get the next notification
2064          * and status.
2065          */
2066         assert(next_method);
2067
2068         while (iterator->queue->length == 0) {
2069                 BT_LOGD_STR("Calling user's \"next\" method.");
2070                 next_return = next_method(priv_iterator);
2071                 BT_LOGD("User method returned: status=%s",
2072                         bt_notification_iterator_status_string(next_return.status));
2073                 if (next_return.status < 0) {
2074                         BT_LOGW_STR("User method failed.");
2075                         status = next_return.status;
2076                         goto end;
2077                 }
2078
2079                 if (iterator->state == BT_NOTIFICATION_ITERATOR_STATE_FINALIZED ||
2080                                 iterator->state == BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED) {
2081                         /*
2082                          * The user's "next" method, somehow, cancelled
2083                          * its own notification iterator. This can
2084                          * happen, for example, when the user's method
2085                          * removes the port on which there's the
2086                          * connection from which the iterator was
2087                          * created. In this case, said connection is
2088                          * ended, and all its notification iterators are
2089                          * finalized.
2090                          *
2091                          * Only bt_put() the returned notification if
2092                          * the status is
2093                          * BT_NOTIFICATION_ITERATOR_STATUS_OK because
2094                          * otherwise this field could be garbage.
2095                          */
2096                         if (next_return.status ==
2097                                         BT_NOTIFICATION_ITERATOR_STATUS_OK) {
2098                                 bt_put(next_return.notification);
2099                         }
2100
2101                         status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
2102                         goto end;
2103                 }
2104
2105                 switch (next_return.status) {
2106                 case BT_NOTIFICATION_ITERATOR_STATUS_END:
2107                         ret = handle_end(iterator);
2108                         if (ret) {
2109                                 BT_LOGW_STR("Cannot handle end of iteration.");
2110                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
2111                                 goto end;
2112                         }
2113
2114                         assert(iterator->state ==
2115                                 BT_NOTIFICATION_ITERATOR_STATE_ACTIVE);
2116                         iterator->state = BT_NOTIFICATION_ITERATOR_STATE_ENDED;
2117
2118                         if (iterator->queue->length == 0) {
2119                                 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
2120                         }
2121
2122                         BT_LOGD("Set new status: status=%s",
2123                                 bt_notification_iterator_status_string(status));
2124                         goto end;
2125                 case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
2126                         status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN;
2127                         goto end;
2128                 case BT_NOTIFICATION_ITERATOR_STATUS_OK:
2129                         if (!next_return.notification) {
2130                                 BT_LOGW_STR("User method returned BT_NOTIFICATION_ITERATOR_STATUS_OK, but notification is NULL.");
2131                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
2132                                 goto end;
2133                         }
2134
2135                         /*
2136                          * Ignore some notifications which are always
2137                          * automatically generated by the notification
2138                          * iterator to make sure they have valid values.
2139                          */
2140                         switch (next_return.notification->type) {
2141                         case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS:
2142                         case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS:
2143                                 BT_LOGV("Ignoring discarded elements notification returned by notification iterator's \"next\" method: "
2144                                         "notif-type=%s",
2145                                         bt_notification_type_string(next_return.notification->type));
2146                                 BT_PUT(next_return.notification);
2147                                 continue;
2148                         default:
2149                                 break;
2150                         }
2151
2152                         /*
2153                          * We know the notification is valid. Before we
2154                          * push it to the head of the queue, push the
2155                          * appropriate automatic notifications if any.
2156                          */
2157                         ret = enqueue_notification_and_automatic(iterator,
2158                                 next_return.notification);
2159                         BT_PUT(next_return.notification);
2160                         if (ret) {
2161                                 BT_LOGW("Cannot enqueue notification and automatic notifications.");
2162                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
2163                                 goto end;
2164                         }
2165                         break;
2166                 default:
2167                         /* Unknown non-error status */
2168                         abort();
2169                 }
2170         }
2171
2172 end:
2173         return status;
2174 }
2175
2176 enum bt_notification_iterator_status
2177 bt_notification_iterator_next(struct bt_notification_iterator *iterator)
2178 {
2179         enum bt_notification_iterator_status status;
2180
2181         if (!iterator) {
2182                 BT_LOGW_STR("Invalid parameter: notification iterator is NULL.");
2183                 status = BT_NOTIFICATION_ITERATOR_STATUS_INVALID;
2184                 goto end;
2185         }
2186
2187         BT_LOGD("Notification iterator's \"next\": iter-addr=%p", iterator);
2188
2189         /*
2190          * Make sure that the iterator's queue contains at least one
2191          * notification.
2192          */
2193         status = ensure_queue_has_notifications(iterator);
2194         if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
2195                 /* Not an error */
2196                 goto end;
2197         }
2198
2199         /*
2200          * Move the notification at the tail of the queue to the
2201          * iterator's current notification.
2202          */
2203         assert(iterator->queue->length > 0);
2204         bt_put(iterator->current_notification);
2205         iterator->current_notification = g_queue_pop_tail(iterator->queue);
2206         assert(iterator->current_notification);
2207
2208 end:
2209         return status;
2210 }
2211
2212 struct bt_component *bt_notification_iterator_get_component(
2213                 struct bt_notification_iterator *iterator)
2214 {
2215         return bt_get(iterator->upstream_component);
2216 }
2217
2218 struct bt_private_component *
2219 bt_private_notification_iterator_get_private_component(
2220                 struct bt_private_notification_iterator *private_iterator)
2221 {
2222         return bt_private_component_from_component(
2223                 bt_notification_iterator_get_component(
2224                         bt_notification_iterator_from_private(private_iterator)));
2225 }
2226
2227 enum bt_notification_iterator_status bt_notification_iterator_seek_time(
2228                 struct bt_notification_iterator *iterator,
2229                 enum bt_notification_iterator_seek_origin seek_origin,
2230                 int64_t time)
2231 {
2232         enum bt_notification_iterator_status ret =
2233                         BT_NOTIFICATION_ITERATOR_STATUS_UNSUPPORTED;
2234         return ret;
2235 }