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