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         BT_MOVE(*user_iterator, iterator);
 593
 594 end:
 595         bt_put(iterator);
 596         return status;
 597 }
 598
 599 void *bt_private_notification_iterator_get_user_data(
 600                 struct bt_private_notification_iterator *private_iterator)
 601 {
 602         struct bt_notification_iterator *iterator =
 603                 bt_notification_iterator_from_private(private_iterator);
 604
 605         return iterator ? iterator->user_data : NULL;
 606 }
 607
 608 enum bt_notification_iterator_status
 609 bt_private_notification_iterator_set_user_data(
 610                 struct bt_private_notification_iterator *private_iterator,
 611                 void *data)
 612 {
 613         enum bt_notification_iterator_status ret =
 614                         BT_NOTIFICATION_ITERATOR_STATUS_OK;
 615         struct bt_notification_iterator *iterator =
 616                 bt_notification_iterator_from_private(private_iterator);
 617
 618         if (!iterator) {
 619                 BT_LOGW_STR("Invalid parameter: notification iterator is NULL.");
 620                 ret = BT_NOTIFICATION_ITERATOR_STATUS_INVALID;
 621                 goto end;
 622         }
 623
 624         iterator->user_data = data;
 625         BT_LOGV("Set notification iterator's user data: "
 626                 "iter-addr=%p, user-data-addr=%p", iterator, data);
 627
 628 end:
 629         return ret;
 630 }
 631
 632 struct bt_notification *bt_notification_iterator_get_notification(
 633                 struct bt_notification_iterator *iterator)
 634 {
 635         struct bt_notification *notification = NULL;
 636
 637         if (!iterator) {
 638                 BT_LOGW_STR("Invalid parameter: notification iterator is NULL.");
 639                 goto end;
 640         }
 641
 642         notification = bt_get(iterator->current_notification);
 643
 644 end:
 645         return notification;
 646 }
 647
 648 static
 649 enum bt_notification_iterator_notif_type
 650 bt_notification_iterator_notif_type_from_notif_type(
 651                 enum bt_notification_type notif_type)
 652 {
 653         enum bt_notification_iterator_notif_type iter_notif_type;
 654
 655         switch (notif_type) {
 656         case BT_NOTIFICATION_TYPE_EVENT:
 657                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT;
 658                 break;
 659         case BT_NOTIFICATION_TYPE_INACTIVITY:
 660                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY;
 661                 break;
 662         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 663                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN;
 664                 break;
 665         case BT_NOTIFICATION_TYPE_STREAM_END:
 666                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END;
 667                 break;
 668         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 669                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN;
 670                 break;
 671         case BT_NOTIFICATION_TYPE_PACKET_END:
 672                 iter_notif_type = BT_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END;
 673                 break;
 674         default:
 675                 abort();
 676         }
 677
 678         return iter_notif_type;
 679 }
 680
 681 static
 682 bt_bool validate_notification(struct bt_notification_iterator *iterator,
 683                 struct bt_notification *notif,
 684                 struct bt_ctf_stream *notif_stream,
 685                 struct bt_ctf_packet *notif_packet)
 686 {
 687         bt_bool is_valid = BT_TRUE;
 688         struct stream_state *stream_state;
 689         struct bt_port *stream_comp_cur_port;
 690
 691         assert(notif_stream);
 692         stream_comp_cur_port =
 693                 bt_ctf_stream_port_for_component(notif_stream,
 694                         iterator->upstream_component);
 695         if (!stream_comp_cur_port) {
 696                 /*
 697                  * This is the first time this notification iterator
 698                  * bumps into this stream. Add an action to map the
 699                  * iterator's upstream component to the iterator's
 700                  * upstream port in this stream.
 701                  */
 702                 struct action action = {
 703                         .type = ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM,
 704                         .payload.map_port_to_comp_in_stream = {
 705                                 .stream = bt_get(notif_stream),
 706                                 .component = bt_get(iterator->upstream_component),
 707                                 .port = bt_get(iterator->upstream_port),
 708                         },
 709                 };
 710
 711                 add_action(iterator, &action);
 712         } else {
 713                 if (stream_comp_cur_port != iterator->upstream_port) {
 714                         /*
 715                          * It looks like two different ports of the same
 716                          * component are emitting notifications which
 717                          * have references to the same stream. This is
 718                          * bad: the API guarantees that it can never
 719                          * happen.
 720                          */
 721                         BT_LOGW("Two different ports of the same component are emitting notifications which refer to the same stream: "
 722                                 "stream-addr=%p, stream-name=\"%s\", "
 723                                 "stream-comp-cur-port-addr=%p, "
 724                                 "stream-comp-cur-port-name=%p, "
 725                                 "iter-upstream-port-addr=%p, "
 726                                 "iter-upstream-port-name=%s",
 727                                 notif_stream,
 728                                 bt_ctf_stream_get_name(notif_stream),
 729                                 stream_comp_cur_port,
 730                                 bt_port_get_name(stream_comp_cur_port),
 731                                 iterator->upstream_port,
 732                                 bt_port_get_name(iterator->upstream_port));
 733                         is_valid = BT_FALSE;
 734                         goto end;
 735                 }
 736
 737         }
 738
 739         stream_state = g_hash_table_lookup(iterator->stream_states,
 740                 notif_stream);
 741         if (stream_state) {
 742                 BT_LOGV("Stream state already exists: "
 743                         "stream-addr=%p, stream-name=\"%s\", "
 744                         "stream-state-addr=%p",
 745                         notif_stream,
 746                         bt_ctf_stream_get_name(notif_stream), stream_state);
 747
 748                 if (stream_state->is_ended) {
 749                         /*
 750                          * There's a new notification which has a
 751                          * reference to a stream which, from this
 752                          * iterator's point of view, is ended ("end of
 753                          * stream" notification was returned). This is
 754                          * bad: the API guarantees that it can never
 755                          * happen.
 756                          */
 757                         BT_LOGW("Stream is already ended: "
 758                                 "stream-addr=%p, stream-name=\"%s\"",
 759                                 notif_stream,
 760                                 bt_ctf_stream_get_name(notif_stream));
 761                         is_valid = BT_FALSE;
 762                         goto end;
 763                 }
 764
 765                 switch (notif->type) {
 766                 case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 767                         /*
 768                          * We already have a stream state, which means
 769                          * we already returned a "stream begin"
 770                          * notification: this is an invalid duplicate.
 771                          */
 772                         BT_LOGW("Duplicate stream beginning notification: "
 773                                 "stream-addr=%p, stream-name=\"%s\"",
 774                                 notif_stream,
 775                                 bt_ctf_stream_get_name(notif_stream));
 776                         is_valid = BT_FALSE;
 777                         goto end;
 778                 case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 779                         if (notif_packet == stream_state->cur_packet) {
 780                                 /* Duplicate "packet begin" notification */
 781                                 BT_LOGW("Duplicate stream beginning notification: "
 782                                         "stream-addr=%p, stream-name=\"%s\", "
 783                                         "packet-addr=%p",
 784                                         notif_stream,
 785                                         bt_ctf_stream_get_name(notif_stream),
 786                                         notif_packet);
 787                                 is_valid = BT_FALSE;
 788                                 goto end;
 789                         }
 790                         break;
 791                 default:
 792                         break;
 793                 }
 794         }
 795
 796 end:
 797         return is_valid;
 798 }
 799
 800 static
 801 bt_bool is_subscribed_to_notification_type(struct bt_notification_iterator *iterator,
 802                 enum bt_notification_type notif_type)
 803 {
 804         uint32_t iter_notif_type =
 805                 (uint32_t) bt_notification_iterator_notif_type_from_notif_type(
 806                         notif_type);
 807
 808         return (iter_notif_type & iterator->subscription_mask) ? BT_TRUE : BT_FALSE;
 809 }
 810
 811 static
 812 void add_action_push_notif(struct bt_notification_iterator *iterator,
 813                 struct bt_notification *notif)
 814 {
 815         struct action action = {
 816                 .type = ACTION_TYPE_PUSH_NOTIF,
 817         };
 818
 819         assert(notif);
 820
 821         if (!is_subscribed_to_notification_type(iterator, notif->type)) {
 822                 return;
 823         }
 824
 825         action.payload.push_notif.notif = bt_get(notif);
 826         add_action(iterator, &action);
 827         BT_LOGV("Added \"push notification\" action: notif-addr=%p", notif);
 828 }
 829
 830 static
 831 int add_action_push_notif_stream_begin(
 832                 struct bt_notification_iterator *iterator,
 833                 struct bt_ctf_stream *stream)
 834 {
 835         int ret = 0;
 836         struct bt_notification *stream_begin_notif = NULL;
 837
 838         if (!is_subscribed_to_notification_type(iterator,
 839                         BT_NOTIFICATION_TYPE_STREAM_BEGIN)) {
 840                 BT_LOGV("Not adding \"push stream beginning notification\" action: "
 841                         "notification iterator is not subscribed: addr=%p",
 842                         iterator);
 843                 goto end;
 844         }
 845
 846         assert(stream);
 847         stream_begin_notif = bt_notification_stream_begin_create(stream);
 848         if (!stream_begin_notif) {
 849                 BT_LOGE_STR("Cannot create stream beginning notification.");
 850                 goto error;
 851         }
 852
 853         add_action_push_notif(iterator, stream_begin_notif);
 854         BT_LOGV("Added \"push stream beginning notification\" action: "
 855                 "stream-addr=%p, stream-name=\"%s\"",
 856                 stream, bt_ctf_stream_get_name(stream));
 857         goto end;
 858
 859 error:
 860         ret = -1;
 861
 862 end:
 863         bt_put(stream_begin_notif);
 864         return ret;
 865 }
 866
 867 static
 868 int add_action_push_notif_stream_end(
 869                 struct bt_notification_iterator *iterator,
 870                 struct bt_ctf_stream *stream)
 871 {
 872         int ret = 0;
 873         struct bt_notification *stream_end_notif = NULL;
 874
 875         if (!is_subscribed_to_notification_type(iterator,
 876                         BT_NOTIFICATION_TYPE_STREAM_END)) {
 877                 BT_LOGV("Not adding \"push stream end notification\" action: "
 878                         "notification iterator is not subscribed: addr=%p",
 879                         iterator);
 880                 goto end;
 881         }
 882
 883         assert(stream);
 884         stream_end_notif = bt_notification_stream_end_create(stream);
 885         if (!stream_end_notif) {
 886                 BT_LOGE_STR("Cannot create stream end notification.");
 887                 goto error;
 888         }
 889
 890         add_action_push_notif(iterator, stream_end_notif);
 891         BT_LOGV("Added \"push stream end notification\" action: "
 892                 "stream-addr=%p, stream-name=\"%s\"",
 893                 stream, bt_ctf_stream_get_name(stream));
 894         goto end;
 895
 896 error:
 897         ret = -1;
 898
 899 end:
 900         bt_put(stream_end_notif);
 901         return ret;
 902 }
 903
 904 static
 905 int add_action_push_notif_packet_begin(
 906                 struct bt_notification_iterator *iterator,
 907                 struct bt_ctf_packet *packet)
 908 {
 909         int ret = 0;
 910         struct bt_notification *packet_begin_notif = NULL;
 911
 912         if (!is_subscribed_to_notification_type(iterator,
 913                         BT_NOTIFICATION_TYPE_PACKET_BEGIN)) {
 914                 BT_LOGV("Not adding \"push packet beginning notification\" action: "
 915                         "notification iterator is not subscribed: addr=%p",
 916                         iterator);
 917                 goto end;
 918         }
 919
 920         assert(packet);
 921         packet_begin_notif = bt_notification_packet_begin_create(packet);
 922         if (!packet_begin_notif) {
 923                 BT_LOGE_STR("Cannot create packet beginning notification.");
 924                 goto error;
 925         }
 926
 927         add_action_push_notif(iterator, packet_begin_notif);
 928         BT_LOGV("Added \"push packet beginning notification\" action: "
 929                 "packet-addr=%p", packet);
 930         goto end;
 931
 932 error:
 933         ret = -1;
 934
 935 end:
 936         bt_put(packet_begin_notif);
 937         return ret;
 938 }
 939
 940 static
 941 int add_action_push_notif_packet_end(
 942                 struct bt_notification_iterator *iterator,
 943                 struct bt_ctf_packet *packet)
 944 {
 945         int ret = 0;
 946         struct bt_notification *packet_end_notif = NULL;
 947
 948         if (!is_subscribed_to_notification_type(iterator,
 949                         BT_NOTIFICATION_TYPE_PACKET_END)) {
 950                 BT_LOGV("Not adding \"push packet end notification\" action: "
 951                         "notification iterator is not subscribed: addr=%p",
 952                         iterator);
 953                 goto end;
 954         }
 955
 956         assert(packet);
 957         packet_end_notif = bt_notification_packet_end_create(packet);
 958         if (!packet_end_notif) {
 959                 BT_LOGE_STR("Cannot create packet end notification.");
 960                 goto error;
 961         }
 962
 963         add_action_push_notif(iterator, packet_end_notif);
 964         BT_LOGV("Added \"push packet end notification\" action: "
 965                 "packet-addr=%p", packet);
 966         goto end;
 967
 968 error:
 969         ret = -1;
 970
 971 end:
 972         bt_put(packet_end_notif);
 973         return ret;
 974 }
 975
 976 static
 977 void add_action_set_stream_state_is_ended(
 978                 struct bt_notification_iterator *iterator,
 979                 struct stream_state *stream_state)
 980 {
 981         struct action action = {
 982                 .type = ACTION_TYPE_SET_STREAM_STATE_IS_ENDED,
 983                 .payload.set_stream_state_is_ended = {
 984                         .stream_state = stream_state,
 985                 },
 986         };
 987
 988         assert(stream_state);
 989         add_action(iterator, &action);
 990         BT_LOGV("Added \"set stream state's ended\" action: "
 991                 "stream-state-addr=%p", stream_state);
 992 }
 993
 994 static
 995 void add_action_set_stream_state_cur_packet(
 996                 struct bt_notification_iterator *iterator,
 997                 struct stream_state *stream_state,
 998                 struct bt_ctf_packet *packet)
 999 {
1000         struct action action = {
1001                 .type = ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET,
1002                 .payload.set_stream_state_cur_packet = {
1003                         .stream_state = stream_state,
1004                         .packet = bt_get(packet),
1005                 },
1006         };
1007
1008         assert(stream_state);
1009         add_action(iterator, &action);
1010         BT_LOGV("Added \"set stream state's current packet\" action: "
1011                 "stream-state-addr=%p, packet-addr=%p",
1012                 stream_state, packet);
1013 }
1014
1015 static
1016 int ensure_stream_state_exists(struct bt_notification_iterator *iterator,
1017                 struct bt_notification *stream_begin_notif,
1018                 struct bt_ctf_stream *notif_stream,
1019                 struct stream_state **stream_state)
1020 {
1021         int ret = 0;
1022
1023         if (!notif_stream) {
1024                 /*
1025                  * The notification does not reference any stream: no
1026                  * need to get or create a stream state.
1027                  */
1028                 goto end;
1029         }
1030
1031         *stream_state = g_hash_table_lookup(iterator->stream_states,
1032                 notif_stream);
1033         if (!*stream_state) {
1034                 /*
1035                  * This iterator did not bump into this stream yet:
1036                  * create a stream state and a "stream begin"
1037                  * notification.
1038                  */
1039                 struct action action = {
1040                         .type = ACTION_TYPE_ADD_STREAM_STATE,
1041                         .payload.add_stream_state = {
1042                                 .stream = bt_get(notif_stream),
1043                                 .stream_state = NULL,
1044                         },
1045                 };
1046
1047                 *stream_state = create_stream_state(notif_stream);
1048                 if (!stream_state) {
1049                         BT_LOGE_STR("Cannot create stream state.");
1050                         goto error;
1051                 }
1052
1053                 action.payload.add_stream_state.stream_state =
1054                         *stream_state;
1055                 add_action(iterator, &action);
1056
1057                 if (stream_begin_notif) {
1058                         add_action_push_notif(iterator, stream_begin_notif);
1059                 } else {
1060                         ret = add_action_push_notif_stream_begin(iterator,
1061                                 notif_stream);
1062                         if (ret) {
1063                                 BT_LOGE_STR("Cannot add \"push stream beginning notification\" action.");
1064                                 goto error;
1065                         }
1066                 }
1067         }
1068
1069         goto end;
1070
1071 error:
1072         destroy_stream_state(*stream_state);
1073         ret = -1;
1074
1075 end:
1076         return ret;
1077 }
1078
1079 static
1080 int handle_packet_switch(struct bt_notification_iterator *iterator,
1081                 struct bt_notification *packet_begin_notif,
1082                 struct bt_ctf_packet *new_packet,
1083                 struct stream_state *stream_state)
1084 {
1085         int ret = 0;
1086
1087         if (stream_state->cur_packet == new_packet) {
1088                 goto end;
1089         }
1090
1091         BT_LOGV("Handling packet switch: "
1092                 "cur-packet-addr=%p, new-packet-addr=%p",
1093                 stream_state->cur_packet, new_packet);
1094
1095         if (stream_state->cur_packet) {
1096                 /* End of the current packet */
1097                 ret = add_action_push_notif_packet_end(iterator,
1098                         stream_state->cur_packet);
1099                 if (ret) {
1100                         BT_LOGE_STR("Cannot add \"push packet end notification\" action.");
1101                         goto error;
1102                 }
1103         }
1104
1105         /* Beginning of the new packet */
1106         if (packet_begin_notif) {
1107                 add_action_push_notif(iterator, packet_begin_notif);
1108         } else if (new_packet) {
1109                 ret = add_action_push_notif_packet_begin(iterator,
1110                         new_packet);
1111                 if (ret) {
1112                         BT_LOGE_STR("Cannot add \"push packet beginning notification\" action.");
1113                         goto error;
1114                 }
1115         }
1116
1117         add_action_set_stream_state_cur_packet(iterator, stream_state,
1118                 new_packet);
1119         goto end;
1120
1121 error:
1122         ret = -1;
1123
1124 end:
1125         return ret;
1126 }
1127
1128 static
1129 int handle_notif_stream_begin(
1130                 struct bt_notification_iterator *iterator,
1131                 struct bt_notification *notif,
1132                 struct bt_ctf_stream *notif_stream)
1133 {
1134         int ret = 0;
1135         struct stream_state *stream_state;
1136
1137         assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_BEGIN);
1138         assert(notif_stream);
1139         ret = ensure_stream_state_exists(iterator, notif, notif_stream,
1140                 &stream_state);
1141         if (ret) {
1142                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1143                 goto error;
1144         }
1145
1146         goto end;
1147
1148 error:
1149         ret = -1;
1150
1151 end:
1152         return ret;
1153 }
1154
1155 static
1156 int handle_notif_stream_end(
1157                 struct bt_notification_iterator *iterator,
1158                 struct bt_notification *notif,
1159                 struct bt_ctf_stream *notif_stream)
1160 {
1161         int ret = 0;
1162         struct stream_state *stream_state;
1163
1164         assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_END);
1165         assert(notif_stream);
1166         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1167                 &stream_state);
1168         if (ret) {
1169                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1170                 goto error;
1171         }
1172
1173         ret = handle_packet_switch(iterator, NULL, NULL, stream_state);
1174         if (ret) {
1175                 BT_LOGE_STR("Cannot handle packet switch.");
1176                 goto error;
1177         }
1178
1179         add_action_push_notif(iterator, notif);
1180         add_action_set_stream_state_is_ended(iterator, stream_state);
1181         goto end;
1182
1183 error:
1184         ret = -1;
1185
1186 end:
1187         return ret;
1188 }
1189
1190 static
1191 int handle_notif_packet_begin(
1192                 struct bt_notification_iterator *iterator,
1193                 struct bt_notification *notif,
1194                 struct bt_ctf_stream *notif_stream,
1195                 struct bt_ctf_packet *notif_packet)
1196 {
1197         int ret = 0;
1198         struct stream_state *stream_state;
1199
1200         assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_BEGIN);
1201         assert(notif_packet);
1202         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1203                 &stream_state);
1204         if (ret) {
1205                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1206                 goto error;
1207         }
1208
1209         ret = handle_packet_switch(iterator, notif, notif_packet, stream_state);
1210         if (ret) {
1211                 BT_LOGE_STR("Cannot handle packet switch.");
1212                 goto error;
1213         }
1214
1215         goto end;
1216
1217 error:
1218         ret = -1;
1219
1220 end:
1221         return ret;
1222 }
1223
1224 static
1225 int handle_notif_packet_end(
1226                 struct bt_notification_iterator *iterator,
1227                 struct bt_notification *notif,
1228                 struct bt_ctf_stream *notif_stream,
1229                 struct bt_ctf_packet *notif_packet)
1230 {
1231         int ret = 0;
1232         struct stream_state *stream_state;
1233
1234         assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_END);
1235         assert(notif_packet);
1236         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1237                 &stream_state);
1238         if (ret) {
1239                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1240                 goto error;
1241         }
1242
1243         ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state);
1244         if (ret) {
1245                 BT_LOGE_STR("Cannot handle packet switch.");
1246                 goto error;
1247         }
1248
1249         /* End of the current packet */
1250         add_action_push_notif(iterator, notif);
1251         add_action_set_stream_state_cur_packet(iterator, stream_state, NULL);
1252         goto end;
1253
1254 error:
1255         ret = -1;
1256
1257 end:
1258         return ret;
1259 }
1260
1261 static
1262 int handle_notif_event(
1263                 struct bt_notification_iterator *iterator,
1264                 struct bt_notification *notif,
1265                 struct bt_ctf_stream *notif_stream,
1266                 struct bt_ctf_packet *notif_packet)
1267 {
1268         int ret = 0;
1269         struct stream_state *stream_state;
1270
1271         assert(notif->type == BT_NOTIFICATION_TYPE_EVENT);
1272         assert(notif_packet);
1273         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1274                 &stream_state);
1275         if (ret) {
1276                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1277                 goto error;
1278         }
1279
1280         ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state);
1281         if (ret) {
1282                 BT_LOGE_STR("Cannot handle packet switch.");
1283                 goto error;
1284         }
1285
1286         add_action_push_notif(iterator, notif);
1287         goto end;
1288
1289 error:
1290         ret = -1;
1291
1292 end:
1293         return ret;
1294 }
1295
1296 static
1297 int enqueue_notification_and_automatic(
1298                 struct bt_notification_iterator *iterator,
1299                 struct bt_notification *notif)
1300 {
1301         int ret = 0;
1302         struct bt_ctf_event *notif_event = NULL;
1303         struct bt_ctf_stream *notif_stream = NULL;
1304         struct bt_ctf_packet *notif_packet = NULL;
1305
1306         assert(notif);
1307
1308         BT_LOGV("Enqueuing user notification and automatic notifications: "
1309                 "iter-addr=%p, notif-addr=%p", iterator, notif);
1310
1311         // TODO: Skip most of this if the iterator is only subscribed
1312         //       to event/inactivity notifications.
1313
1314         /* Get the stream and packet referred by the notification */
1315         switch (notif->type) {
1316         case BT_NOTIFICATION_TYPE_EVENT:
1317                 notif_event = bt_notification_event_borrow_event(notif);
1318                 assert(notif_event);
1319                 notif_packet = bt_ctf_event_borrow_packet(notif_event);
1320                 assert(notif_packet);
1321                 break;
1322         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
1323                 notif_stream =
1324                         bt_notification_stream_begin_borrow_stream(notif);
1325                 assert(notif_stream);
1326                 break;
1327         case BT_NOTIFICATION_TYPE_STREAM_END:
1328                 notif_stream = bt_notification_stream_end_borrow_stream(notif);
1329                 assert(notif_stream);
1330                 break;
1331         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
1332                 notif_packet =
1333                         bt_notification_packet_begin_borrow_packet(notif);
1334                 assert(notif_packet);
1335                 break;
1336         case BT_NOTIFICATION_TYPE_PACKET_END:
1337                 notif_packet = bt_notification_packet_end_borrow_packet(notif);
1338                 assert(notif_packet);
1339                 break;
1340         case BT_NOTIFICATION_TYPE_INACTIVITY:
1341                 /* Always valid */
1342                 goto handle_notif;
1343         default:
1344                 /*
1345                  * Invalid type of notification. Only the notification
1346                  * types above are allowed to be returned by a user
1347                  * component.
1348                  */
1349                 goto error;
1350         }
1351
1352         if (notif_packet) {
1353                 notif_stream = bt_ctf_packet_borrow_stream(notif_packet);
1354                 assert(notif_stream);
1355         }
1356
1357         if (!notif_stream) {
1358                 /*
1359                  * The notification has no reference to a stream: it
1360                  * cannot cause the creation of automatic notifications.
1361                  */
1362                 BT_LOGV_STR("Notification has no reference to any stream: skipping automatic notification generation.");
1363                 goto end;
1364         }
1365
1366         if (!validate_notification(iterator, notif, notif_stream,
1367                         notif_packet)) {
1368                 BT_LOGW_STR("Invalid notification.");
1369                 goto error;
1370         }
1371
1372 handle_notif:
1373         switch (notif->type) {
1374         case BT_NOTIFICATION_TYPE_EVENT:
1375                 ret = handle_notif_event(iterator, notif, notif_stream,
1376                         notif_packet);
1377                 break;
1378         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
1379                 ret = handle_notif_stream_begin(iterator, notif, notif_stream);
1380                 break;
1381         case BT_NOTIFICATION_TYPE_STREAM_END:
1382                 ret = handle_notif_stream_end(iterator, notif, notif_stream);
1383                 break;
1384         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
1385                 ret = handle_notif_packet_begin(iterator, notif, notif_stream,
1386                         notif_packet);
1387                 break;
1388         case BT_NOTIFICATION_TYPE_PACKET_END:
1389                 ret = handle_notif_packet_end(iterator, notif, notif_stream,
1390                         notif_packet);
1391                 break;
1392         case BT_NOTIFICATION_TYPE_INACTIVITY:
1393                 add_action_push_notif(iterator, notif);
1394                 break;
1395         default:
1396                 break;
1397         }
1398
1399         if (ret) {
1400                 BT_LOGW_STR("Failed to handle notification for automatic notification generation.");
1401                 goto error;
1402         }
1403
1404         apply_actions(iterator);
1405         BT_LOGV("Enqueued user notification and automatic notifications: "
1406                 "iter-addr=%p, notif-addr=%p", iterator, notif);
1407         goto end;
1408
1409 error:
1410         ret = -1;
1411
1412 end:
1413         return ret;
1414 }
1415
1416 static
1417 int handle_end(struct bt_notification_iterator *iterator)
1418 {
1419         GHashTableIter stream_state_iter;
1420         gpointer stream_gptr, stream_state_gptr;
1421         int ret = 0;
1422
1423         BT_LOGV("Handling end of iteration: addr=%p", iterator);
1424
1425         /*
1426          * Emit a "stream end" notification for each non-ended stream
1427          * known by this iterator and mark them as ended.
1428          */
1429         g_hash_table_iter_init(&stream_state_iter, iterator->stream_states);
1430
1431         while (g_hash_table_iter_next(&stream_state_iter, &stream_gptr,
1432                         &stream_state_gptr)) {
1433                 struct stream_state *stream_state = stream_state_gptr;
1434
1435                 assert(stream_state_gptr);
1436
1437                 if (stream_state->is_ended) {
1438                         continue;
1439                 }
1440
1441                 ret = handle_packet_switch(iterator, NULL, NULL, stream_state);
1442                 if (ret) {
1443                         BT_LOGE_STR("Cannot handle packet switch.");
1444                         goto error;
1445                 }
1446
1447                 ret = add_action_push_notif_stream_end(iterator, stream_gptr);
1448                 if (ret) {
1449                         BT_LOGE_STR("Cannot add \"push stream end notification\" action.");
1450                         goto error;
1451                 }
1452
1453                 add_action_set_stream_state_is_ended(iterator, stream_state);
1454         }
1455
1456         apply_actions(iterator);
1457         BT_LOGV("Handled end of iteration: addr=%p", iterator);
1458         goto end;
1459
1460 error:
1461         ret = -1;
1462
1463 end:
1464         return ret;
1465 }
1466
1467 static
1468 enum bt_notification_iterator_status ensure_queue_has_notifications(
1469                 struct bt_notification_iterator *iterator)
1470 {
1471         struct bt_private_notification_iterator *priv_iterator =
1472                 bt_private_notification_iterator_from_notification_iterator(iterator);
1473         bt_component_class_notification_iterator_next_method next_method = NULL;
1474         struct bt_notification_iterator_next_return next_return = {
1475                 .status = BT_NOTIFICATION_ITERATOR_STATUS_OK,
1476                 .notification = NULL,
1477         };
1478         enum bt_notification_iterator_status status =
1479                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
1480         int ret;
1481
1482         assert(iterator);
1483         BT_LOGD("Ensuring that notification iterator's queue has at least one notification: "
1484                 "iter-addr=%p, queue-size=%u, iter-state=%s",
1485                 iterator, iterator->queue->length,
1486                 bt_notification_iterator_state_string(iterator->state));
1487
1488         if (iterator->queue->length > 0) {
1489                 /*
1490                  * We already have enough. Even if this notification
1491                  * iterator is finalized, its user can still flush its
1492                  * current queue's content by calling its "next" method
1493                  * since this content is local and has no impact on what
1494                  * used to be the iterator's upstream component.
1495                  */
1496                 BT_LOGD_STR("Queue already has at least one notification.");
1497                 goto end;
1498         }
1499
1500         switch (iterator->state) {
1501         case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED:
1502         case BT_NOTIFICATION_ITERATOR_STATE_FINALIZED:
1503                 BT_LOGD_STR("Notification iterator's \"next\" called, but it is finalized.");
1504                 status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
1505                 goto end;
1506         case BT_NOTIFICATION_ITERATOR_STATE_ENDED:
1507                 BT_LOGD_STR("Notification iterator is ended.");
1508                 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
1509                 goto end;
1510         default:
1511                 break;
1512         }
1513
1514         assert(iterator->upstream_component);
1515         assert(iterator->upstream_component->class);
1516
1517         /* Pick the appropriate "next" method */
1518         switch (iterator->upstream_component->class->type) {
1519         case BT_COMPONENT_CLASS_TYPE_SOURCE:
1520         {
1521                 struct bt_component_class_source *source_class =
1522                         container_of(iterator->upstream_component->class,
1523                                 struct bt_component_class_source, parent);
1524
1525                 assert(source_class->methods.iterator.next);
1526                 next_method = source_class->methods.iterator.next;
1527                 break;
1528         }
1529         case BT_COMPONENT_CLASS_TYPE_FILTER:
1530         {
1531                 struct bt_component_class_filter *filter_class =
1532                         container_of(iterator->upstream_component->class,
1533                                 struct bt_component_class_filter, parent);
1534
1535                 assert(filter_class->methods.iterator.next);
1536                 next_method = filter_class->methods.iterator.next;
1537                 break;
1538         }
1539         default:
1540                 abort();
1541         }
1542
1543         /*
1544          * Call the user's "next" method to get the next notification
1545          * and status.
1546          */
1547         assert(next_method);
1548
1549         while (iterator->queue->length == 0) {
1550                 BT_LOGD_STR("Calling user's \"next\" method.");
1551                 next_return = next_method(priv_iterator);
1552                 BT_LOGD("User method returned: status=%s",
1553                         bt_notification_iterator_status_string(next_return.status));
1554                 if (next_return.status < 0) {
1555                         BT_LOGW_STR("User method failed.");
1556                         status = next_return.status;
1557                         goto end;
1558                 }
1559
1560                 if (iterator->state == BT_NOTIFICATION_ITERATOR_STATE_FINALIZED ||
1561                                 iterator->state == BT_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED) {
1562                         /*
1563                          * The user's "next" method, somehow, cancelled
1564                          * its own notification iterator. This can
1565                          * happen, for example, when the user's method
1566                          * removes the port on which there's the
1567                          * connection from which the iterator was
1568                          * created. In this case, said connection is
1569                          * ended, and all its notification iterators are
1570                          * finalized.
1571                          *
1572                          * Only bt_put() the returned notification if
1573                          * the status is
1574                          * BT_NOTIFICATION_ITERATOR_STATUS_OK because
1575                          * otherwise this field could be garbage.
1576                          */
1577                         if (next_return.status ==
1578                                         BT_NOTIFICATION_ITERATOR_STATUS_OK) {
1579                                 bt_put(next_return.notification);
1580                         }
1581
1582                         status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
1583                         goto end;
1584                 }
1585
1586                 switch (next_return.status) {
1587                 case BT_NOTIFICATION_ITERATOR_STATUS_END:
1588                         ret = handle_end(iterator);
1589                         if (ret) {
1590                                 BT_LOGW_STR("Cannot handle end of iteration.");
1591                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
1592                                 goto end;
1593                         }
1594
1595                         assert(iterator->state ==
1596                                 BT_NOTIFICATION_ITERATOR_STATE_ACTIVE);
1597                         iterator->state = BT_NOTIFICATION_ITERATOR_STATE_ENDED;
1598
1599                         if (iterator->queue->length == 0) {
1600                                 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
1601                         }
1602
1603                         BT_LOGD("Set new status: status=%s",
1604                                 bt_notification_iterator_status_string(status));
1605                         goto end;
1606                 case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
1607                         status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN;
1608                         goto end;
1609                 case BT_NOTIFICATION_ITERATOR_STATUS_OK:
1610                         if (!next_return.notification) {
1611                                 BT_LOGW_STR("User method returned BT_NOTIFICATION_ITERATOR_STATUS_OK, but notification is NULL.");
1612                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
1613                                 goto end;
1614                         }
1615
1616                         /*
1617                          * We know the notification is valid. Before we
1618                          * push it to the head of the queue, push the
1619                          * appropriate automatic notifications if any.
1620                          */
1621                         ret = enqueue_notification_and_automatic(iterator,
1622                                 next_return.notification);
1623                         BT_PUT(next_return.notification);
1624                         if (ret) {
1625                                 BT_LOGW("Cannot enqueue notification and automatic notifications.");
1626                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
1627                                 goto end;
1628                         }
1629                         break;
1630                 default:
1631                         /* Unknown non-error status */
1632                         abort();
1633                 }
1634         }
1635
1636 end:
1637         return status;
1638 }
1639
1640 enum bt_notification_iterator_status
1641 bt_notification_iterator_next(struct bt_notification_iterator *iterator)
1642 {
1643         enum bt_notification_iterator_status status;
1644
1645         if (!iterator) {
1646                 BT_LOGW_STR("Invalid parameter: notification iterator is NULL.");
1647                 status = BT_NOTIFICATION_ITERATOR_STATUS_INVALID;
1648                 goto end;
1649         }
1650
1651         BT_LOGD("Notification iterator's \"next\": iter-addr=%p", iterator);
1652
1653         /*
1654          * Make sure that the iterator's queue contains at least one
1655          * notification.
1656          */
1657         status = ensure_queue_has_notifications(iterator);
1658         if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
1659                 /* Not an error */
1660                 goto end;
1661         }
1662
1663         /*
1664          * Move the notification at the tail of the queue to the
1665          * iterator's current notification.
1666          */
1667         assert(iterator->queue->length > 0);
1668         bt_put(iterator->current_notification);
1669         iterator->current_notification = g_queue_pop_tail(iterator->queue);
1670         assert(iterator->current_notification);
1671
1672 end:
1673         return status;
1674 }
1675
1676 struct bt_component *bt_notification_iterator_get_component(
1677                 struct bt_notification_iterator *iterator)
1678 {
1679         return bt_get(iterator->upstream_component);
1680 }
1681
1682 struct bt_private_component *
1683 bt_private_notification_iterator_get_private_component(
1684                 struct bt_private_notification_iterator *private_iterator)
1685 {
1686         return bt_private_component_from_component(
1687                 bt_notification_iterator_get_component(
1688                         bt_notification_iterator_from_private(private_iterator)));
1689 }
1690
1691 enum bt_notification_iterator_status bt_notification_iterator_seek_time(
1692                 struct bt_notification_iterator *iterator,
1693                 enum bt_notification_iterator_seek_origin seek_origin,
1694                 int64_t time)
1695 {
1696         enum bt_notification_iterator_status ret =
1697                         BT_NOTIFICATION_ITERATOR_STATUS_UNSUPPORTED;
1698         return ret;
1699 }