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