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