lib/graph/iterator.c

   1 /*
   2  * iterator.c
   3  *
   4  * Babeltrace Notification Iterator
   5  *
   6  * Copyright 2015 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   7  * Copyright 2017 Philippe Proulx <pproulx@efficios.com>
   8  *
   9  * Permission is hereby granted, free of charge, to any person obtaining a copy
  10  * of this software and associated documentation files (the "Software"), to deal
  11  * in the Software without restriction, including without limitation the rights
  12  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  13  * copies of the Software, and to permit persons to whom the Software is
  14  * furnished to do so, subject to the following conditions:
  15  *
  16  * The above copyright notice and this permission notice shall be included in
  17  * all copies or substantial portions of the Software.
  18  *
  19  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  20  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  21  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  22  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  23  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  24  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  25  * SOFTWARE.
  26  */
  27
  28 #define BT_LOG_TAG "NOTIF-ITER"
  29 #include <babeltrace/lib-logging-internal.h>
  30
  31 #include <babeltrace/compiler-internal.h>
  32 #include <babeltrace/ref.h>
  33 #include <babeltrace/ctf-ir/fields.h>
  34 #include <babeltrace/ctf-ir/field-types.h>
  35 #include <babeltrace/ctf-ir/field-types-internal.h>
  36 #include <babeltrace/ctf-ir/event-internal.h>
  37 #include <babeltrace/ctf-ir/packet-internal.h>
  38 #include <babeltrace/ctf-ir/stream-internal.h>
  39 #include <babeltrace/graph/connection.h>
  40 #include <babeltrace/graph/connection-internal.h>
  41 #include <babeltrace/graph/component.h>
  42 #include <babeltrace/graph/component-source-internal.h>
  43 #include <babeltrace/graph/component-class-internal.h>
  44 #include <babeltrace/graph/component-class-sink-colander-internal.h>
  45 #include <babeltrace/graph/component-sink.h>
  46 #include <babeltrace/graph/notification.h>
  47 #include <babeltrace/graph/notification-iterator.h>
  48 #include <babeltrace/graph/notification-iterator-internal.h>
  49 #include <babeltrace/graph/notification-internal.h>
  50 #include <babeltrace/graph/notification-event.h>
  51 #include <babeltrace/graph/notification-event-internal.h>
  52 #include <babeltrace/graph/notification-packet.h>
  53 #include <babeltrace/graph/notification-packet-internal.h>
  54 #include <babeltrace/graph/notification-stream.h>
  55 #include <babeltrace/graph/notification-stream-internal.h>
  56 #include <babeltrace/graph/notification-discarded-elements-internal.h>
  57 #include <babeltrace/graph/port.h>
  58 #include <babeltrace/graph/graph-internal.h>
  59 #include <babeltrace/types.h>
  60 #include <stdint.h>
  61 #include <inttypes.h>
  62 #include <stdlib.h>
  63
  64 struct discarded_elements_state {
  65         struct bt_clock_value *cur_begin;
  66         uint64_t cur_count;
  67 };
  68
  69 struct stream_state {
  70         struct bt_stream *stream; /* owned by this */
  71         struct bt_packet *cur_packet; /* owned by this */
  72         struct discarded_elements_state discarded_packets_state;
  73         struct discarded_elements_state discarded_events_state;
  74         bt_bool is_ended;
  75 };
  76
  77 enum action_type {
  78         ACTION_TYPE_PUSH_NOTIF,
  79         ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM,
  80         ACTION_TYPE_ADD_STREAM_STATE,
  81         ACTION_TYPE_SET_STREAM_STATE_IS_ENDED,
  82         ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET,
  83         ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS,
  84         ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS,
  85 };
  86
  87 struct action {
  88         enum action_type type;
  89         union {
  90                 /* ACTION_TYPE_PUSH_NOTIF */
  91                 struct {
  92                         struct bt_notification *notif; /* owned by this */
  93                 } push_notif;
  94
  95                 /* ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM */
  96                 struct {
  97                         struct bt_stream *stream; /* owned by this */
  98                         struct bt_component *component; /* owned by this */
  99                         struct bt_port *port; /* owned by this */
 100                 } map_port_to_comp_in_stream;
 101
 102                 /* ACTION_TYPE_ADD_STREAM_STATE */
 103                 struct {
 104                         struct bt_stream *stream; /* owned by this */
 105                         struct stream_state *stream_state; /* owned by this */
 106                 } add_stream_state;
 107
 108                 /* ACTION_TYPE_SET_STREAM_STATE_IS_ENDED */
 109                 struct {
 110                         struct stream_state *stream_state; /* weak */
 111                 } set_stream_state_is_ended;
 112
 113                 /* ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET */
 114                 struct {
 115                         struct stream_state *stream_state; /* weak */
 116                         struct bt_packet *packet; /* owned by this */
 117                 } set_stream_state_cur_packet;
 118
 119                 /* ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS */
 120                 /* ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS */
 121                 struct {
 122                         struct stream_state *stream_state; /* weak */
 123                         struct bt_clock_value *cur_begin; /* owned by this */
 124                         uint64_t cur_count;
 125                 } update_stream_state_discarded_elements;
 126         } payload;
 127 };
 128
 129 static
 130 void stream_destroy_listener(struct bt_stream *stream, void *data)
 131 {
 132         struct bt_notification_iterator_private_connection *iterator = data;
 133
 134         /* Remove associated stream state */
 135         g_hash_table_remove(iterator->stream_states, stream);
 136 }
 137
 138 static
 139 void destroy_stream_state(struct stream_state *stream_state)
 140 {
 141         if (!stream_state) {
 142                 return;
 143         }
 144
 145         BT_LOGV("Destroying stream state: stream-state-addr=%p", stream_state);
 146         BT_LOGV_STR("Putting stream state's current packet.");
 147         bt_put(stream_state->cur_packet);
 148         BT_LOGV_STR("Putting stream state's stream.");
 149         bt_put(stream_state->stream);
 150         bt_put(stream_state->discarded_packets_state.cur_begin);
 151         bt_put(stream_state->discarded_events_state.cur_begin);
 152         g_free(stream_state);
 153 }
 154
 155 static
 156 void destroy_action(struct action *action)
 157 {
 158         assert(action);
 159
 160         switch (action->type) {
 161         case ACTION_TYPE_PUSH_NOTIF:
 162                 BT_PUT(action->payload.push_notif.notif);
 163                 break;
 164         case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM:
 165                 BT_PUT(action->payload.map_port_to_comp_in_stream.stream);
 166                 BT_PUT(action->payload.map_port_to_comp_in_stream.component);
 167                 BT_PUT(action->payload.map_port_to_comp_in_stream.port);
 168                 break;
 169         case ACTION_TYPE_ADD_STREAM_STATE:
 170                 BT_PUT(action->payload.add_stream_state.stream);
 171                 destroy_stream_state(
 172                         action->payload.add_stream_state.stream_state);
 173                 action->payload.add_stream_state.stream_state = NULL;
 174                 break;
 175         case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET:
 176                 BT_PUT(action->payload.set_stream_state_cur_packet.packet);
 177                 break;
 178         case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED:
 179                 break;
 180         case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS:
 181         case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS:
 182                 BT_PUT(action->payload.update_stream_state_discarded_elements.cur_begin);
 183                 break;
 184         default:
 185                 BT_LOGF("Unexpected action's type: type=%d", action->type);
 186                 abort();
 187         }
 188 }
 189
 190 static
 191 void add_action(struct bt_notification_iterator_private_connection *iterator,
 192                 struct action *action)
 193 {
 194         g_array_append_val(iterator->actions, *action);
 195 }
 196
 197 static
 198 void clear_actions(struct bt_notification_iterator_private_connection *iterator)
 199 {
 200         size_t i;
 201
 202         for (i = 0; i < iterator->actions->len; i++) {
 203                 struct action *action = &g_array_index(iterator->actions,
 204                         struct action, i);
 205
 206                 destroy_action(action);
 207         }
 208
 209         g_array_set_size(iterator->actions, 0);
 210 }
 211
 212 static inline
 213 const char *action_type_string(enum action_type type)
 214 {
 215         switch (type) {
 216         case ACTION_TYPE_PUSH_NOTIF:
 217                 return "ACTION_TYPE_PUSH_NOTIF";
 218         case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM:
 219                 return "ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM";
 220         case ACTION_TYPE_ADD_STREAM_STATE:
 221                 return "ACTION_TYPE_ADD_STREAM_STATE";
 222         case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED:
 223                 return "ACTION_TYPE_SET_STREAM_STATE_IS_ENDED";
 224         case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET:
 225                 return "ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET";
 226         case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS:
 227                 return "ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS";
 228         case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS:
 229                 return "ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS";
 230         default:
 231                 return "(unknown)";
 232         }
 233 }
 234
 235 static
 236 void apply_actions(struct bt_notification_iterator_private_connection *iterator)
 237 {
 238         size_t i;
 239
 240         BT_LOGV("Applying notification's iterator current actions: "
 241                 "count=%u", iterator->actions->len);
 242
 243         for (i = 0; i < iterator->actions->len; i++) {
 244                 struct action *action = &g_array_index(iterator->actions,
 245                         struct action, i);
 246
 247                 BT_LOGV("Applying action: index=%zu, type=%s",
 248                         i, action_type_string(action->type));
 249
 250                 switch (action->type) {
 251                 case ACTION_TYPE_PUSH_NOTIF:
 252                         /* Move notification to queue */
 253                         g_queue_push_head(iterator->queue,
 254                                 action->payload.push_notif.notif);
 255                         bt_notification_freeze(
 256                                 action->payload.push_notif.notif);
 257                         action->payload.push_notif.notif = NULL;
 258                         break;
 259                 case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM:
 260                         bt_stream_map_component_to_port(
 261                                 action->payload.map_port_to_comp_in_stream.stream,
 262                                 action->payload.map_port_to_comp_in_stream.component,
 263                                 action->payload.map_port_to_comp_in_stream.port);
 264                         break;
 265                 case ACTION_TYPE_ADD_STREAM_STATE:
 266                         /* Move stream state to hash table */
 267                         g_hash_table_insert(iterator->stream_states,
 268                                 action->payload.add_stream_state.stream,
 269                                 action->payload.add_stream_state.stream_state);
 270
 271                         action->payload.add_stream_state.stream_state = NULL;
 272                         break;
 273                 case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED:
 274                         /*
 275                          * We know that this stream is ended. We need to
 276                          * remember this as long as the stream exists to
 277                          * enforce that the same stream does not end
 278                          * twice.
 279                          *
 280                          * Here we add a destroy listener to the stream
 281                          * which we put after (becomes weak as the hash
 282                          * table key). If we were the last object to own
 283                          * this stream, the destroy listener is called
 284                          * when we call bt_put() which removes this
 285                          * stream state completely. This is important
 286                          * because the memory used by this stream object
 287                          * could be reused for another stream, and they
 288                          * must have different states.
 289                          */
 290                         bt_stream_add_destroy_listener(
 291                                 action->payload.set_stream_state_is_ended.stream_state->stream,
 292                                 stream_destroy_listener, iterator);
 293                         action->payload.set_stream_state_is_ended.stream_state->is_ended = BT_TRUE;
 294                         BT_PUT(action->payload.set_stream_state_is_ended.stream_state->stream);
 295                         break;
 296                 case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET:
 297                         /* Move packet to stream state's current packet */
 298                         BT_MOVE(action->payload.set_stream_state_cur_packet.stream_state->cur_packet,
 299                                 action->payload.set_stream_state_cur_packet.packet);
 300                         break;
 301                 case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS:
 302                 case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS:
 303                 {
 304                         struct discarded_elements_state *state;
 305
 306                         if (action->type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS) {
 307                                 state = &action->payload.update_stream_state_discarded_elements.stream_state->discarded_packets_state;
 308                         } else {
 309                                 state = &action->payload.update_stream_state_discarded_elements.stream_state->discarded_events_state;
 310                         }
 311
 312                         BT_MOVE(state->cur_begin,
 313                                 action->payload.update_stream_state_discarded_elements.cur_begin);
 314                         state->cur_count = action->payload.update_stream_state_discarded_elements.cur_count;
 315                         break;
 316                 }
 317                 default:
 318                         BT_LOGF("Unexpected action's type: type=%d",
 319                                 action->type);
 320                         abort();
 321                 }
 322         }
 323
 324         clear_actions(iterator);
 325 }
 326
 327 static
 328 struct stream_state *create_stream_state(struct bt_stream *stream)
 329 {
 330         struct stream_state *stream_state = g_new0(struct stream_state, 1);
 331
 332         if (!stream_state) {
 333                 BT_LOGE_STR("Failed to allocate one stream state.");
 334                 goto end;
 335         }
 336
 337         /*
 338          * The packet index is a monotonic counter which may not start
 339          * at 0 at the beginning of the stream. We therefore need to
 340          * have an internal object initial state of -1ULL to distinguish
 341          * between initial state and having seen a packet with
 342          * seqnum = 0.
 343          */
 344         stream_state->discarded_packets_state.cur_count = -1ULL;
 345
 346         /*
 347          * We keep a reference to the stream until we know it's ended
 348          * because we need to be able to create an automatic "stream
 349          * end" notification when the user's "next" method returns
 350          * BT_NOTIFICATION_ITERATOR_STATUS_END.
 351          *
 352          * We put this reference when the stream is marked as ended.
 353          */
 354         stream_state->stream = bt_get(stream);
 355         BT_LOGV("Created stream state: stream-addr=%p, stream-name=\"%s\", "
 356                 "stream-state-addr=%p",
 357                 stream, bt_stream_get_name(stream), stream_state);
 358
 359 end:
 360         return stream_state;
 361 }
 362
 363 static
 364 void destroy_base_notification_iterator(struct bt_object *obj)
 365 {
 366         struct bt_notification_iterator *iterator =
 367                 container_of(obj, struct bt_notification_iterator, base);
 368
 369         BT_LOGD_STR("Putting current notification.");
 370         bt_put(iterator->current_notification);
 371         g_free(iterator);
 372 }
 373
 374 static
 375 void bt_private_connection_notification_iterator_destroy(struct bt_object *obj)
 376 {
 377         struct bt_notification_iterator_private_connection *iterator;
 378
 379         assert(obj);
 380
 381         /*
 382          * The notification iterator's reference count is 0 if we're
 383          * here. Increment it to avoid a double-destroy (possibly
 384          * infinitely recursive). This could happen for example if the
 385          * notification iterator's finalization function does bt_get()
 386          * (or anything that causes bt_get() to be called) on itself
 387          * (ref. count goes from 0 to 1), and then bt_put(): the
 388          * reference count would go from 1 to 0 again and this function
 389          * would be called again.
 390          */
 391         obj->ref_count.count++;
 392         iterator = (void *) container_of(obj, struct bt_notification_iterator, base);
 393         BT_LOGD("Destroying private connection notification iterator object: addr=%p",
 394                 iterator);
 395         bt_private_connection_notification_iterator_finalize(iterator);
 396
 397         if (iterator->queue) {
 398                 struct bt_notification *notif;
 399
 400                 BT_LOGD("Putting notifications in queue.");
 401
 402                 while ((notif = g_queue_pop_tail(iterator->queue))) {
 403                         bt_put(notif);
 404                 }
 405
 406                 g_queue_free(iterator->queue);
 407         }
 408
 409         if (iterator->stream_states) {
 410                 /*
 411                  * Remove our destroy listener from each stream which
 412                  * has a state in this iterator. Otherwise the destroy
 413                  * listener would be called with an invalid/other
 414                  * notification iterator object.
 415                  */
 416                 GHashTableIter ht_iter;
 417                 gpointer stream_gptr, stream_state_gptr;
 418
 419                 g_hash_table_iter_init(&ht_iter, iterator->stream_states);
 420
 421                 while (g_hash_table_iter_next(&ht_iter, &stream_gptr, &stream_state_gptr)) {
 422                         assert(stream_gptr);
 423
 424                         BT_LOGD_STR("Removing stream's destroy listener for notification iterator.");
 425                         bt_stream_remove_destroy_listener(
 426                                 (void *) stream_gptr, stream_destroy_listener,
 427                                 iterator);
 428                 }
 429
 430                 g_hash_table_destroy(iterator->stream_states);
 431         }
 432
 433         if (iterator->actions) {
 434                 g_array_free(iterator->actions, TRUE);
 435         }
 436
 437         if (iterator->connection) {
 438                 /*
 439                  * Remove ourself from the originating connection so
 440                  * that it does not try to finalize a dangling pointer
 441                  * later.
 442                  */
 443                 bt_connection_remove_iterator(iterator->connection, iterator);
 444         }
 445
 446         destroy_base_notification_iterator(obj);
 447 }
 448
 449 BT_HIDDEN
 450 void bt_private_connection_notification_iterator_finalize(
 451                 struct bt_notification_iterator_private_connection *iterator)
 452 {
 453         struct bt_component_class *comp_class = NULL;
 454         bt_component_class_notification_iterator_finalize_method
 455                 finalize_method = NULL;
 456
 457         assert(iterator);
 458
 459         switch (iterator->state) {
 460         case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED:
 461                 /* Skip user finalization if user initialization failed */
 462                 BT_LOGD("Not finalizing non-initialized notification iterator: "
 463                         "addr=%p", iterator);
 464                 return;
 465         case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED:
 466         case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED:
 467                 /* Already finalized */
 468                 BT_LOGD("Not finalizing notification iterator: already finalized: "
 469                         "addr=%p", iterator);
 470                 return;
 471         default:
 472                 break;
 473         }
 474
 475         BT_LOGD("Finalizing notification iterator: addr=%p", iterator);
 476
 477         if (iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ENDED) {
 478                 BT_LOGD("Updating notification iterator's state: "
 479                         "new-state=BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED");
 480                 iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED;
 481         } else {
 482                 BT_LOGD("Updating notification iterator's state: "
 483                         "new-state=BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED");
 484                 iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED;
 485         }
 486
 487         assert(iterator->upstream_component);
 488         comp_class = iterator->upstream_component->class;
 489
 490         /* Call user-defined destroy method */
 491         switch (comp_class->type) {
 492         case BT_COMPONENT_CLASS_TYPE_SOURCE:
 493         {
 494                 struct bt_component_class_source *source_class;
 495
 496                 source_class = container_of(comp_class, struct bt_component_class_source, parent);
 497                 finalize_method = source_class->methods.iterator.finalize;
 498                 break;
 499         }
 500         case BT_COMPONENT_CLASS_TYPE_FILTER:
 501         {
 502                 struct bt_component_class_filter *filter_class;
 503
 504                 filter_class = container_of(comp_class, struct bt_component_class_filter, parent);
 505                 finalize_method = filter_class->methods.iterator.finalize;
 506                 break;
 507         }
 508         default:
 509                 /* Unreachable */
 510                 abort();
 511         }
 512
 513         if (finalize_method) {
 514                 BT_LOGD("Calling user's finalization method: addr=%p",
 515                         iterator);
 516                 finalize_method(
 517                         bt_private_connection_private_notification_iterator_from_notification_iterator(iterator));
 518         }
 519
 520         iterator->upstream_component = NULL;
 521         iterator->upstream_port = NULL;
 522         BT_LOGD("Finalized notification iterator: addr=%p", iterator);
 523 }
 524
 525 BT_HIDDEN
 526 void bt_private_connection_notification_iterator_set_connection(
 527                 struct bt_notification_iterator_private_connection *iterator,
 528                 struct bt_connection *connection)
 529 {
 530         assert(iterator);
 531         iterator->connection = connection;
 532         BT_LOGV("Set notification iterator's connection: "
 533                 "iter-addr=%p, conn-addr=%p", iterator, connection);
 534 }
 535
 536 static
 537 int create_subscription_mask_from_notification_types(
 538                 struct bt_notification_iterator_private_connection *iterator,
 539                 const enum bt_notification_type *notif_types)
 540 {
 541         const enum bt_notification_type *notif_type;
 542         int ret = 0;
 543
 544         assert(notif_types);
 545         iterator->subscription_mask = 0;
 546
 547         for (notif_type = notif_types;
 548                         *notif_type != BT_NOTIFICATION_TYPE_SENTINEL;
 549                         notif_type++) {
 550                 switch (*notif_type) {
 551                 case BT_NOTIFICATION_TYPE_ALL:
 552                         iterator->subscription_mask |=
 553                                 BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT |
 554                                 BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY |
 555                                 BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN |
 556                                 BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END |
 557                                 BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN |
 558                                 BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END |
 559                                 BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_EVENTS |
 560                                 BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_PACKETS;
 561                         break;
 562                 case BT_NOTIFICATION_TYPE_EVENT:
 563                         iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT;
 564                         break;
 565                 case BT_NOTIFICATION_TYPE_INACTIVITY:
 566                         iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY;
 567                         break;
 568                 case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 569                         iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN;
 570                         break;
 571                 case BT_NOTIFICATION_TYPE_STREAM_END:
 572                         iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END;
 573                         break;
 574                 case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 575                         iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN;
 576                         break;
 577                 case BT_NOTIFICATION_TYPE_PACKET_END:
 578                         iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END;
 579                         break;
 580                 case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS:
 581                         iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_EVENTS;
 582                         break;
 583                 case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS:
 584                         iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_PACKETS;
 585                         break;
 586                 default:
 587                         ret = -1;
 588                         goto end;
 589                 }
 590
 591                 BT_LOGV("Added notification type to subscription mask: "
 592                         "type=%s, mask=%x",
 593                         bt_notification_type_string(*notif_type),
 594                         iterator->subscription_mask);
 595         }
 596
 597 end:
 598         return ret;
 599 }
 600
 601 static
 602 void init_notification_iterator(struct bt_notification_iterator *iterator,
 603                 enum bt_notification_iterator_type type,
 604                 bt_object_release_func destroy)
 605 {
 606         bt_object_init(iterator, destroy);
 607         iterator->type = type;
 608 }
 609
 610 BT_HIDDEN
 611 enum bt_connection_status bt_private_connection_notification_iterator_create(
 612                 struct bt_component *upstream_comp,
 613                 struct bt_port *upstream_port,
 614                 const enum bt_notification_type *notification_types,
 615                 struct bt_connection *connection,
 616                 struct bt_notification_iterator_private_connection **user_iterator)
 617 {
 618         enum bt_connection_status status = BT_CONNECTION_STATUS_OK;
 619         enum bt_component_class_type type;
 620         struct bt_notification_iterator_private_connection *iterator = NULL;
 621
 622         assert(upstream_comp);
 623         assert(upstream_port);
 624         assert(notification_types);
 625         assert(bt_port_is_connected(upstream_port));
 626         assert(user_iterator);
 627         BT_LOGD("Creating notification iterator on private connection: "
 628                 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
 629                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 630                 "conn-addr=%p",
 631                 upstream_comp, bt_component_get_name(upstream_comp),
 632                 upstream_port, bt_port_get_name(upstream_port),
 633                 connection);
 634         type = bt_component_get_class_type(upstream_comp);
 635         assert(type == BT_COMPONENT_CLASS_TYPE_SOURCE ||
 636                 type == BT_COMPONENT_CLASS_TYPE_FILTER);
 637         iterator = g_new0(struct bt_notification_iterator_private_connection, 1);
 638         if (!iterator) {
 639                 BT_LOGE_STR("Failed to allocate one private connection notification iterator.");
 640                 status = BT_CONNECTION_STATUS_NOMEM;
 641                 goto end;
 642         }
 643
 644         init_notification_iterator((void *) iterator,
 645                 BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION,
 646                 bt_private_connection_notification_iterator_destroy);
 647
 648         if (create_subscription_mask_from_notification_types(iterator,
 649                         notification_types)) {
 650                 BT_LOGW_STR("Cannot create subscription mask from notification types.");
 651                 status = BT_CONNECTION_STATUS_INVALID;
 652                 goto end;
 653         }
 654
 655         iterator->stream_states = g_hash_table_new_full(g_direct_hash,
 656                 g_direct_equal, NULL, (GDestroyNotify) destroy_stream_state);
 657         if (!iterator->stream_states) {
 658                 BT_LOGE_STR("Failed to allocate a GHashTable.");
 659                 status = BT_CONNECTION_STATUS_NOMEM;
 660                 goto end;
 661         }
 662
 663         iterator->queue = g_queue_new();
 664         if (!iterator->queue) {
 665                 BT_LOGE_STR("Failed to allocate a GQueue.");
 666                 status = BT_CONNECTION_STATUS_NOMEM;
 667                 goto end;
 668         }
 669
 670         iterator->actions = g_array_new(FALSE, FALSE, sizeof(struct action));
 671         if (!iterator->actions) {
 672                 BT_LOGE_STR("Failed to allocate a GArray.");
 673                 status = BT_CONNECTION_STATUS_NOMEM;
 674                 goto end;
 675         }
 676
 677         iterator->upstream_component = upstream_comp;
 678         iterator->upstream_port = upstream_port;
 679         iterator->connection = connection;
 680         iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED;
 681         BT_LOGD("Created notification iterator: "
 682                 "upstream-comp-addr=%p, upstream-comp-name=\"%s\", "
 683                 "upstream-port-addr=%p, upstream-port-name=\"%s\", "
 684                 "conn-addr=%p, iter-addr=%p",
 685                 upstream_comp, bt_component_get_name(upstream_comp),
 686                 upstream_port, bt_port_get_name(upstream_port),
 687                 connection, iterator);
 688
 689         /* Move reference to user */
 690         *user_iterator = iterator;
 691         iterator = NULL;
 692
 693 end:
 694         bt_put(iterator);
 695         return status;
 696 }
 697
 698 void *bt_private_connection_private_notification_iterator_get_user_data(
 699                 struct bt_private_connection_private_notification_iterator *private_iterator)
 700 {
 701         struct bt_notification_iterator_private_connection *iterator =
 702                 bt_private_connection_notification_iterator_borrow_from_private(private_iterator);
 703
 704         return iterator ? iterator->user_data : NULL;
 705 }
 706
 707 enum bt_notification_iterator_status
 708 bt_private_connection_private_notification_iterator_set_user_data(
 709                 struct bt_private_connection_private_notification_iterator *private_iterator,
 710                 void *data)
 711 {
 712         enum bt_notification_iterator_status ret =
 713                         BT_NOTIFICATION_ITERATOR_STATUS_OK;
 714         struct bt_notification_iterator_private_connection *iterator =
 715                 bt_private_connection_notification_iterator_borrow_from_private(private_iterator);
 716
 717         if (!iterator) {
 718                 BT_LOGW_STR("Invalid parameter: notification iterator is NULL.");
 719                 ret = BT_NOTIFICATION_ITERATOR_STATUS_INVALID;
 720                 goto end;
 721         }
 722
 723         iterator->user_data = data;
 724         BT_LOGV("Set notification iterator's user data: "
 725                 "iter-addr=%p, user-data-addr=%p", iterator, data);
 726
 727 end:
 728         return ret;
 729 }
 730
 731 struct bt_notification *bt_notification_iterator_get_notification(
 732                 struct bt_notification_iterator *iterator)
 733 {
 734         struct bt_notification *notification = NULL;
 735
 736         if (!iterator) {
 737                 BT_LOGW_STR("Invalid parameter: notification iterator is NULL.");
 738                 goto end;
 739         }
 740
 741         notification = bt_get(
 742                 bt_notification_iterator_borrow_current_notification(iterator));
 743
 744 end:
 745         return notification;
 746 }
 747
 748 static
 749 enum bt_private_connection_notification_iterator_notif_type
 750 bt_notification_iterator_notif_type_from_notif_type(
 751                 enum bt_notification_type notif_type)
 752 {
 753         enum bt_private_connection_notification_iterator_notif_type iter_notif_type;
 754
 755         switch (notif_type) {
 756         case BT_NOTIFICATION_TYPE_EVENT:
 757                 iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT;
 758                 break;
 759         case BT_NOTIFICATION_TYPE_INACTIVITY:
 760                 iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY;
 761                 break;
 762         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 763                 iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN;
 764                 break;
 765         case BT_NOTIFICATION_TYPE_STREAM_END:
 766                 iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END;
 767                 break;
 768         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 769                 iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN;
 770                 break;
 771         case BT_NOTIFICATION_TYPE_PACKET_END:
 772                 iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END;
 773                 break;
 774         case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS:
 775                 iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_EVENTS;
 776                 break;
 777         case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS:
 778                 iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_PACKETS;
 779                 break;
 780         default:
 781                 abort();
 782         }
 783
 784         return iter_notif_type;
 785 }
 786
 787 static
 788 bt_bool validate_notification(
 789                 struct bt_notification_iterator_private_connection *iterator,
 790                 struct bt_notification *notif,
 791                 struct bt_stream *notif_stream,
 792                 struct bt_packet *notif_packet)
 793 {
 794         bt_bool is_valid = BT_TRUE;
 795         struct stream_state *stream_state;
 796         struct bt_port *stream_comp_cur_port;
 797
 798         assert(notif_stream);
 799         stream_comp_cur_port =
 800                 bt_stream_port_for_component(notif_stream,
 801                         iterator->upstream_component);
 802         if (!stream_comp_cur_port) {
 803                 /*
 804                  * This is the first time this notification iterator
 805                  * bumps into this stream. Add an action to map the
 806                  * iterator's upstream component to the iterator's
 807                  * upstream port in this stream.
 808                  */
 809                 struct action action = {
 810                         .type = ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM,
 811                         .payload.map_port_to_comp_in_stream = {
 812                                 .stream = bt_get(notif_stream),
 813                                 .component = bt_get(iterator->upstream_component),
 814                                 .port = bt_get(iterator->upstream_port),
 815                         },
 816                 };
 817
 818                 add_action(iterator, &action);
 819         } else {
 820                 if (stream_comp_cur_port != iterator->upstream_port) {
 821                         /*
 822                          * It looks like two different ports of the same
 823                          * component are emitting notifications which
 824                          * have references to the same stream. This is
 825                          * bad: the API guarantees that it can never
 826                          * happen.
 827                          */
 828                         BT_LOGW("Two different ports of the same component are emitting notifications which refer to the same stream: "
 829                                 "stream-addr=%p, stream-name=\"%s\", "
 830                                 "stream-comp-cur-port-addr=%p, "
 831                                 "stream-comp-cur-port-name=%p, "
 832                                 "iter-upstream-port-addr=%p, "
 833                                 "iter-upstream-port-name=%s",
 834                                 notif_stream,
 835                                 bt_stream_get_name(notif_stream),
 836                                 stream_comp_cur_port,
 837                                 bt_port_get_name(stream_comp_cur_port),
 838                                 iterator->upstream_port,
 839                                 bt_port_get_name(iterator->upstream_port));
 840                         is_valid = BT_FALSE;
 841                         goto end;
 842                 }
 843
 844         }
 845
 846         stream_state = g_hash_table_lookup(iterator->stream_states,
 847                 notif_stream);
 848         if (stream_state) {
 849                 BT_LOGV("Stream state already exists: "
 850                         "stream-addr=%p, stream-name=\"%s\", "
 851                         "stream-state-addr=%p",
 852                         notif_stream,
 853                         bt_stream_get_name(notif_stream), stream_state);
 854
 855                 if (stream_state->is_ended) {
 856                         /*
 857                          * There's a new notification which has a
 858                          * reference to a stream which, from this
 859                          * iterator's point of view, is ended ("end of
 860                          * stream" notification was returned). This is
 861                          * bad: the API guarantees that it can never
 862                          * happen.
 863                          */
 864                         BT_LOGW("Stream is already ended: "
 865                                 "stream-addr=%p, stream-name=\"%s\"",
 866                                 notif_stream,
 867                                 bt_stream_get_name(notif_stream));
 868                         is_valid = BT_FALSE;
 869                         goto end;
 870                 }
 871
 872                 switch (notif->type) {
 873                 case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
 874                         /*
 875                          * We already have a stream state, which means
 876                          * we already returned a "stream begin"
 877                          * notification: this is an invalid duplicate.
 878                          */
 879                         BT_LOGW("Duplicate stream beginning notification: "
 880                                 "stream-addr=%p, stream-name=\"%s\"",
 881                                 notif_stream,
 882                                 bt_stream_get_name(notif_stream));
 883                         is_valid = BT_FALSE;
 884                         goto end;
 885                 case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
 886                         if (notif_packet == stream_state->cur_packet) {
 887                                 /* Duplicate "packet begin" notification */
 888                                 BT_LOGW("Duplicate stream beginning notification: "
 889                                         "stream-addr=%p, stream-name=\"%s\", "
 890                                         "packet-addr=%p",
 891                                         notif_stream,
 892                                         bt_stream_get_name(notif_stream),
 893                                         notif_packet);
 894                                 is_valid = BT_FALSE;
 895                                 goto end;
 896                         }
 897                         break;
 898                 default:
 899                         break;
 900                 }
 901         }
 902
 903 end:
 904         return is_valid;
 905 }
 906
 907 static
 908 bt_bool is_subscribed_to_notification_type(
 909                 struct bt_notification_iterator_private_connection *iterator,
 910                 enum bt_notification_type notif_type)
 911 {
 912         uint32_t iter_notif_type =
 913                 (uint32_t) bt_notification_iterator_notif_type_from_notif_type(
 914                         notif_type);
 915
 916         return (iter_notif_type & iterator->subscription_mask) ? BT_TRUE : BT_FALSE;
 917 }
 918
 919 static
 920 void add_action_push_notif(
 921                 struct bt_notification_iterator_private_connection *iterator,
 922                 struct bt_notification *notif)
 923 {
 924         struct action action = {
 925                 .type = ACTION_TYPE_PUSH_NOTIF,
 926         };
 927
 928         assert(notif);
 929
 930         if (!is_subscribed_to_notification_type(iterator, notif->type)) {
 931                 return;
 932         }
 933
 934         action.payload.push_notif.notif = bt_get(notif);
 935         add_action(iterator, &action);
 936         BT_LOGV("Added \"push notification\" action: notif-addr=%p", notif);
 937 }
 938
 939 static
 940 int add_action_push_notif_stream_begin(
 941                 struct bt_notification_iterator_private_connection *iterator,
 942                 struct bt_stream *stream)
 943 {
 944         int ret = 0;
 945         struct bt_notification *stream_begin_notif = NULL;
 946
 947         if (!is_subscribed_to_notification_type(iterator,
 948                         BT_NOTIFICATION_TYPE_STREAM_BEGIN)) {
 949                 BT_LOGV("Not adding \"push stream beginning notification\" action: "
 950                         "notification iterator is not subscribed: addr=%p",
 951                         iterator);
 952                 goto end;
 953         }
 954
 955         assert(stream);
 956         stream_begin_notif = bt_notification_stream_begin_create(stream);
 957         if (!stream_begin_notif) {
 958                 BT_LOGE_STR("Cannot create stream beginning notification.");
 959                 goto error;
 960         }
 961
 962         add_action_push_notif(iterator, stream_begin_notif);
 963         BT_LOGV("Added \"push stream beginning notification\" action: "
 964                 "stream-addr=%p, stream-name=\"%s\"",
 965                 stream, bt_stream_get_name(stream));
 966         goto end;
 967
 968 error:
 969         ret = -1;
 970
 971 end:
 972         bt_put(stream_begin_notif);
 973         return ret;
 974 }
 975
 976 static
 977 int add_action_push_notif_stream_end(
 978                 struct bt_notification_iterator_private_connection *iterator,
 979                 struct bt_stream *stream)
 980 {
 981         int ret = 0;
 982         struct bt_notification *stream_end_notif = NULL;
 983
 984         if (!is_subscribed_to_notification_type(iterator,
 985                         BT_NOTIFICATION_TYPE_STREAM_END)) {
 986                 BT_LOGV("Not adding \"push stream end notification\" action: "
 987                         "notification iterator is not subscribed: addr=%p",
 988                         iterator);
 989                 goto end;
 990         }
 991
 992         assert(stream);
 993         stream_end_notif = bt_notification_stream_end_create(stream);
 994         if (!stream_end_notif) {
 995                 BT_LOGE_STR("Cannot create stream end notification.");
 996                 goto error;
 997         }
 998
 999         add_action_push_notif(iterator, stream_end_notif);
1000         BT_LOGV("Added \"push stream end notification\" action: "
1001                 "stream-addr=%p, stream-name=\"%s\"",
1002                 stream, bt_stream_get_name(stream));
1003         goto end;
1004
1005 error:
1006         ret = -1;
1007
1008 end:
1009         bt_put(stream_end_notif);
1010         return ret;
1011 }
1012
1013 static
1014 int add_action_push_notif_packet_begin(
1015                 struct bt_notification_iterator_private_connection *iterator,
1016                 struct bt_packet *packet)
1017 {
1018         int ret = 0;
1019         struct bt_notification *packet_begin_notif = NULL;
1020
1021         if (!is_subscribed_to_notification_type(iterator,
1022                         BT_NOTIFICATION_TYPE_PACKET_BEGIN)) {
1023                 BT_LOGV("Not adding \"push packet beginning notification\" action: "
1024                         "notification iterator is not subscribed: addr=%p",
1025                         iterator);
1026                 goto end;
1027         }
1028
1029         assert(packet);
1030         packet_begin_notif = bt_notification_packet_begin_create(packet);
1031         if (!packet_begin_notif) {
1032                 BT_LOGE_STR("Cannot create packet beginning notification.");
1033                 goto error;
1034         }
1035
1036         add_action_push_notif(iterator, packet_begin_notif);
1037         BT_LOGV("Added \"push packet beginning notification\" action: "
1038                 "packet-addr=%p", packet);
1039         goto end;
1040
1041 error:
1042         ret = -1;
1043
1044 end:
1045         bt_put(packet_begin_notif);
1046         return ret;
1047 }
1048
1049 static
1050 int add_action_push_notif_packet_end(
1051                 struct bt_notification_iterator_private_connection *iterator,
1052                 struct bt_packet *packet)
1053 {
1054         int ret = 0;
1055         struct bt_notification *packet_end_notif = NULL;
1056
1057         if (!is_subscribed_to_notification_type(iterator,
1058                         BT_NOTIFICATION_TYPE_PACKET_END)) {
1059                 BT_LOGV("Not adding \"push packet end notification\" action: "
1060                         "notification iterator is not subscribed: addr=%p",
1061                         iterator);
1062                 goto end;
1063         }
1064
1065         assert(packet);
1066         packet_end_notif = bt_notification_packet_end_create(packet);
1067         if (!packet_end_notif) {
1068                 BT_LOGE_STR("Cannot create packet end notification.");
1069                 goto error;
1070         }
1071
1072         add_action_push_notif(iterator, packet_end_notif);
1073         BT_LOGV("Added \"push packet end notification\" action: "
1074                 "packet-addr=%p", packet);
1075         goto end;
1076
1077 error:
1078         ret = -1;
1079
1080 end:
1081         bt_put(packet_end_notif);
1082         return ret;
1083 }
1084
1085 static
1086 void add_action_set_stream_state_is_ended(
1087                 struct bt_notification_iterator_private_connection *iterator,
1088                 struct stream_state *stream_state)
1089 {
1090         struct action action = {
1091                 .type = ACTION_TYPE_SET_STREAM_STATE_IS_ENDED,
1092                 .payload.set_stream_state_is_ended = {
1093                         .stream_state = stream_state,
1094                 },
1095         };
1096
1097         assert(stream_state);
1098         add_action(iterator, &action);
1099         BT_LOGV("Added \"set stream state's ended\" action: "
1100                 "stream-state-addr=%p", stream_state);
1101 }
1102
1103 static
1104 void add_action_set_stream_state_cur_packet(
1105                 struct bt_notification_iterator_private_connection *iterator,
1106                 struct stream_state *stream_state,
1107                 struct bt_packet *packet)
1108 {
1109         struct action action = {
1110                 .type = ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET,
1111                 .payload.set_stream_state_cur_packet = {
1112                         .stream_state = stream_state,
1113                         .packet = bt_get(packet),
1114                 },
1115         };
1116
1117         assert(stream_state);
1118         add_action(iterator, &action);
1119         BT_LOGV("Added \"set stream state's current packet\" action: "
1120                 "stream-state-addr=%p, packet-addr=%p",
1121                 stream_state, packet);
1122 }
1123
1124 static
1125 void add_action_update_stream_state_discarded_elements(
1126                 struct bt_notification_iterator_private_connection *iterator,
1127                 enum action_type type,
1128                 struct stream_state *stream_state,
1129                 struct bt_clock_value *cur_begin,
1130                 uint64_t cur_count)
1131 {
1132         struct action action = {
1133                 .type = type,
1134                 .payload.update_stream_state_discarded_elements = {
1135                         .stream_state = stream_state,
1136                         .cur_begin = bt_get(cur_begin),
1137                         .cur_count = cur_count,
1138                 },
1139         };
1140
1141         assert(stream_state);
1142         assert(type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS ||
1143                         type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS);
1144         add_action(iterator, &action);
1145         if (type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS) {
1146                 BT_LOGV("Added \"update stream state's discarded packets\" action: "
1147                         "stream-state-addr=%p, cur-begin-addr=%p, cur-count=%" PRIu64,
1148                         stream_state, cur_begin, cur_count);
1149         } else if (type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS) {
1150                 BT_LOGV("Added \"update stream state's discarded events\" action: "
1151                         "stream-state-addr=%p, cur-begin-addr=%p, cur-count=%" PRIu64,
1152                         stream_state, cur_begin, cur_count);
1153         }
1154 }
1155
1156 static
1157 int ensure_stream_state_exists(
1158                 struct bt_notification_iterator_private_connection *iterator,
1159                 struct bt_notification *stream_begin_notif,
1160                 struct bt_stream *notif_stream,
1161                 struct stream_state **_stream_state)
1162 {
1163         int ret = 0;
1164         struct stream_state *stream_state = NULL;
1165
1166         if (!notif_stream) {
1167                 /*
1168                  * The notification does not reference any stream: no
1169                  * need to get or create a stream state.
1170                  */
1171                 goto end;
1172         }
1173
1174         stream_state = g_hash_table_lookup(iterator->stream_states,
1175                 notif_stream);
1176         if (!stream_state) {
1177                 /*
1178                  * This iterator did not bump into this stream yet:
1179                  * create a stream state and a "stream begin"
1180                  * notification.
1181                  */
1182                 struct action action = {
1183                         .type = ACTION_TYPE_ADD_STREAM_STATE,
1184                         .payload.add_stream_state = {
1185                                 .stream = bt_get(notif_stream),
1186                                 .stream_state = NULL,
1187                         },
1188                 };
1189
1190                 stream_state = create_stream_state(notif_stream);
1191                 if (!stream_state) {
1192                         BT_LOGE_STR("Cannot create stream state.");
1193                         goto error;
1194                 }
1195
1196                 action.payload.add_stream_state.stream_state = stream_state;
1197                 add_action(iterator, &action);
1198
1199                 if (stream_begin_notif) {
1200                         add_action_push_notif(iterator, stream_begin_notif);
1201                 } else {
1202                         ret = add_action_push_notif_stream_begin(iterator,
1203                                 notif_stream);
1204                         if (ret) {
1205                                 BT_LOGE_STR("Cannot add \"push stream beginning notification\" action.");
1206                                 goto error;
1207                         }
1208                 }
1209         }
1210         goto end;
1211
1212 error:
1213         destroy_stream_state(stream_state);
1214         stream_state = NULL;
1215         ret = -1;
1216
1217 end:
1218         *_stream_state = stream_state;
1219         return ret;
1220 }
1221
1222 static
1223 struct bt_field *get_struct_field_uint(struct bt_field *struct_field,
1224                 const char *field_name)
1225 {
1226         struct bt_field *field = NULL;
1227         struct bt_field_type *ft = NULL;
1228
1229         field = bt_field_structure_get_field_by_name(struct_field,
1230                 field_name);
1231         if (!field) {
1232                 BT_LOGV_STR("`%s` field does not exist.");
1233                 goto end;
1234         }
1235
1236         if (!bt_field_is_integer(field)) {
1237                 BT_LOGV("Skipping `%s` field because its type is not an integer field type: "
1238                         "field-addr=%p, ft-addr=%p, ft-id=%s", field_name,
1239                         field, ft, bt_field_type_id_string(
1240                                 bt_field_type_get_type_id(ft)));
1241                 BT_PUT(field);
1242                 goto end;
1243         }
1244
1245         ft = bt_field_get_type(field);
1246         assert(ft);
1247
1248         if (bt_field_type_integer_is_signed(ft)) {
1249                 BT_LOGV("Skipping `%s` integer field because its type is signed: "
1250                         "field-addr=%p, ft-addr=%p", field_name, field, ft);
1251                 BT_PUT(field);
1252                 goto end;
1253         }
1254
1255 end:
1256         bt_put(ft);
1257         return field;
1258 }
1259
1260 static
1261 uint64_t get_packet_context_events_discarded(struct bt_packet *packet)
1262 {
1263         struct bt_field *packet_context = NULL;
1264         struct bt_field *field = NULL;
1265         uint64_t retval = -1ULL;
1266         int ret;
1267
1268         packet_context = bt_packet_get_context(packet);
1269         if (!packet_context) {
1270                 goto end;
1271         }
1272
1273         field = get_struct_field_uint(packet_context, "events_discarded");
1274         if (!field) {
1275                 BT_LOGV("`events_discarded` field does not exist in packet's context field: "
1276                         "packet-addr=%p, packet-context-field-addr=%p",
1277                         packet, packet_context);
1278                 goto end;
1279         }
1280
1281         assert(bt_field_is_integer(field));
1282         ret = bt_field_unsigned_integer_get_value(field, &retval);
1283         if (ret) {
1284                 BT_LOGV("Cannot get raw value of packet's context field's `events_discarded` integer field: "
1285                         "packet-addr=%p, field-addr=%p",
1286                         packet, field);
1287                 retval = -1ULL;
1288                 goto end;
1289         }
1290
1291 end:
1292         bt_put(packet_context);
1293         bt_put(field);
1294         return retval;
1295 }
1296
1297 static
1298 uint64_t get_packet_context_packet_seq_num(struct bt_packet *packet)
1299 {
1300         struct bt_field *packet_context = NULL;
1301         struct bt_field *field = NULL;
1302         uint64_t retval = -1ULL;
1303         int ret;
1304
1305         packet_context = bt_packet_get_context(packet);
1306         if (!packet_context) {
1307                 goto end;
1308         }
1309
1310         field = get_struct_field_uint(packet_context, "packet_seq_num");
1311         if (!field) {
1312                 BT_LOGV("`packet_seq_num` field does not exist in packet's context field: "
1313                         "packet-addr=%p, packet-context-field-addr=%p",
1314                         packet, packet_context);
1315                 goto end;
1316         }
1317
1318         assert(bt_field_is_integer(field));
1319         ret = bt_field_unsigned_integer_get_value(field, &retval);
1320         if (ret) {
1321                 BT_LOGV("Cannot get raw value of packet's context field's `packet_seq_num` integer field: "
1322                         "packet-addr=%p, field-addr=%p",
1323                         packet, field);
1324                 retval = -1ULL;
1325                 goto end;
1326         }
1327
1328 end:
1329         bt_put(packet_context);
1330         bt_put(field);
1331         return retval;
1332 }
1333
1334 static
1335 int handle_discarded_packets(
1336                 struct bt_notification_iterator_private_connection *iterator,
1337                 struct bt_packet *packet,
1338                 struct bt_clock_value *ts_begin,
1339                 struct bt_clock_value *ts_end,
1340                 struct stream_state *stream_state)
1341 {
1342         struct bt_notification *notif = NULL;
1343         uint64_t diff;
1344         uint64_t prev_count, next_count;
1345         int ret = 0;
1346
1347         next_count = get_packet_context_packet_seq_num(packet);
1348         if (next_count == -1ULL) {
1349                 /*
1350                  * Stream does not have seqnum field, skip discarded
1351                  * packets feature.
1352                  */
1353                 goto end;
1354         }
1355         prev_count = stream_state->discarded_packets_state.cur_count;
1356
1357         if (prev_count != -1ULL) {
1358                 if (next_count < prev_count) {
1359                         BT_LOGW("Current value of packet's context field's `packet_seq_num` field is lesser than the previous value for the same stream: "
1360                                 "not updating the stream state's current value: "
1361                                 "packet-addr=%p, prev-count=%" PRIu64 ", "
1362                                 "cur-count=%" PRIu64,
1363                                 packet, prev_count, next_count);
1364                         goto end;
1365                 }
1366                 if (next_count == prev_count) {
1367                         BT_LOGW("Current value of packet's context field's `packet_seq_num` field is equal to the previous value for the same stream: "
1368                                 "not updating the stream state's current value: "
1369                                 "packet-addr=%p, prev-count=%" PRIu64 ", "
1370                                 "cur-count=%" PRIu64,
1371                                 packet, prev_count, next_count);
1372                         goto end;
1373                 }
1374
1375                 diff = next_count - prev_count;
1376                 if (diff > 1) {
1377                         /*
1378                          * Add a discarded packets notification. The packets
1379                          * are considered to be lost between the state's last time
1380                          * and the current packet's beginning time.
1381                          * The counter is expected to monotonically increase of
1382                          * 1 for each packet. Therefore, the number of missing
1383                          * packets is 'diff - 1'.
1384                          */
1385                         notif = bt_notification_discarded_elements_create(
1386                                 BT_NOTIFICATION_TYPE_DISCARDED_PACKETS,
1387                                 stream_state->stream,
1388                                 stream_state->discarded_packets_state.cur_begin,
1389                                 ts_begin, diff - 1);
1390                         if (!notif) {
1391                                 BT_LOGE_STR("Cannot create discarded packets notification.");
1392                                 ret = -1;
1393                                 goto end;
1394                         }
1395
1396                         add_action_push_notif(iterator, notif);
1397                 }
1398         }
1399
1400         add_action_update_stream_state_discarded_elements(iterator,
1401                 ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS,
1402                 stream_state, ts_end, next_count);
1403
1404 end:
1405         bt_put(notif);
1406         return ret;
1407 }
1408
1409 static
1410 int handle_discarded_events(
1411                 struct bt_notification_iterator_private_connection *iterator,
1412                 struct bt_packet *packet,
1413                 struct bt_clock_value *ts_begin,
1414                 struct bt_clock_value *ts_end,
1415                 struct stream_state *stream_state)
1416 {
1417         struct bt_notification *notif = NULL;
1418         uint64_t diff;
1419         uint64_t next_count;
1420         int ret = 0;
1421
1422         next_count = get_packet_context_events_discarded(packet);
1423         if (next_count == -1ULL) {
1424                 next_count = stream_state->discarded_events_state.cur_count;
1425                 goto update_state;
1426         }
1427
1428         if (next_count < stream_state->discarded_events_state.cur_count) {
1429                 BT_LOGW("Current value of packet's context field's `events_discarded` field is lesser than the previous value for the same stream: "
1430                         "not updating the stream state's current value: "
1431                         "packet-addr=%p, prev-count=%" PRIu64 ", "
1432                         "cur-count=%" PRIu64,
1433                         packet, stream_state->discarded_events_state.cur_count,
1434                         next_count);
1435                 goto end;
1436         }
1437
1438         diff = next_count - stream_state->discarded_events_state.cur_count;
1439         if (diff > 0) {
1440                 /*
1441                  * Add a discarded events notification. The events are
1442                  * considered to be lost betweem the state's last time
1443                  * and the current packet's end time.
1444                  */
1445                 notif = bt_notification_discarded_elements_create(
1446                         BT_NOTIFICATION_TYPE_DISCARDED_EVENTS,
1447                         stream_state->stream,
1448                         stream_state->discarded_events_state.cur_begin,
1449                         ts_end, diff);
1450                 if (!notif) {
1451                         BT_LOGE_STR("Cannot create discarded events notification.");
1452                         ret = -1;
1453                         goto end;
1454                 }
1455
1456                 add_action_push_notif(iterator, notif);
1457         }
1458
1459 update_state:
1460         add_action_update_stream_state_discarded_elements(iterator,
1461                 ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS,
1462                 stream_state, ts_end, next_count);
1463
1464 end:
1465         bt_put(notif);
1466         return ret;
1467 }
1468
1469 static
1470 int get_field_clock_value(struct bt_field *root_field,
1471                 const char *field_name,
1472                 struct bt_clock_value **user_clock_val)
1473 {
1474         struct bt_field *field;
1475         struct bt_field_type *ft = NULL;
1476         struct bt_clock_class *clock_class = NULL;
1477         struct bt_clock_value *clock_value = NULL;
1478         uint64_t val;
1479         int ret = 0;
1480
1481         field = get_struct_field_uint(root_field, field_name);
1482         if (!field) {
1483                 /* Not an error: skip this */
1484                 goto end;
1485         }
1486
1487         ft = bt_field_get_type(field);
1488         assert(ft);
1489         clock_class = bt_field_type_integer_get_mapped_clock_class(ft);
1490         if (!clock_class) {
1491                 BT_LOGW("Integer field type has no mapped clock class but it's expected to have one: "
1492                         "ft-addr=%p", ft);
1493                 ret = -1;
1494                 goto end;
1495         }
1496
1497         ret = bt_field_unsigned_integer_get_value(field, &val);
1498         if (ret) {
1499                 BT_LOGW("Cannot get integer field's raw value: "
1500                         "field-addr=%p", field);
1501                 ret = -1;
1502                 goto end;
1503         }
1504
1505         clock_value = bt_clock_value_create(clock_class, val);
1506         if (!clock_value) {
1507                 BT_LOGE_STR("Cannot create clock value from clock class.");
1508                 ret = -1;
1509                 goto end;
1510         }
1511
1512         /* Move clock value to user */
1513         *user_clock_val = clock_value;
1514         clock_value = NULL;
1515
1516 end:
1517         bt_put(field);
1518         bt_put(ft);
1519         bt_put(clock_class);
1520         bt_put(clock_value);
1521         return ret;
1522 }
1523
1524 static
1525 int get_ts_begin_ts_end_from_packet(struct bt_packet *packet,
1526                 struct bt_clock_value **user_ts_begin,
1527                 struct bt_clock_value **user_ts_end)
1528 {
1529         struct bt_field *packet_context = NULL;
1530         struct bt_clock_value *ts_begin = NULL;
1531         struct bt_clock_value *ts_end = NULL;
1532         int ret = 0;
1533
1534         packet_context = bt_packet_get_context(packet);
1535         if (!packet_context) {
1536                 goto end;
1537         }
1538
1539         ret = get_field_clock_value(packet_context, "timestamp_begin",
1540                 &ts_begin);
1541         if (ret) {
1542                 BT_LOGW("Cannot create clock value for packet context's `timestamp_begin` field: "
1543                         "packet-addr=%p, packet-context-field-addr=%p",
1544                         packet, packet_context);
1545                 goto end;
1546         }
1547
1548         ret = get_field_clock_value(packet_context, "timestamp_end",
1549                 &ts_end);
1550         if (ret) {
1551                 BT_LOGW("Cannot create clock value for packet context's `timestamp_begin` field: "
1552                         "packet-addr=%p, packet-context-field-addr=%p",
1553                         packet, packet_context);
1554                 goto end;
1555         }
1556
1557         /* Move clock values to user */
1558         *user_ts_begin = ts_begin;
1559         ts_begin = NULL;
1560         *user_ts_end = ts_end;
1561         ts_end = NULL;
1562
1563 end:
1564         bt_put(packet_context);
1565         bt_put(ts_begin);
1566         bt_put(ts_end);
1567         return ret;
1568 }
1569
1570 static
1571 int handle_discarded_elements(
1572                 struct bt_notification_iterator_private_connection *iterator,
1573                 struct bt_packet *packet, struct stream_state *stream_state)
1574 {
1575         struct bt_clock_value *ts_begin = NULL;
1576         struct bt_clock_value *ts_end = NULL;
1577         int ret;
1578
1579         ret = get_ts_begin_ts_end_from_packet(packet, &ts_begin, &ts_end);
1580         if (ret) {
1581                 BT_LOGW("Cannot get packet's beginning or end clock values: "
1582                         "packet-addr=%p, ret=%d", packet, ret);
1583                 ret = -1;
1584                 goto end;
1585         }
1586
1587         ret = handle_discarded_packets(iterator, packet, ts_begin, ts_end,
1588                 stream_state);
1589         if (ret) {
1590                 BT_LOGW("Cannot handle discarded packets for packet: "
1591                         "packet-addr=%p, ret=%d", packet, ret);
1592                 ret = -1;
1593                 goto end;
1594         }
1595
1596         ret = handle_discarded_events(iterator, packet, ts_begin, ts_end,
1597                 stream_state);
1598         if (ret) {
1599                 BT_LOGW("Cannot handle discarded events for packet: "
1600                         "packet-addr=%p, ret=%d", packet, ret);
1601                 ret = -1;
1602                 goto end;
1603         }
1604
1605 end:
1606         bt_put(ts_begin);
1607         bt_put(ts_end);
1608         return ret;
1609 }
1610
1611 static
1612 int handle_packet_switch(
1613                 struct bt_notification_iterator_private_connection *iterator,
1614                 struct bt_notification *packet_begin_notif,
1615                 struct bt_packet *new_packet,
1616                 struct stream_state *stream_state)
1617 {
1618         int ret = 0;
1619
1620         if (stream_state->cur_packet == new_packet) {
1621                 goto end;
1622         }
1623
1624         BT_LOGV("Handling packet switch: "
1625                 "cur-packet-addr=%p, new-packet-addr=%p",
1626                 stream_state->cur_packet, new_packet);
1627
1628         if (stream_state->cur_packet) {
1629                 /* End of the current packet */
1630                 ret = add_action_push_notif_packet_end(iterator,
1631                         stream_state->cur_packet);
1632                 if (ret) {
1633                         BT_LOGE_STR("Cannot add \"push packet end notification\" action.");
1634                         goto error;
1635                 }
1636         }
1637
1638         /*
1639          * Check the new packet's context fields for discarded packets
1640          * and events to emit those automatic notifications.
1641          */
1642         ret = handle_discarded_elements(iterator, new_packet, stream_state);
1643         if (ret) {
1644                 BT_LOGE_STR("Cannot handle discarded elements for new packet.");
1645                 goto error;
1646         }
1647
1648         /* Beginning of the new packet */
1649         if (packet_begin_notif) {
1650                 add_action_push_notif(iterator, packet_begin_notif);
1651         } else if (new_packet) {
1652                 ret = add_action_push_notif_packet_begin(iterator,
1653                         new_packet);
1654                 if (ret) {
1655                         BT_LOGE_STR("Cannot add \"push packet beginning notification\" action.");
1656                         goto error;
1657                 }
1658         }
1659
1660         add_action_set_stream_state_cur_packet(iterator, stream_state,
1661                 new_packet);
1662         goto end;
1663
1664 error:
1665         ret = -1;
1666
1667 end:
1668         return ret;
1669 }
1670
1671 static
1672 int handle_notif_stream_begin(
1673                 struct bt_notification_iterator_private_connection *iterator,
1674                 struct bt_notification *notif,
1675                 struct bt_stream *notif_stream)
1676 {
1677         int ret = 0;
1678         struct stream_state *stream_state;
1679
1680         assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_BEGIN);
1681         assert(notif_stream);
1682         ret = ensure_stream_state_exists(iterator, notif, notif_stream,
1683                 &stream_state);
1684         if (ret) {
1685                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1686                 goto error;
1687         }
1688
1689         goto end;
1690
1691 error:
1692         ret = -1;
1693
1694 end:
1695         return ret;
1696 }
1697
1698 static
1699 int handle_notif_stream_end(
1700                 struct bt_notification_iterator_private_connection *iterator,
1701                 struct bt_notification *notif,
1702                 struct bt_stream *notif_stream)
1703 {
1704         int ret = 0;
1705         struct stream_state *stream_state;
1706
1707         assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_END);
1708         assert(notif_stream);
1709         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1710                 &stream_state);
1711         if (ret) {
1712                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1713                 goto error;
1714         }
1715
1716         ret = handle_packet_switch(iterator, NULL, NULL, stream_state);
1717         if (ret) {
1718                 BT_LOGE_STR("Cannot handle packet switch.");
1719                 goto error;
1720         }
1721
1722         add_action_push_notif(iterator, notif);
1723         add_action_set_stream_state_is_ended(iterator, stream_state);
1724         goto end;
1725
1726 error:
1727         ret = -1;
1728
1729 end:
1730         return ret;
1731 }
1732
1733 static
1734 int handle_notif_discarded_elements(
1735                 struct bt_notification_iterator_private_connection *iterator,
1736                 struct bt_notification *notif,
1737                 struct bt_stream *notif_stream)
1738 {
1739         int ret = 0;
1740         struct stream_state *stream_state;
1741
1742         assert(notif->type == BT_NOTIFICATION_TYPE_DISCARDED_EVENTS ||
1743                         notif->type == BT_NOTIFICATION_TYPE_DISCARDED_PACKETS);
1744         assert(notif_stream);
1745         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1746                 &stream_state);
1747         if (ret) {
1748                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1749                 goto error;
1750         }
1751
1752         add_action_push_notif(iterator, notif);
1753         goto end;
1754
1755 error:
1756         ret = -1;
1757
1758 end:
1759         return ret;
1760 }
1761
1762 static
1763 int handle_notif_packet_begin(
1764                 struct bt_notification_iterator_private_connection *iterator,
1765                 struct bt_notification *notif,
1766                 struct bt_stream *notif_stream,
1767                 struct bt_packet *notif_packet)
1768 {
1769         int ret = 0;
1770         struct stream_state *stream_state;
1771
1772         assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_BEGIN);
1773         assert(notif_packet);
1774         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1775                 &stream_state);
1776         if (ret) {
1777                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1778                 goto error;
1779         }
1780
1781         ret = handle_packet_switch(iterator, notif, notif_packet, stream_state);
1782         if (ret) {
1783                 BT_LOGE_STR("Cannot handle packet switch.");
1784                 goto error;
1785         }
1786
1787         goto end;
1788
1789 error:
1790         ret = -1;
1791
1792 end:
1793         return ret;
1794 }
1795
1796 static
1797 int handle_notif_packet_end(
1798                 struct bt_notification_iterator_private_connection *iterator,
1799                 struct bt_notification *notif,
1800                 struct bt_stream *notif_stream,
1801                 struct bt_packet *notif_packet)
1802 {
1803         int ret = 0;
1804         struct stream_state *stream_state;
1805
1806         assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_END);
1807         assert(notif_packet);
1808         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1809                 &stream_state);
1810         if (ret) {
1811                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1812                 goto error;
1813         }
1814
1815         ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state);
1816         if (ret) {
1817                 BT_LOGE_STR("Cannot handle packet switch.");
1818                 goto error;
1819         }
1820
1821         /* End of the current packet */
1822         add_action_push_notif(iterator, notif);
1823         add_action_set_stream_state_cur_packet(iterator, stream_state, NULL);
1824         goto end;
1825
1826 error:
1827         ret = -1;
1828
1829 end:
1830         return ret;
1831 }
1832
1833 static
1834 int handle_notif_event(
1835                 struct bt_notification_iterator_private_connection *iterator,
1836                 struct bt_notification *notif,
1837                 struct bt_stream *notif_stream,
1838                 struct bt_packet *notif_packet)
1839 {
1840         int ret = 0;
1841         struct stream_state *stream_state;
1842
1843         assert(notif->type == BT_NOTIFICATION_TYPE_EVENT);
1844         assert(notif_packet);
1845         ret = ensure_stream_state_exists(iterator, NULL, notif_stream,
1846                 &stream_state);
1847         if (ret) {
1848                 BT_LOGE_STR("Cannot ensure that stream state exists.");
1849                 goto error;
1850         }
1851
1852         ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state);
1853         if (ret) {
1854                 BT_LOGE_STR("Cannot handle packet switch.");
1855                 goto error;
1856         }
1857
1858         add_action_push_notif(iterator, notif);
1859         goto end;
1860
1861 error:
1862         ret = -1;
1863
1864 end:
1865         return ret;
1866 }
1867
1868 static
1869 int enqueue_notification_and_automatic(
1870                 struct bt_notification_iterator_private_connection *iterator,
1871                 struct bt_notification *notif)
1872 {
1873         int ret = 0;
1874         struct bt_event *notif_event = NULL;
1875         struct bt_stream *notif_stream = NULL;
1876         struct bt_packet *notif_packet = NULL;
1877
1878         assert(notif);
1879
1880         BT_LOGV("Enqueuing user notification and automatic notifications: "
1881                 "iter-addr=%p, notif-addr=%p", iterator, notif);
1882
1883         // TODO: Skip most of this if the iterator is only subscribed
1884         //       to event/inactivity notifications.
1885
1886         /* Get the stream and packet referred by the notification */
1887         switch (notif->type) {
1888         case BT_NOTIFICATION_TYPE_EVENT:
1889                 notif_event = bt_notification_event_borrow_event(notif);
1890                 assert(notif_event);
1891                 notif_packet = bt_event_borrow_packet(notif_event);
1892                 assert(notif_packet);
1893                 break;
1894         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
1895                 notif_stream =
1896                         bt_notification_stream_begin_borrow_stream(notif);
1897                 assert(notif_stream);
1898                 break;
1899         case BT_NOTIFICATION_TYPE_STREAM_END:
1900                 notif_stream = bt_notification_stream_end_borrow_stream(notif);
1901                 assert(notif_stream);
1902                 break;
1903         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
1904                 notif_packet =
1905                         bt_notification_packet_begin_borrow_packet(notif);
1906                 assert(notif_packet);
1907                 break;
1908         case BT_NOTIFICATION_TYPE_PACKET_END:
1909                 notif_packet = bt_notification_packet_end_borrow_packet(notif);
1910                 assert(notif_packet);
1911                 break;
1912         case BT_NOTIFICATION_TYPE_INACTIVITY:
1913                 /* Always valid */
1914                 goto handle_notif;
1915         default:
1916                 /*
1917                  * Invalid type of notification. Only the notification
1918                  * types above are allowed to be returned by a user
1919                  * component.
1920                  */
1921                 BT_LOGF("Unexpected notification type at this point: "
1922                         "notif-addr=%p, notif-type=%s", notif,
1923                         bt_notification_type_string(notif->type));
1924                 abort();
1925         }
1926
1927         if (notif_packet) {
1928                 notif_stream = bt_packet_borrow_stream(notif_packet);
1929                 assert(notif_stream);
1930         }
1931
1932         if (!notif_stream) {
1933                 /*
1934                  * The notification has no reference to a stream: it
1935                  * cannot cause the creation of automatic notifications.
1936                  */
1937                 BT_LOGV_STR("Notification has no reference to any stream: skipping automatic notification generation.");
1938                 goto end;
1939         }
1940
1941         if (!validate_notification(iterator, notif, notif_stream,
1942                         notif_packet)) {
1943                 BT_LOGW_STR("Invalid notification.");
1944                 goto error;
1945         }
1946
1947 handle_notif:
1948         switch (notif->type) {
1949         case BT_NOTIFICATION_TYPE_EVENT:
1950                 ret = handle_notif_event(iterator, notif, notif_stream,
1951                         notif_packet);
1952                 break;
1953         case BT_NOTIFICATION_TYPE_STREAM_BEGIN:
1954                 ret = handle_notif_stream_begin(iterator, notif, notif_stream);
1955                 break;
1956         case BT_NOTIFICATION_TYPE_STREAM_END:
1957                 ret = handle_notif_stream_end(iterator, notif, notif_stream);
1958                 break;
1959         case BT_NOTIFICATION_TYPE_PACKET_BEGIN:
1960                 ret = handle_notif_packet_begin(iterator, notif, notif_stream,
1961                         notif_packet);
1962                 break;
1963         case BT_NOTIFICATION_TYPE_PACKET_END:
1964                 ret = handle_notif_packet_end(iterator, notif, notif_stream,
1965                         notif_packet);
1966                 break;
1967         case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS:
1968         case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS:
1969                 ret = handle_notif_discarded_elements(iterator, notif,
1970                         notif_stream);
1971                 break;
1972         case BT_NOTIFICATION_TYPE_INACTIVITY:
1973                 add_action_push_notif(iterator, notif);
1974                 break;
1975         default:
1976                 break;
1977         }
1978
1979         if (ret) {
1980                 BT_LOGW_STR("Failed to handle notification for automatic notification generation.");
1981                 goto error;
1982         }
1983
1984         apply_actions(iterator);
1985         BT_LOGV("Enqueued user notification and automatic notifications: "
1986                 "iter-addr=%p, notif-addr=%p", iterator, notif);
1987         goto end;
1988
1989 error:
1990         ret = -1;
1991
1992 end:
1993         return ret;
1994 }
1995
1996 static
1997 int handle_end(struct bt_notification_iterator_private_connection *iterator)
1998 {
1999         GHashTableIter stream_state_iter;
2000         gpointer stream_gptr, stream_state_gptr;
2001         int ret = 0;
2002
2003         BT_LOGV("Handling end of iteration: addr=%p", iterator);
2004
2005         /*
2006          * Emit a "stream end" notification for each non-ended stream
2007          * known by this iterator and mark them as ended.
2008          */
2009         g_hash_table_iter_init(&stream_state_iter, iterator->stream_states);
2010
2011         while (g_hash_table_iter_next(&stream_state_iter, &stream_gptr,
2012                         &stream_state_gptr)) {
2013                 struct stream_state *stream_state = stream_state_gptr;
2014
2015                 assert(stream_state_gptr);
2016
2017                 if (stream_state->is_ended) {
2018                         continue;
2019                 }
2020
2021                 ret = handle_packet_switch(iterator, NULL, NULL, stream_state);
2022                 if (ret) {
2023                         BT_LOGE_STR("Cannot handle packet switch.");
2024                         goto error;
2025                 }
2026
2027                 ret = add_action_push_notif_stream_end(iterator, stream_gptr);
2028                 if (ret) {
2029                         BT_LOGE_STR("Cannot add \"push stream end notification\" action.");
2030                         goto error;
2031                 }
2032
2033                 add_action_set_stream_state_is_ended(iterator, stream_state);
2034         }
2035
2036         apply_actions(iterator);
2037         BT_LOGV("Handled end of iteration: addr=%p", iterator);
2038         goto end;
2039
2040 error:
2041         ret = -1;
2042
2043 end:
2044         return ret;
2045 }
2046
2047 static
2048 enum bt_notification_iterator_status ensure_queue_has_notifications(
2049                 struct bt_notification_iterator_private_connection *iterator)
2050 {
2051         struct bt_private_connection_private_notification_iterator *priv_iterator =
2052                 bt_private_connection_private_notification_iterator_from_notification_iterator(iterator);
2053         bt_component_class_notification_iterator_next_method next_method = NULL;
2054         struct bt_notification_iterator_next_method_return next_return = {
2055                 .status = BT_NOTIFICATION_ITERATOR_STATUS_OK,
2056                 .notification = NULL,
2057         };
2058         enum bt_notification_iterator_status status =
2059                 BT_NOTIFICATION_ITERATOR_STATUS_OK;
2060         int ret;
2061
2062         assert(iterator);
2063         BT_LOGD("Ensuring that notification iterator's queue has at least one notification: "
2064                 "iter-addr=%p, queue-size=%u, iter-state=%s",
2065                 iterator, iterator->queue->length,
2066                 bt_private_connection_notification_iterator_state_string(iterator->state));
2067
2068         if (iterator->queue->length > 0) {
2069                 /*
2070                  * We already have enough. Even if this notification
2071                  * iterator is finalized, its user can still flush its
2072                  * current queue's content by calling its "next" method
2073                  * since this content is local and has no impact on what
2074                  * used to be the iterator's upstream component.
2075                  */
2076                 BT_LOGD_STR("Queue already has at least one notification.");
2077                 goto end;
2078         }
2079
2080         switch (iterator->state) {
2081         case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED:
2082         case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED:
2083                 BT_LOGD_STR("Notification iterator's \"next\" called, but it is finalized.");
2084                 status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
2085                 goto end;
2086         case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ENDED:
2087                 BT_LOGD_STR("Notification iterator is ended.");
2088                 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
2089                 goto end;
2090         default:
2091                 break;
2092         }
2093
2094         assert(iterator->upstream_component);
2095         assert(iterator->upstream_component->class);
2096
2097         /* Pick the appropriate "next" method */
2098         switch (iterator->upstream_component->class->type) {
2099         case BT_COMPONENT_CLASS_TYPE_SOURCE:
2100         {
2101                 struct bt_component_class_source *source_class =
2102                         container_of(iterator->upstream_component->class,
2103                                 struct bt_component_class_source, parent);
2104
2105                 assert(source_class->methods.iterator.next);
2106                 next_method = source_class->methods.iterator.next;
2107                 break;
2108         }
2109         case BT_COMPONENT_CLASS_TYPE_FILTER:
2110         {
2111                 struct bt_component_class_filter *filter_class =
2112                         container_of(iterator->upstream_component->class,
2113                                 struct bt_component_class_filter, parent);
2114
2115                 assert(filter_class->methods.iterator.next);
2116                 next_method = filter_class->methods.iterator.next;
2117                 break;
2118         }
2119         default:
2120                 abort();
2121         }
2122
2123         /*
2124          * Call the user's "next" method to get the next notification
2125          * and status.
2126          */
2127         assert(next_method);
2128
2129         while (iterator->queue->length == 0) {
2130                 BT_LOGD_STR("Calling user's \"next\" method.");
2131                 next_return = next_method(priv_iterator);
2132                 BT_LOGD("User method returned: status=%s",
2133                         bt_notification_iterator_status_string(next_return.status));
2134                 if (next_return.status < 0) {
2135                         BT_LOGW_STR("User method failed.");
2136                         status = next_return.status;
2137                         goto end;
2138                 }
2139
2140                 if (iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED ||
2141                                 iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED) {
2142                         /*
2143                          * The user's "next" method, somehow, cancelled
2144                          * its own notification iterator. This can
2145                          * happen, for example, when the user's method
2146                          * removes the port on which there's the
2147                          * connection from which the iterator was
2148                          * created. In this case, said connection is
2149                          * ended, and all its notification iterators are
2150                          * finalized.
2151                          *
2152                          * Only bt_put() the returned notification if
2153                          * the status is
2154                          * BT_NOTIFICATION_ITERATOR_STATUS_OK because
2155                          * otherwise this field could be garbage.
2156                          */
2157                         if (next_return.status ==
2158                                         BT_NOTIFICATION_ITERATOR_STATUS_OK) {
2159                                 bt_put(next_return.notification);
2160                         }
2161
2162                         status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
2163                         goto end;
2164                 }
2165
2166                 switch (next_return.status) {
2167                 case BT_NOTIFICATION_ITERATOR_STATUS_END:
2168                         ret = handle_end(iterator);
2169                         if (ret) {
2170                                 BT_LOGW_STR("Cannot handle end of iteration.");
2171                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
2172                                 goto end;
2173                         }
2174
2175                         assert(iterator->state ==
2176                                 BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ACTIVE);
2177                         iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ENDED;
2178
2179                         if (iterator->queue->length == 0) {
2180                                 status = BT_NOTIFICATION_ITERATOR_STATUS_END;
2181                         }
2182
2183                         BT_LOGD("Set new status: status=%s",
2184                                 bt_notification_iterator_status_string(status));
2185                         goto end;
2186                 case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN:
2187                         status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN;
2188                         goto end;
2189                 case BT_NOTIFICATION_ITERATOR_STATUS_OK:
2190                         if (!next_return.notification) {
2191                                 BT_LOGW_STR("User method returned BT_NOTIFICATION_ITERATOR_STATUS_OK, but notification is NULL.");
2192                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
2193                                 goto end;
2194                         }
2195
2196                         /*
2197                          * Ignore some notifications which are always
2198                          * automatically generated by the notification
2199                          * iterator to make sure they have valid values.
2200                          */
2201                         switch (next_return.notification->type) {
2202                         case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS:
2203                         case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS:
2204                                 BT_LOGV("Ignoring discarded elements notification returned by notification iterator's \"next\" method: "
2205                                         "notif-type=%s",
2206                                         bt_notification_type_string(next_return.notification->type));
2207                                 BT_PUT(next_return.notification);
2208                                 continue;
2209                         default:
2210                                 break;
2211                         }
2212
2213                         /*
2214                          * We know the notification is valid. Before we
2215                          * push it to the head of the queue, push the
2216                          * appropriate automatic notifications if any.
2217                          */
2218                         ret = enqueue_notification_and_automatic(iterator,
2219                                 next_return.notification);
2220                         BT_PUT(next_return.notification);
2221                         if (ret) {
2222                                 BT_LOGW("Cannot enqueue notification and automatic notifications.");
2223                                 status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
2224                                 goto end;
2225                         }
2226                         break;
2227                 default:
2228                         /* Unknown non-error status */
2229                         abort();
2230                 }
2231         }
2232
2233 end:
2234         return status;
2235 }
2236
2237 enum bt_notification_iterator_status
2238 bt_notification_iterator_next(struct bt_notification_iterator *iterator)
2239 {
2240         enum bt_notification_iterator_status status;
2241
2242         if (!iterator) {
2243                 BT_LOGW_STR("Invalid parameter: notification iterator is NULL.");
2244                 status = BT_NOTIFICATION_ITERATOR_STATUS_INVALID;
2245                 goto end;
2246         }
2247
2248         BT_LOGD("Notification iterator's \"next\": iter-addr=%p", iterator);
2249
2250         switch (iterator->type) {
2251         case BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION:
2252         {
2253                 struct bt_notification_iterator_private_connection *priv_conn_iter =
2254                         (void *) iterator;
2255                 struct bt_notification *notif;
2256
2257                 /*
2258                  * Make sure that the iterator's queue contains at least
2259                  * one notification.
2260                  */
2261                 status = ensure_queue_has_notifications(priv_conn_iter);
2262                 if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
2263                         goto end;
2264                 }
2265
2266                 /*
2267                  * Move the notification at the tail of the queue to the
2268                  * iterator's current notification.
2269                  */
2270                 assert(priv_conn_iter->queue->length > 0);
2271                 notif = g_queue_pop_tail(priv_conn_iter->queue);
2272                 bt_notification_iterator_replace_current_notification(
2273                         iterator, notif);
2274                 bt_put(notif);
2275                 break;
2276         }
2277         case BT_NOTIFICATION_ITERATOR_TYPE_OUTPUT_PORT:
2278         {
2279                 struct bt_notification_iterator_output_port *out_port_iter =
2280                         (void *) iterator;
2281
2282                 /*
2283                  * Keep current notification in case there's an error:
2284                  * restore this notification so that the current
2285                  * notification is not changed from the user's point of
2286                  * view.
2287                  */
2288                 struct bt_notification *old_notif =
2289                         bt_get(bt_notification_iterator_borrow_current_notification(iterator));
2290                 enum bt_graph_status graph_status;
2291
2292                 /*
2293                  * Put current notification since it's possibly
2294                  * about to be replaced by a new one by the
2295                  * colander sink.
2296                  */
2297                 bt_notification_iterator_replace_current_notification(
2298                         iterator, NULL);
2299                 graph_status = bt_graph_consume_sink_no_check(
2300                         out_port_iter->graph,
2301                         out_port_iter->colander);
2302                 switch (graph_status) {
2303                 case BT_GRAPH_STATUS_CANCELED:
2304                         status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED;
2305                         break;
2306                 case BT_GRAPH_STATUS_AGAIN:
2307                         status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN;
2308                         break;
2309                 case BT_GRAPH_STATUS_END:
2310                         status = BT_NOTIFICATION_ITERATOR_STATUS_END;
2311                         break;
2312                 case BT_GRAPH_STATUS_NOMEM:
2313                         status = BT_NOTIFICATION_ITERATOR_STATUS_NOMEM;
2314                         break;
2315                 case BT_GRAPH_STATUS_OK:
2316                         status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
2317                         assert(bt_notification_iterator_borrow_current_notification(iterator));
2318                         break;
2319                 default:
2320                         /* Other errors */
2321                         status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
2322                 }
2323
2324                 if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) {
2325                         /* Error/exception: restore old notification */
2326                         bt_notification_iterator_replace_current_notification(
2327                                 iterator, old_notif);
2328                 }
2329
2330                 bt_put(old_notif);
2331                 break;
2332         }
2333         default:
2334                 BT_LOGF("Unknown notification iterator type: addr=%p, type=%d",
2335                         iterator, iterator->type);
2336                 abort();
2337         }
2338
2339 end:
2340         return status;
2341 }
2342
2343 struct bt_component *bt_private_connection_notification_iterator_get_component(
2344                 struct bt_notification_iterator *iterator)
2345 {
2346         struct bt_component *comp = NULL;
2347         struct bt_notification_iterator_private_connection *iter_priv_conn;
2348
2349         if (!iterator) {
2350                 BT_LOGW_STR("Invalid parameter: notification iterator is NULL.");
2351                 goto end;
2352         }
2353
2354         if (iterator->type != BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION) {
2355                 BT_LOGW_STR("Invalid parameter: notification iterator was not created from a private connection.");
2356                 goto end;
2357         }
2358
2359         iter_priv_conn = (void *) iterator;
2360         comp = bt_get(iter_priv_conn->upstream_component);
2361
2362 end:
2363         return comp;
2364 }
2365
2366 struct bt_private_component *
2367 bt_private_connection_private_notification_iterator_get_private_component(
2368                 struct bt_private_connection_private_notification_iterator *private_iterator)
2369 {
2370         return bt_private_component_from_component(
2371                 bt_private_connection_notification_iterator_get_component(
2372                         (void *) bt_private_connection_notification_iterator_borrow_from_private(private_iterator)));
2373 }
2374
2375 static
2376 void bt_output_port_notification_iterator_destroy(struct bt_object *obj)
2377 {
2378         struct bt_notification_iterator_output_port *iterator =
2379                 (void *) container_of(obj, struct bt_notification_iterator, base);
2380
2381         BT_LOGD("Destroying output port notification iterator object: addr=%p",
2382                 iterator);
2383         BT_LOGD_STR("Putting graph.");
2384         bt_put(iterator->graph);
2385         BT_LOGD_STR("Putting output port.");
2386         bt_put(iterator->output_port);
2387         BT_LOGD_STR("Putting colander sink component.");
2388         bt_put(iterator->colander);
2389         destroy_base_notification_iterator(obj);
2390 }
2391
2392 struct bt_notification_iterator *bt_output_port_notification_iterator_create(
2393                 struct bt_port *output_port,
2394                 const char *colander_component_name,
2395                 const enum bt_notification_type *notification_types)
2396 {
2397         struct bt_notification_iterator *iterator_base = NULL;
2398         struct bt_notification_iterator_output_port *iterator = NULL;
2399         struct bt_component_class *colander_comp_cls = NULL;
2400         struct bt_component *output_port_comp = NULL;
2401         struct bt_component *colander_comp;
2402         struct bt_graph *graph = NULL;
2403         enum bt_graph_status graph_status;
2404         const char *colander_comp_name;
2405         struct bt_port *colander_in_port = NULL;
2406         struct bt_component_class_sink_colander_data colander_data;
2407
2408         if (!output_port) {
2409                 BT_LOGW_STR("Invalid parameter: port is NULL.");
2410                 goto error;
2411         }
2412
2413         if (bt_port_get_type(output_port) != BT_PORT_TYPE_OUTPUT) {
2414                 BT_LOGW_STR("Invalid parameter: port is not an output port.");
2415                 goto error;
2416         }
2417
2418         output_port_comp = bt_port_get_component(output_port);
2419         if (!output_port_comp) {
2420                 BT_LOGW("Cannot get output port's component: "
2421                         "port-addr=%p, port-name=\"%s\"",
2422                         output_port, bt_port_get_name(output_port));
2423                 goto error;
2424         }
2425
2426         graph = bt_component_get_graph(output_port_comp);
2427         assert(graph);
2428
2429         /* Create notification iterator */
2430         BT_LOGD("Creating notification iterator on output port: "
2431                 "comp-addr=%p, comp-name\"%s\", port-addr=%p, port-name=\"%s\"",
2432                 output_port_comp, bt_component_get_name(output_port_comp),
2433                 output_port, bt_port_get_name(output_port));
2434         iterator = g_new0(struct bt_notification_iterator_output_port, 1);
2435         if (!iterator) {
2436                 BT_LOGE_STR("Failed to allocate one output port notification iterator.");
2437                 goto error;
2438         }
2439
2440         init_notification_iterator((void *) iterator,
2441                 BT_NOTIFICATION_ITERATOR_TYPE_OUTPUT_PORT,
2442                 bt_output_port_notification_iterator_destroy);
2443
2444         /* Create colander component */
2445         colander_comp_cls = bt_component_class_sink_colander_get();
2446         if (!colander_comp_cls) {
2447                 BT_LOGW("Cannot get colander sink component class.");
2448                 goto error;
2449         }
2450
2451         BT_MOVE(iterator->graph, graph);
2452         iterator_base = (void *) iterator;
2453         colander_comp_name =
2454                 colander_component_name ? colander_component_name : "colander";
2455         colander_data.notification = &iterator_base->current_notification;
2456         colander_data.notification_types = notification_types;
2457         graph_status = bt_graph_add_component_with_init_method_data(
2458                 iterator->graph, colander_comp_cls, colander_comp_name,
2459                 NULL, &colander_data, &iterator->colander);
2460         if (graph_status != BT_GRAPH_STATUS_OK) {
2461                 BT_LOGW("Cannot add colander sink component to graph: "
2462                         "graph-addr=%p, name=\"%s\", graph-status=%s",
2463                         iterator->graph, colander_comp_name,
2464                         bt_graph_status_string(graph_status));
2465                 goto error;
2466         }
2467
2468         /*
2469          * Connect provided output port to the colander component's
2470          * input port.
2471          */
2472         colander_in_port = bt_component_sink_get_input_port_by_index(
2473                 iterator->colander, 0);
2474         assert(colander_in_port);
2475         graph_status = bt_graph_connect_ports(iterator->graph,
2476                 output_port, colander_in_port, NULL);
2477         if (graph_status != BT_GRAPH_STATUS_OK) {
2478                 BT_LOGW("Cannot add colander sink component to graph: "
2479                         "graph-addr=%p, name=\"%s\", graph-status=%s",
2480                         iterator->graph, colander_comp_name,
2481                         bt_graph_status_string(graph_status));
2482                 goto error;
2483         }
2484
2485         /*
2486          * At this point everything went fine. Make the graph
2487          * nonconsumable forever so that only this notification iterator
2488          * can consume (thanks to bt_graph_consume_sink_no_check()).
2489          * This avoids leaking the notification created by the colander
2490          * sink and moved to the base notification iterator's current
2491          * notification member.
2492          */
2493         bt_graph_set_can_consume(iterator->graph, BT_FALSE);
2494         goto end;
2495
2496 error:
2497         if (iterator && iterator->graph && iterator->colander) {
2498                 int ret;
2499
2500                 /* Remove created colander component from graph if any */
2501                 colander_comp = iterator->colander;
2502                 BT_PUT(iterator->colander);
2503
2504                 /*
2505                  * At this point the colander component's reference
2506                  * count is 0 because iterator->colander was the only
2507                  * owner. We also know that it is not connected because
2508                  * this is the last operation before this function
2509                  * succeeds.
2510                  *
2511                  * Since we honor the preconditions here,
2512                  * bt_graph_remove_unconnected_component() always
2513                  * succeeds.
2514                  */
2515                 ret = bt_graph_remove_unconnected_component(iterator->graph,
2516                         colander_comp);
2517                 assert(ret == 0);
2518         }
2519
2520         BT_PUT(iterator);
2521
2522 end:
2523         bt_put(colander_in_port);
2524         bt_put(colander_comp_cls);
2525         bt_put(output_port_comp);
2526         bt_put(graph);
2527         return (void *) iterator;
2528 }
2529
2530 struct bt_notification_iterator *
2531 bt_private_connection_notification_iterator_from_private(
2532                 struct bt_private_connection_private_notification_iterator *private_notification_iterator)
2533 {
2534         return bt_get(
2535                 bt_private_connection_notification_iterator_borrow_from_private(
2536                         private_notification_iterator));
2537 }