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