X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=lib%2Fgraph%2Fiterator.c;h=1e02eb33c5eff3613ae6809a4d2c7322749cf205;hb=2d8aeedf7e6b830747a71bceaa0f8a35e97293b9;hp=c202087417188c7a5e42ff7c227bf1420383bb99;hpb=b7f20dbaad41b6124ab844961b4884f36fbd2b4c;p=babeltrace.git diff --git a/lib/graph/iterator.c b/lib/graph/iterator.c index c2020874..1e02eb33 100644 --- a/lib/graph/iterator.c +++ b/lib/graph/iterator.c @@ -1,10 +1,6 @@ /* - * iterator.c - * - * Babeltrace Notification Iterator - * + * Copyright 2017-2018 Philippe Proulx * Copyright 2015 Jérémie Galarneau - * Copyright 2017 Philippe Proulx * * Permission is hereby granted, free of charge, to any person obtaining a copy * of this software and associated documentation files (the "Software"), to deal @@ -25,414 +21,123 @@ * SOFTWARE. */ -#define BT_LOG_TAG "NOTIF-ITER" +#define BT_LOG_TAG "MSG-ITER" #include #include -#include -#include -#include -#include -#include -#include -#include -#include +#include +#include +#include +#include +#include +#include +#include +#include +#include #include -#include +#include +#include #include #include #include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include #include #include +#include +#include #include #include #include -struct discarded_elements_state { - struct bt_ctf_clock_value *cur_begin; - uint64_t cur_count; -}; - -struct stream_state { - struct bt_ctf_stream *stream; /* owned by this */ - struct bt_ctf_packet *cur_packet; /* owned by this */ - struct discarded_elements_state discarded_packets_state; - struct discarded_elements_state discarded_events_state; - bt_bool is_ended; -}; - -enum action_type { - ACTION_TYPE_PUSH_NOTIF, - ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM, - ACTION_TYPE_ADD_STREAM_STATE, - ACTION_TYPE_SET_STREAM_STATE_IS_ENDED, - ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET, - ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS, - ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS, -}; - -struct action { - enum action_type type; - union { - /* ACTION_TYPE_PUSH_NOTIF */ - struct { - struct bt_notification *notif; /* owned by this */ - } push_notif; - - /* ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM */ - struct { - struct bt_ctf_stream *stream; /* owned by this */ - struct bt_component *component; /* owned by this */ - struct bt_port *port; /* owned by this */ - } map_port_to_comp_in_stream; - - /* ACTION_TYPE_ADD_STREAM_STATE */ - struct { - struct bt_ctf_stream *stream; /* owned by this */ - struct stream_state *stream_state; /* owned by this */ - } add_stream_state; - - /* ACTION_TYPE_SET_STREAM_STATE_IS_ENDED */ - struct { - struct stream_state *stream_state; /* weak */ - } set_stream_state_is_ended; - - /* ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET */ - struct { - struct stream_state *stream_state; /* weak */ - struct bt_ctf_packet *packet; /* owned by this */ - } set_stream_state_cur_packet; - - /* ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS */ - /* ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS */ - struct { - struct stream_state *stream_state; /* weak */ - struct bt_ctf_clock_value *cur_begin; /* owned by this */ - uint64_t cur_count; - } update_stream_state_discarded_elements; - } payload; -}; - -static -void stream_destroy_listener(struct bt_ctf_stream *stream, void *data) -{ - struct bt_notification_iterator_private_connection *iterator = data; - - /* Remove associated stream state */ - g_hash_table_remove(iterator->stream_states, stream); -} - -static -void destroy_stream_state(struct stream_state *stream_state) -{ - if (!stream_state) { - return; - } - - BT_LOGV("Destroying stream state: stream-state-addr=%p", stream_state); - BT_LOGV_STR("Putting stream state's current packet."); - bt_put(stream_state->cur_packet); - BT_LOGV_STR("Putting stream state's stream."); - bt_put(stream_state->stream); - bt_put(stream_state->discarded_packets_state.cur_begin); - bt_put(stream_state->discarded_events_state.cur_begin); - g_free(stream_state); -} - -static -void destroy_action(struct action *action) -{ - assert(action); - - switch (action->type) { - case ACTION_TYPE_PUSH_NOTIF: - BT_PUT(action->payload.push_notif.notif); - break; - case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM: - BT_PUT(action->payload.map_port_to_comp_in_stream.stream); - BT_PUT(action->payload.map_port_to_comp_in_stream.component); - BT_PUT(action->payload.map_port_to_comp_in_stream.port); - break; - case ACTION_TYPE_ADD_STREAM_STATE: - BT_PUT(action->payload.add_stream_state.stream); - destroy_stream_state( - action->payload.add_stream_state.stream_state); - action->payload.add_stream_state.stream_state = NULL; - break; - case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET: - BT_PUT(action->payload.set_stream_state_cur_packet.packet); - break; - case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED: - break; - case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS: - case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS: - BT_PUT(action->payload.update_stream_state_discarded_elements.cur_begin); - break; - default: - BT_LOGF("Unexpected action's type: type=%d", action->type); - abort(); - } -} - -static -void add_action(struct bt_notification_iterator_private_connection *iterator, - struct action *action) -{ - g_array_append_val(iterator->actions, *action); -} - -static -void clear_actions(struct bt_notification_iterator_private_connection *iterator) -{ - size_t i; - - for (i = 0; i < iterator->actions->len; i++) { - struct action *action = &g_array_index(iterator->actions, - struct action, i); - - destroy_action(action); - } +/* + * TODO: Use graph's state (number of active iterators, etc.) and + * possibly system specifications to make a better guess than this. + */ +#define MSG_BATCH_SIZE 15 - g_array_set_size(iterator->actions, 0); -} +#define BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(_iter) \ + BT_ASSERT_PRE((_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE || \ + (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED || \ + (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN || \ + (_iter)->state == BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR, \ + "Message iterator is in the wrong state: %!+i", _iter) static inline -const char *action_type_string(enum action_type type) +void _set_self_comp_port_input_msg_iterator_state( + struct bt_self_component_port_input_message_iterator *iterator, + enum bt_self_component_port_input_message_iterator_state state) { - switch (type) { - case ACTION_TYPE_PUSH_NOTIF: - return "ACTION_TYPE_PUSH_NOTIF"; - case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM: - return "ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM"; - case ACTION_TYPE_ADD_STREAM_STATE: - return "ACTION_TYPE_ADD_STREAM_STATE"; - case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED: - return "ACTION_TYPE_SET_STREAM_STATE_IS_ENDED"; - case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET: - return "ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET"; - case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS: - return "ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS"; - case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS: - return "ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS"; - default: - return "(unknown)"; - } + BT_ASSERT(iterator); + BT_LIB_LOGD("Updating message iterator's state: new-state=%s", + bt_self_component_port_input_message_iterator_state_string(state)); + iterator->state = state; } -static -void apply_actions(struct bt_notification_iterator_private_connection *iterator) -{ - size_t i; - - BT_LOGV("Applying notification's iterator current actions: " - "count=%u", iterator->actions->len); - - for (i = 0; i < iterator->actions->len; i++) { - struct action *action = &g_array_index(iterator->actions, - struct action, i); - - BT_LOGV("Applying action: index=%zu, type=%s", - i, action_type_string(action->type)); - - switch (action->type) { - case ACTION_TYPE_PUSH_NOTIF: - /* Move notification to queue */ - g_queue_push_head(iterator->queue, - action->payload.push_notif.notif); - bt_notification_freeze( - action->payload.push_notif.notif); - action->payload.push_notif.notif = NULL; - break; - case ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM: - bt_ctf_stream_map_component_to_port( - action->payload.map_port_to_comp_in_stream.stream, - action->payload.map_port_to_comp_in_stream.component, - action->payload.map_port_to_comp_in_stream.port); - break; - case ACTION_TYPE_ADD_STREAM_STATE: - /* Move stream state to hash table */ - g_hash_table_insert(iterator->stream_states, - action->payload.add_stream_state.stream, - action->payload.add_stream_state.stream_state); - - action->payload.add_stream_state.stream_state = NULL; - break; - case ACTION_TYPE_SET_STREAM_STATE_IS_ENDED: - /* - * We know that this stream is ended. We need to - * remember this as long as the stream exists to - * enforce that the same stream does not end - * twice. - * - * Here we add a destroy listener to the stream - * which we put after (becomes weak as the hash - * table key). If we were the last object to own - * this stream, the destroy listener is called - * when we call bt_put() which removes this - * stream state completely. This is important - * because the memory used by this stream object - * could be reused for another stream, and they - * must have different states. - */ - bt_ctf_stream_add_destroy_listener( - action->payload.set_stream_state_is_ended.stream_state->stream, - stream_destroy_listener, iterator); - action->payload.set_stream_state_is_ended.stream_state->is_ended = BT_TRUE; - BT_PUT(action->payload.set_stream_state_is_ended.stream_state->stream); - break; - case ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET: - /* Move packet to stream state's current packet */ - BT_MOVE(action->payload.set_stream_state_cur_packet.stream_state->cur_packet, - action->payload.set_stream_state_cur_packet.packet); - break; - case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS: - case ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS: - { - struct discarded_elements_state *state; - - if (action->type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS) { - state = &action->payload.update_stream_state_discarded_elements.stream_state->discarded_packets_state; - } else { - state = &action->payload.update_stream_state_discarded_elements.stream_state->discarded_events_state; - } - - BT_MOVE(state->cur_begin, - action->payload.update_stream_state_discarded_elements.cur_begin); - state->cur_count = action->payload.update_stream_state_discarded_elements.cur_count; - break; - } - default: - BT_LOGF("Unexpected action's type: type=%d", - action->type); - abort(); - } - } - - clear_actions(iterator); -} +#ifdef BT_DEV_MODE +# define set_self_comp_port_input_msg_iterator_state _set_self_comp_port_input_msg_iterator_state +#else +# define set_self_comp_port_input_msg_iterator_state(_a, _b) ((void) _a); ((void) _b); +#endif static -struct stream_state *create_stream_state(struct bt_ctf_stream *stream) +void destroy_base_message_iterator(struct bt_object *obj) { - struct stream_state *stream_state = g_new0(struct stream_state, 1); - - if (!stream_state) { - BT_LOGE_STR("Failed to allocate one stream state."); - goto end; - } - - /* - * The packet index is a monotonic counter which may not start - * at 0 at the beginning of the stream. We therefore need to - * have an internal object initial state of -1ULL to distinguish - * between initial state and having seen a packet with - * seqnum = 0. - */ - stream_state->discarded_packets_state.cur_count = -1ULL; - - /* - * We keep a reference to the stream until we know it's ended - * because we need to be able to create an automatic "stream - * end" notification when the user's "next" method returns - * BT_NOTIFICATION_ITERATOR_STATUS_END. - * - * We put this reference when the stream is marked as ended. - */ - stream_state->stream = bt_get(stream); - BT_LOGV("Created stream state: stream-addr=%p, stream-name=\"%s\", " - "stream-state-addr=%p", - stream, bt_ctf_stream_get_name(stream), stream_state); + struct bt_message_iterator *iterator = (void *) obj; -end: - return stream_state; -} + BT_ASSERT(iterator); -static -void destroy_base_notification_iterator(struct bt_object *obj) -{ - struct bt_notification_iterator *iterator = - container_of(obj, struct bt_notification_iterator, base); + if (iterator->msgs) { + g_ptr_array_free(iterator->msgs, TRUE); + iterator->msgs = NULL; + } - BT_LOGD_STR("Putting current notification."); - bt_put(iterator->current_notification); g_free(iterator); } static -void bt_private_connection_notification_iterator_destroy(struct bt_object *obj) +void bt_self_component_port_input_message_iterator_destroy(struct bt_object *obj) { - struct bt_notification_iterator_private_connection *iterator; + struct bt_self_component_port_input_message_iterator *iterator; - assert(obj); + BT_ASSERT(obj); /* - * The notification iterator's reference count is 0 if we're + * The message iterator's reference count is 0 if we're * here. Increment it to avoid a double-destroy (possibly * infinitely recursive). This could happen for example if the - * notification iterator's finalization function does bt_get() - * (or anything that causes bt_get() to be called) on itself - * (ref. count goes from 0 to 1), and then bt_put(): the - * reference count would go from 1 to 0 again and this function - * would be called again. + * message iterator's finalization function does + * bt_object_get_ref() (or anything that causes + * bt_object_get_ref() to be called) on itself (ref. count goes + * from 0 to 1), and then bt_object_put_ref(): the reference + * count would go from 1 to 0 again and this function would be + * called again. */ - obj->ref_count.count++; - iterator = (void *) container_of(obj, struct bt_notification_iterator, base); - BT_LOGD("Destroying private connection notification iterator object: addr=%p", - iterator); - bt_private_connection_notification_iterator_finalize(iterator); - - if (iterator->queue) { - struct bt_notification *notif; - - BT_LOGD("Putting notifications in queue."); - - while ((notif = g_queue_pop_tail(iterator->queue))) { - bt_put(notif); - } - - g_queue_free(iterator->queue); - } - - if (iterator->stream_states) { - /* - * Remove our destroy listener from each stream which - * has a state in this iterator. Otherwise the destroy - * listener would be called with an invalid/other - * notification iterator object. - */ - GHashTableIter ht_iter; - gpointer stream_gptr, stream_state_gptr; - - g_hash_table_iter_init(&ht_iter, iterator->stream_states); - - while (g_hash_table_iter_next(&ht_iter, &stream_gptr, &stream_state_gptr)) { - assert(stream_gptr); - - BT_LOGD_STR("Removing stream's destroy listener for notification iterator."); - bt_ctf_stream_remove_destroy_listener( - (void *) stream_gptr, stream_destroy_listener, - iterator); - } - - g_hash_table_destroy(iterator->stream_states); - } - - if (iterator->actions) { - g_array_free(iterator->actions, TRUE); - } + obj->ref_count++; + iterator = (void *) obj; + BT_LIB_LOGD("Destroying self component input port message iterator object: " + "%!+i", iterator); + bt_self_component_port_input_message_iterator_try_finalize(iterator); if (iterator->connection) { /* @@ -441,68 +146,75 @@ void bt_private_connection_notification_iterator_destroy(struct bt_object *obj) * later. */ bt_connection_remove_iterator(iterator->connection, iterator); + iterator->connection = NULL; + } + + if (iterator->auto_seek_msgs) { + while (!g_queue_is_empty(iterator->auto_seek_msgs)) { + bt_object_put_no_null_check( + g_queue_pop_tail(iterator->auto_seek_msgs)); + } + + g_queue_free(iterator->auto_seek_msgs); + iterator->auto_seek_msgs = NULL; } - destroy_base_notification_iterator(obj); + destroy_base_message_iterator(obj); } BT_HIDDEN -void bt_private_connection_notification_iterator_finalize( - struct bt_notification_iterator_private_connection *iterator) +void bt_self_component_port_input_message_iterator_try_finalize( + struct bt_self_component_port_input_message_iterator *iterator) { + typedef void (*method_t)(void *); + struct bt_component_class *comp_class = NULL; - bt_component_class_notification_iterator_finalize_method - finalize_method = NULL; + method_t method = NULL; - assert(iterator); + BT_ASSERT(iterator); switch (iterator->state) { - case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED: + case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED: /* Skip user finalization if user initialization failed */ - BT_LOGD("Not finalizing non-initialized notification iterator: " - "addr=%p", iterator); - return; - case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED: - case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED: + BT_LIB_LOGD("Not finalizing non-initialized message iterator: " + "%!+i", iterator); + goto end; + case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED: /* Already finalized */ - BT_LOGD("Not finalizing notification iterator: already finalized: " - "addr=%p", iterator); - return; + BT_LIB_LOGD("Not finalizing message iterator: already finalized: " + "%!+i", iterator); + goto end; + case BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING: + /* Already finalized */ + BT_LIB_LOGF("Message iterator is already being finalized: " + "%!+i", iterator); + abort(); default: break; } - BT_LOGD("Finalizing notification iterator: addr=%p", iterator); - - if (iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ENDED) { - BT_LOGD("Updating notification iterator's state: " - "new-state=BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED"); - iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED; - } else { - BT_LOGD("Updating notification iterator's state: " - "new-state=BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED"); - iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED; - } - - assert(iterator->upstream_component); + BT_LIB_LOGD("Finalizing message iterator: %!+i", iterator); + set_self_comp_port_input_msg_iterator_state(iterator, + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZING); + BT_ASSERT(iterator->upstream_component); comp_class = iterator->upstream_component->class; /* Call user-defined destroy method */ switch (comp_class->type) { case BT_COMPONENT_CLASS_TYPE_SOURCE: { - struct bt_component_class_source *source_class; + struct bt_component_class_source *src_comp_cls = + (void *) comp_class; - source_class = container_of(comp_class, struct bt_component_class_source, parent); - finalize_method = source_class->methods.iterator.finalize; + method = (method_t) src_comp_cls->methods.msg_iter_finalize; break; } case BT_COMPONENT_CLASS_TYPE_FILTER: { - struct bt_component_class_filter *filter_class; + struct bt_component_class_filter *flt_comp_cls = + (void *) comp_class; - filter_class = container_of(comp_class, struct bt_component_class_filter, parent); - finalize_method = filter_class->methods.iterator.finalize; + method = (method_t) flt_comp_cls->methods.msg_iter_finalize; break; } default: @@ -510,2028 +222,1264 @@ void bt_private_connection_notification_iterator_finalize( abort(); } - if (finalize_method) { - BT_LOGD("Calling user's finalization method: addr=%p", + if (method) { + BT_LIB_LOGD("Calling user's finalization method: %!+i", iterator); - finalize_method( - bt_private_connection_private_notification_iterator_from_notification_iterator(iterator)); + method(iterator); } iterator->upstream_component = NULL; iterator->upstream_port = NULL; - BT_LOGD("Finalized notification iterator: addr=%p", iterator); + set_self_comp_port_input_msg_iterator_state(iterator, + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_FINALIZED); + BT_LIB_LOGD("Finalized message iterator: %!+i", iterator); + +end: + return; } BT_HIDDEN -void bt_private_connection_notification_iterator_set_connection( - struct bt_notification_iterator_private_connection *iterator, +void bt_self_component_port_input_message_iterator_set_connection( + struct bt_self_component_port_input_message_iterator *iterator, struct bt_connection *connection) { - assert(iterator); + BT_ASSERT(iterator); iterator->connection = connection; - BT_LOGV("Set notification iterator's connection: " - "iter-addr=%p, conn-addr=%p", iterator, connection); + BT_LIB_LOGV("Set message iterator's connection: " + "%![iter-]+i, %![conn-]+x", iterator, connection); } static -int create_subscription_mask_from_notification_types( - struct bt_notification_iterator_private_connection *iterator, - const enum bt_notification_type *notif_types) +int init_message_iterator(struct bt_message_iterator *iterator, + enum bt_message_iterator_type type, + bt_object_release_func destroy) { - const enum bt_notification_type *notif_type; int ret = 0; - assert(notif_types); - iterator->subscription_mask = 0; - - for (notif_type = notif_types; - *notif_type != BT_NOTIFICATION_TYPE_SENTINEL; - notif_type++) { - switch (*notif_type) { - case BT_NOTIFICATION_TYPE_ALL: - iterator->subscription_mask |= - BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT | - BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY | - BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN | - BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END | - BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN | - BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END | - BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_EVENTS | - BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_PACKETS; - break; - case BT_NOTIFICATION_TYPE_EVENT: - iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT; - break; - case BT_NOTIFICATION_TYPE_INACTIVITY: - iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY; - break; - case BT_NOTIFICATION_TYPE_STREAM_BEGIN: - iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN; - break; - case BT_NOTIFICATION_TYPE_STREAM_END: - iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END; - break; - case BT_NOTIFICATION_TYPE_PACKET_BEGIN: - iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN; - break; - case BT_NOTIFICATION_TYPE_PACKET_END: - iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END; - break; - case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS: - iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_EVENTS; - break; - case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS: - iterator->subscription_mask |= BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_PACKETS; - break; - default: - ret = -1; - goto end; - } - - BT_LOGV("Added notification type to subscription mask: " - "type=%s, mask=%x", - bt_notification_type_string(*notif_type), - iterator->subscription_mask); + bt_object_init_shared(&iterator->base, destroy); + iterator->type = type; + iterator->msgs = g_ptr_array_new(); + if (!iterator->msgs) { + BT_LOGE_STR("Failed to allocate a GPtrArray."); + ret = -1; + goto end; } + g_ptr_array_set_size(iterator->msgs, MSG_BATCH_SIZE); + end: return ret; } static -void init_notification_iterator(struct bt_notification_iterator *iterator, - enum bt_notification_iterator_type type, - bt_object_release_func destroy) +bt_bool can_seek_ns_from_origin_true( + struct bt_self_component_port_input_message_iterator *iterator, + int64_t ns_from_origin) { - bt_object_init(iterator, destroy); - iterator->type = type; + return BT_TRUE; } -BT_HIDDEN -enum bt_connection_status bt_private_connection_notification_iterator_create( +static +bt_bool can_seek_beginning_true( + struct bt_self_component_port_input_message_iterator *iterator) +{ + return BT_TRUE; +} + +static +struct bt_self_component_port_input_message_iterator * +bt_self_component_port_input_message_iterator_create_initial( struct bt_component *upstream_comp, - struct bt_port *upstream_port, - const enum bt_notification_type *notification_types, - struct bt_connection *connection, - struct bt_notification_iterator_private_connection **user_iterator) + struct bt_port *upstream_port) { - enum bt_connection_status status = BT_CONNECTION_STATUS_OK; - enum bt_component_class_type type; - struct bt_notification_iterator_private_connection *iterator = NULL; - - assert(upstream_comp); - assert(upstream_port); - assert(notification_types); - assert(bt_port_is_connected(upstream_port)); - assert(user_iterator); - BT_LOGD("Creating notification iterator on private connection: " - "upstream-comp-addr=%p, upstream-comp-name=\"%s\", " - "upstream-port-addr=%p, upstream-port-name=\"%s\", " - "conn-addr=%p", - upstream_comp, bt_component_get_name(upstream_comp), - upstream_port, bt_port_get_name(upstream_port), - connection); - type = bt_component_get_class_type(upstream_comp); - assert(type == BT_COMPONENT_CLASS_TYPE_SOURCE || - type == BT_COMPONENT_CLASS_TYPE_FILTER); - iterator = g_new0(struct bt_notification_iterator_private_connection, 1); + int ret; + struct bt_self_component_port_input_message_iterator *iterator = NULL; + + BT_ASSERT(upstream_comp); + BT_ASSERT(upstream_port); + BT_ASSERT(bt_port_is_connected(upstream_port)); + BT_LIB_LOGD("Creating initial message iterator on self component input port: " + "%![up-comp-]+c, %![up-port-]+p", upstream_comp, upstream_port); + BT_ASSERT(bt_component_get_class_type(upstream_comp) == + BT_COMPONENT_CLASS_TYPE_SOURCE || + bt_component_get_class_type(upstream_comp) == + BT_COMPONENT_CLASS_TYPE_FILTER); + iterator = g_new0( + struct bt_self_component_port_input_message_iterator, 1); if (!iterator) { - BT_LOGE_STR("Failed to allocate one private connection notification iterator."); - status = BT_CONNECTION_STATUS_NOMEM; - goto end; - } - - init_notification_iterator((void *) iterator, - BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION, - bt_private_connection_notification_iterator_destroy); - - if (create_subscription_mask_from_notification_types(iterator, - notification_types)) { - BT_LOGW_STR("Cannot create subscription mask from notification types."); - status = BT_CONNECTION_STATUS_INVALID; + BT_LOGE_STR("Failed to allocate one self component input port " + "message iterator."); goto end; } - iterator->stream_states = g_hash_table_new_full(g_direct_hash, - g_direct_equal, NULL, (GDestroyNotify) destroy_stream_state); - if (!iterator->stream_states) { - BT_LOGE_STR("Failed to allocate a GHashTable."); - status = BT_CONNECTION_STATUS_NOMEM; + ret = init_message_iterator((void *) iterator, + BT_MESSAGE_ITERATOR_TYPE_SELF_COMPONENT_PORT_INPUT, + bt_self_component_port_input_message_iterator_destroy); + if (ret) { + /* init_message_iterator() logs errors */ + BT_OBJECT_PUT_REF_AND_RESET(iterator); goto end; } - iterator->queue = g_queue_new(); - if (!iterator->queue) { + iterator->auto_seek_msgs = g_queue_new(); + if (!iterator->auto_seek_msgs) { BT_LOGE_STR("Failed to allocate a GQueue."); - status = BT_CONNECTION_STATUS_NOMEM; - goto end; - } - - iterator->actions = g_array_new(FALSE, FALSE, sizeof(struct action)); - if (!iterator->actions) { - BT_LOGE_STR("Failed to allocate a GArray."); - status = BT_CONNECTION_STATUS_NOMEM; + ret = -1; goto end; } iterator->upstream_component = upstream_comp; iterator->upstream_port = upstream_port; - iterator->connection = connection; - iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_NON_INITIALIZED; - BT_LOGD("Created notification iterator: " - "upstream-comp-addr=%p, upstream-comp-name=\"%s\", " - "upstream-port-addr=%p, upstream-port-name=\"%s\", " - "conn-addr=%p, iter-addr=%p", - upstream_comp, bt_component_get_name(upstream_comp), - upstream_port, bt_port_get_name(upstream_port), - connection, iterator); - - /* Move reference to user */ - *user_iterator = iterator; - iterator = NULL; - -end: - bt_put(iterator); - return status; -} - -void *bt_private_connection_private_notification_iterator_get_user_data( - struct bt_private_connection_private_notification_iterator *private_iterator) -{ - struct bt_notification_iterator_private_connection *iterator = - bt_private_connection_notification_iterator_borrow_from_private(private_iterator); + iterator->connection = iterator->upstream_port->connection; + iterator->graph = bt_component_borrow_graph(upstream_comp); + set_self_comp_port_input_msg_iterator_state(iterator, + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_NON_INITIALIZED); - return iterator ? iterator->user_data : NULL; -} + switch (iterator->upstream_component->class->type) { + case BT_COMPONENT_CLASS_TYPE_SOURCE: + { + struct bt_component_class_source *src_comp_cls = + (void *) iterator->upstream_component->class; + + iterator->methods.next = + (bt_self_component_port_input_message_iterator_next_method) + src_comp_cls->methods.msg_iter_next; + iterator->methods.seek_ns_from_origin = + (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method) + src_comp_cls->methods.msg_iter_seek_ns_from_origin; + iterator->methods.seek_beginning = + (bt_self_component_port_input_message_iterator_seek_beginning_method) + src_comp_cls->methods.msg_iter_seek_beginning; + iterator->methods.can_seek_ns_from_origin = + (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method) + src_comp_cls->methods.msg_iter_can_seek_ns_from_origin; + iterator->methods.can_seek_beginning = + (bt_self_component_port_input_message_iterator_can_seek_beginning_method) + src_comp_cls->methods.msg_iter_can_seek_beginning; + break; + } + case BT_COMPONENT_CLASS_TYPE_FILTER: + { + struct bt_component_class_filter *flt_comp_cls = + (void *) iterator->upstream_component->class; + + iterator->methods.next = + (bt_self_component_port_input_message_iterator_next_method) + flt_comp_cls->methods.msg_iter_next; + iterator->methods.seek_ns_from_origin = + (bt_self_component_port_input_message_iterator_seek_ns_from_origin_method) + flt_comp_cls->methods.msg_iter_seek_ns_from_origin; + iterator->methods.seek_beginning = + (bt_self_component_port_input_message_iterator_seek_beginning_method) + flt_comp_cls->methods.msg_iter_seek_beginning; + iterator->methods.can_seek_ns_from_origin = + (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method) + flt_comp_cls->methods.msg_iter_can_seek_ns_from_origin; + iterator->methods.can_seek_beginning = + (bt_self_component_port_input_message_iterator_can_seek_beginning_method) + flt_comp_cls->methods.msg_iter_can_seek_beginning; + break; + } + default: + abort(); + } -enum bt_notification_iterator_status -bt_private_connection_private_notification_iterator_set_user_data( - struct bt_private_connection_private_notification_iterator *private_iterator, - void *data) -{ - enum bt_notification_iterator_status ret = - BT_NOTIFICATION_ITERATOR_STATUS_OK; - struct bt_notification_iterator_private_connection *iterator = - bt_private_connection_notification_iterator_borrow_from_private(private_iterator); + if (iterator->methods.seek_ns_from_origin && + !iterator->methods.can_seek_ns_from_origin) { + iterator->methods.can_seek_ns_from_origin = + (bt_self_component_port_input_message_iterator_can_seek_ns_from_origin_method) + can_seek_ns_from_origin_true; + } - if (!iterator) { - BT_LOGW_STR("Invalid parameter: notification iterator is NULL."); - ret = BT_NOTIFICATION_ITERATOR_STATUS_INVALID; - goto end; + if (iterator->methods.seek_beginning && + !iterator->methods.can_seek_beginning) { + iterator->methods.can_seek_beginning = + (bt_self_component_port_input_message_iterator_seek_beginning_method) + can_seek_beginning_true; } - iterator->user_data = data; - BT_LOGV("Set notification iterator's user data: " - "iter-addr=%p, user-data-addr=%p", iterator, data); + BT_LIB_LOGD("Created initial message iterator on self component input port: " + "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i", + upstream_port, upstream_comp, iterator); end: - return ret; + return iterator; } -struct bt_notification *bt_notification_iterator_get_notification( - struct bt_notification_iterator *iterator) +struct bt_self_component_port_input_message_iterator * +bt_self_component_port_input_message_iterator_create( + struct bt_self_component_port_input *self_port) { - struct bt_notification *notification = NULL; - + typedef enum bt_self_message_iterator_status (*init_method_t)( + void *, void *, void *); + + init_method_t init_method = NULL; + struct bt_self_component_port_input_message_iterator *iterator = + NULL; + struct bt_port *port = (void *) self_port; + struct bt_port *upstream_port; + struct bt_component *comp; + struct bt_component *upstream_comp; + struct bt_component_class *upstream_comp_cls; + + BT_ASSERT_PRE_NON_NULL(port, "Port"); + comp = bt_port_borrow_component_inline(port); + BT_ASSERT_PRE(bt_port_is_connected(port), + "Port is not connected: %![port-]+p", port); + BT_ASSERT_PRE(comp, "Port is not part of a component: %![port-]+p", + port); + BT_ASSERT_PRE(!bt_component_graph_is_canceled(comp), + "Port's component's graph is canceled: " + "%![port-]+p, %![comp-]+c", port, comp); + BT_ASSERT(port->connection); + upstream_port = port->connection->upstream_port; + BT_ASSERT(upstream_port); + upstream_comp = bt_port_borrow_component_inline(upstream_port); + BT_ASSERT(upstream_comp); + BT_ASSERT_PRE( + bt_component_borrow_graph(upstream_comp)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(upstream_comp)); + upstream_comp_cls = upstream_comp->class; + BT_ASSERT(upstream_comp->class->type == + BT_COMPONENT_CLASS_TYPE_SOURCE || + upstream_comp->class->type == + BT_COMPONENT_CLASS_TYPE_FILTER); + iterator = bt_self_component_port_input_message_iterator_create_initial( + upstream_comp, upstream_port); if (!iterator) { - BT_LOGW_STR("Invalid parameter: notification iterator is NULL."); + BT_LOGW_STR("Cannot create self component input port " + "message iterator."); goto end; } - notification = bt_get( - bt_notification_iterator_borrow_current_notification(iterator)); - -end: - return notification; -} - -static -enum bt_private_connection_notification_iterator_notif_type -bt_notification_iterator_notif_type_from_notif_type( - enum bt_notification_type notif_type) -{ - enum bt_private_connection_notification_iterator_notif_type iter_notif_type; + switch (upstream_comp_cls->type) { + case BT_COMPONENT_CLASS_TYPE_SOURCE: + { + struct bt_component_class_source *src_comp_cls = + (void *) upstream_comp_cls; - switch (notif_type) { - case BT_NOTIFICATION_TYPE_EVENT: - iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_EVENT; + init_method = + (init_method_t) src_comp_cls->methods.msg_iter_init; break; - case BT_NOTIFICATION_TYPE_INACTIVITY: - iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_INACTIVITY; - break; - case BT_NOTIFICATION_TYPE_STREAM_BEGIN: - iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_BEGIN; - break; - case BT_NOTIFICATION_TYPE_STREAM_END: - iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_STREAM_END; - break; - case BT_NOTIFICATION_TYPE_PACKET_BEGIN: - iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_BEGIN; - break; - case BT_NOTIFICATION_TYPE_PACKET_END: - iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_PACKET_END; - break; - case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS: - iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_EVENTS; - break; - case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS: - iter_notif_type = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_NOTIF_TYPE_DISCARDED_PACKETS; + } + case BT_COMPONENT_CLASS_TYPE_FILTER: + { + struct bt_component_class_filter *flt_comp_cls = + (void *) upstream_comp_cls; + + init_method = + (init_method_t) flt_comp_cls->methods.msg_iter_init; break; + } default: + /* Unreachable */ abort(); } - return iter_notif_type; -} + if (init_method) { + int iter_status; -static -bt_bool validate_notification( - struct bt_notification_iterator_private_connection *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream, - struct bt_ctf_packet *notif_packet) -{ - bt_bool is_valid = BT_TRUE; - struct stream_state *stream_state; - struct bt_port *stream_comp_cur_port; - - assert(notif_stream); - stream_comp_cur_port = - bt_ctf_stream_port_for_component(notif_stream, - iterator->upstream_component); - if (!stream_comp_cur_port) { - /* - * This is the first time this notification iterator - * bumps into this stream. Add an action to map the - * iterator's upstream component to the iterator's - * upstream port in this stream. - */ - struct action action = { - .type = ACTION_TYPE_MAP_PORT_TO_COMP_IN_STREAM, - .payload.map_port_to_comp_in_stream = { - .stream = bt_get(notif_stream), - .component = bt_get(iterator->upstream_component), - .port = bt_get(iterator->upstream_port), - }, - }; - - add_action(iterator, &action); - } else { - if (stream_comp_cur_port != iterator->upstream_port) { - /* - * It looks like two different ports of the same - * component are emitting notifications which - * have references to the same stream. This is - * bad: the API guarantees that it can never - * happen. - */ - BT_LOGW("Two different ports of the same component are emitting notifications which refer to the same stream: " - "stream-addr=%p, stream-name=\"%s\", " - "stream-comp-cur-port-addr=%p, " - "stream-comp-cur-port-name=%p, " - "iter-upstream-port-addr=%p, " - "iter-upstream-port-name=%s", - notif_stream, - bt_ctf_stream_get_name(notif_stream), - stream_comp_cur_port, - bt_port_get_name(stream_comp_cur_port), - iterator->upstream_port, - bt_port_get_name(iterator->upstream_port)); - is_valid = BT_FALSE; + BT_LIB_LOGD("Calling user's initialization method: %!+i", iterator); + iter_status = init_method(iterator, upstream_comp, + upstream_port); + BT_LOGD("User method returned: status=%s", + bt_message_iterator_status_string(iter_status)); + if (iter_status != BT_MESSAGE_ITERATOR_STATUS_OK) { + BT_LOGW_STR("Initialization method failed."); + BT_OBJECT_PUT_REF_AND_RESET(iterator); goto end; } - } - stream_state = g_hash_table_lookup(iterator->stream_states, - notif_stream); - if (stream_state) { - BT_LOGV("Stream state already exists: " - "stream-addr=%p, stream-name=\"%s\", " - "stream-state-addr=%p", - notif_stream, - bt_ctf_stream_get_name(notif_stream), stream_state); - - if (stream_state->is_ended) { - /* - * There's a new notification which has a - * reference to a stream which, from this - * iterator's point of view, is ended ("end of - * stream" notification was returned). This is - * bad: the API guarantees that it can never - * happen. - */ - BT_LOGW("Stream is already ended: " - "stream-addr=%p, stream-name=\"%s\"", - notif_stream, - bt_ctf_stream_get_name(notif_stream)); - is_valid = BT_FALSE; - goto end; - } - - switch (notif->type) { - case BT_NOTIFICATION_TYPE_STREAM_BEGIN: - /* - * We already have a stream state, which means - * we already returned a "stream begin" - * notification: this is an invalid duplicate. - */ - BT_LOGW("Duplicate stream beginning notification: " - "stream-addr=%p, stream-name=\"%s\"", - notif_stream, - bt_ctf_stream_get_name(notif_stream)); - is_valid = BT_FALSE; - goto end; - case BT_NOTIFICATION_TYPE_PACKET_BEGIN: - if (notif_packet == stream_state->cur_packet) { - /* Duplicate "packet begin" notification */ - BT_LOGW("Duplicate stream beginning notification: " - "stream-addr=%p, stream-name=\"%s\", " - "packet-addr=%p", - notif_stream, - bt_ctf_stream_get_name(notif_stream), - notif_packet); - is_valid = BT_FALSE; - goto end; - } - break; - default: - break; - } - } + set_self_comp_port_input_msg_iterator_state(iterator, + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE); + g_ptr_array_add(port->connection->iterators, iterator); + BT_LIB_LOGD("Created message iterator on self component input port: " + "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i", + upstream_port, upstream_comp, iterator); end: - return is_valid; + return iterator; } -static -bt_bool is_subscribed_to_notification_type( - struct bt_notification_iterator_private_connection *iterator, - enum bt_notification_type notif_type) +void *bt_self_message_iterator_get_data( + const struct bt_self_message_iterator *self_iterator) { - uint32_t iter_notif_type = - (uint32_t) bt_notification_iterator_notif_type_from_notif_type( - notif_type); + struct bt_self_component_port_input_message_iterator *iterator = + (void *) self_iterator; - return (iter_notif_type & iterator->subscription_mask) ? BT_TRUE : BT_FALSE; + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return iterator->user_data; } -static -void add_action_push_notif( - struct bt_notification_iterator_private_connection *iterator, - struct bt_notification *notif) +void bt_self_message_iterator_set_data( + struct bt_self_message_iterator *self_iterator, void *data) { - struct action action = { - .type = ACTION_TYPE_PUSH_NOTIF, - }; + struct bt_self_component_port_input_message_iterator *iterator = + (void *) self_iterator; - assert(notif); - - if (!is_subscribed_to_notification_type(iterator, notif->type)) { - return; - } - - action.payload.push_notif.notif = bt_get(notif); - add_action(iterator, &action); - BT_LOGV("Added \"push notification\" action: notif-addr=%p", notif); + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + iterator->user_data = data; + BT_LIB_LOGV("Set message iterator's user data: " + "%!+i, user-data-addr=%p", iterator, data); } -static -int add_action_push_notif_stream_begin( - struct bt_notification_iterator_private_connection *iterator, - struct bt_ctf_stream *stream) +enum bt_message_iterator_status +bt_self_component_port_input_message_iterator_next( + struct bt_self_component_port_input_message_iterator *iterator, + bt_message_array_const *msgs, uint64_t *user_count) { - int ret = 0; - struct bt_notification *stream_begin_notif = NULL; + int status = BT_MESSAGE_ITERATOR_STATUS_OK; + + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + BT_ASSERT_PRE_NON_NULL(msgs, "Message array (output)"); + BT_ASSERT_PRE_NON_NULL(user_count, "Message count (output)"); + BT_ASSERT_PRE(iterator->state == + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE, + "Message iterator's \"next\" called, but " + "message iterator is in the wrong state: %!+i", iterator); + BT_ASSERT(iterator->upstream_component); + BT_ASSERT(iterator->upstream_component->class); + BT_ASSERT_PRE( + bt_component_borrow_graph(iterator->upstream_component)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(iterator->upstream_component)); + BT_LIB_LOGD("Getting next self component input port " + "message iterator's messages: %!+i", iterator); - if (!is_subscribed_to_notification_type(iterator, - BT_NOTIFICATION_TYPE_STREAM_BEGIN)) { - BT_LOGV("Not adding \"push stream beginning notification\" action: " - "notification iterator is not subscribed: addr=%p", - iterator); + /* + * Call the user's "next" method to get the next messages + * and status. + */ + BT_ASSERT(iterator->methods.next); + BT_LOGD_STR("Calling user's \"next\" method."); + status = iterator->methods.next(iterator, + (void *) iterator->base.msgs->pdata, MSG_BATCH_SIZE, + user_count); + BT_LOGD("User method returned: status=%s", + bt_message_iterator_status_string(status)); + if (status < 0) { + BT_LOGW_STR("User method failed."); goto end; } - assert(stream); - stream_begin_notif = bt_notification_stream_begin_create(stream); - if (!stream_begin_notif) { - BT_LOGE_STR("Cannot create stream beginning notification."); - goto error; +#ifdef BT_DEV_MODE + /* + * There is no way that this iterator could have been finalized + * during its "next" method, as the only way to do this is to + * put the last iterator's reference, and this can only be done + * by its downstream owner. + * + * For the same reason, there is no way that this iterator could + * have seeked (cannot seek a self message iterator). + */ + BT_ASSERT(iterator->state == + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE); +#endif + + switch (status) { + case BT_MESSAGE_ITERATOR_STATUS_OK: + BT_ASSERT_PRE(*user_count <= MSG_BATCH_SIZE, + "Invalid returned message count: greater than " + "batch size: count=%" PRIu64 ", batch-size=%u", + *user_count, MSG_BATCH_SIZE); + *msgs = (void *) iterator->base.msgs->pdata; + break; + case BT_MESSAGE_ITERATOR_STATUS_AGAIN: + goto end; + case BT_MESSAGE_ITERATOR_STATUS_END: + set_self_comp_port_input_msg_iterator_state(iterator, + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED); + goto end; + default: + /* Unknown non-error status */ + abort(); } - add_action_push_notif(iterator, stream_begin_notif); - BT_LOGV("Added \"push stream beginning notification\" action: " - "stream-addr=%p, stream-name=\"%s\"", - stream, bt_ctf_stream_get_name(stream)); - goto end; - -error: - ret = -1; - end: - bt_put(stream_begin_notif); - return ret; + return status; } -static -int add_action_push_notif_stream_end( - struct bt_notification_iterator_private_connection *iterator, - struct bt_ctf_stream *stream) +enum bt_message_iterator_status bt_port_output_message_iterator_next( + struct bt_port_output_message_iterator *iterator, + bt_message_array_const *msgs_to_user, + uint64_t *count_to_user) { - int ret = 0; - struct bt_notification *stream_end_notif = NULL; + enum bt_message_iterator_status status; + enum bt_graph_status graph_status; - if (!is_subscribed_to_notification_type(iterator, - BT_NOTIFICATION_TYPE_STREAM_END)) { - BT_LOGV("Not adding \"push stream end notification\" action: " - "notification iterator is not subscribed: addr=%p", - iterator); - goto end; - } + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + BT_ASSERT_PRE_NON_NULL(msgs_to_user, "Message array (output)"); + BT_ASSERT_PRE_NON_NULL(count_to_user, "Message count (output)"); + BT_LIB_LOGD("Getting next output port message iterator's messages: " + "%!+i", iterator); + graph_status = bt_graph_consume_sink_no_check(iterator->graph, + iterator->colander); + switch (graph_status) { + case BT_GRAPH_STATUS_CANCELED: + case BT_GRAPH_STATUS_AGAIN: + case BT_GRAPH_STATUS_END: + case BT_GRAPH_STATUS_NOMEM: + status = (int) graph_status; + break; + case BT_GRAPH_STATUS_OK: + status = BT_MESSAGE_ITERATOR_STATUS_OK; - assert(stream); - stream_end_notif = bt_notification_stream_end_create(stream); - if (!stream_end_notif) { - BT_LOGE_STR("Cannot create stream end notification."); - goto error; + /* + * On success, the colander sink moves the messages + * to this iterator's array and sets this iterator's + * message count: move them to the user. + */ + *msgs_to_user = (void *) iterator->base.msgs->pdata; + *count_to_user = iterator->count; + break; + default: + /* Other errors */ + status = BT_MESSAGE_ITERATOR_STATUS_ERROR; } - add_action_push_notif(iterator, stream_end_notif); - BT_LOGV("Added \"push stream end notification\" action: " - "stream-addr=%p, stream-name=\"%s\"", - stream, bt_ctf_stream_get_name(stream)); - goto end; + return status; +} -error: - ret = -1; +struct bt_component * +bt_self_component_port_input_message_iterator_borrow_component( + struct bt_self_component_port_input_message_iterator *iterator) +{ + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return iterator->upstream_component; +} -end: - bt_put(stream_end_notif); - return ret; +const struct bt_component * +bt_self_component_port_input_message_iterator_borrow_component_const( + const struct bt_self_component_port_input_message_iterator *iterator) +{ + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return iterator->upstream_component; } -static -int add_action_push_notif_packet_begin( - struct bt_notification_iterator_private_connection *iterator, - struct bt_ctf_packet *packet) +struct bt_self_component *bt_self_message_iterator_borrow_component( + struct bt_self_message_iterator *self_iterator) { - int ret = 0; - struct bt_notification *packet_begin_notif = NULL; + struct bt_self_component_port_input_message_iterator *iterator = + (void *) self_iterator; - if (!is_subscribed_to_notification_type(iterator, - BT_NOTIFICATION_TYPE_PACKET_BEGIN)) { - BT_LOGV("Not adding \"push packet beginning notification\" action: " - "notification iterator is not subscribed: addr=%p", - iterator); - goto end; - } + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return (void *) iterator->upstream_component; +} - assert(packet); - packet_begin_notif = bt_notification_packet_begin_create(packet); - if (!packet_begin_notif) { - BT_LOGE_STR("Cannot create packet beginning notification."); - goto error; - } +struct bt_self_port_output *bt_self_message_iterator_borrow_port( + struct bt_self_message_iterator *self_iterator) +{ + struct bt_self_component_port_input_message_iterator *iterator = + (void *) self_iterator; - add_action_push_notif(iterator, packet_begin_notif); - BT_LOGV("Added \"push packet beginning notification\" action: " - "packet-addr=%p", packet); - goto end; + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return (void *) iterator->upstream_port; +} -error: - ret = -1; +static +void bt_port_output_message_iterator_destroy(struct bt_object *obj) +{ + struct bt_port_output_message_iterator *iterator = (void *) obj; -end: - bt_put(packet_begin_notif); - return ret; + BT_LIB_LOGD("Destroying output port message iterator object: %!+i", + iterator); + BT_LOGD_STR("Putting graph."); + BT_OBJECT_PUT_REF_AND_RESET(iterator->graph); + BT_LOGD_STR("Putting colander sink component."); + BT_OBJECT_PUT_REF_AND_RESET(iterator->colander); + destroy_base_message_iterator(obj); } -static -int add_action_push_notif_packet_end( - struct bt_notification_iterator_private_connection *iterator, - struct bt_ctf_packet *packet) +struct bt_port_output_message_iterator * +bt_port_output_message_iterator_create(struct bt_graph *graph, + const struct bt_port_output *output_port) { - int ret = 0; - struct bt_notification *packet_end_notif = NULL; + struct bt_port_output_message_iterator *iterator = NULL; + struct bt_component_class_sink *colander_comp_cls = NULL; + struct bt_component *output_port_comp = NULL; + struct bt_component_sink *colander_comp; + enum bt_graph_status graph_status; + struct bt_port_input *colander_in_port = NULL; + struct bt_component_class_sink_colander_data colander_data; + int ret; - if (!is_subscribed_to_notification_type(iterator, - BT_NOTIFICATION_TYPE_PACKET_END)) { - BT_LOGV("Not adding \"push packet end notification\" action: " - "notification iterator is not subscribed: addr=%p", - iterator); + BT_ASSERT_PRE_NON_NULL(graph, "Graph"); + BT_ASSERT_PRE_NON_NULL(output_port, "Output port"); + output_port_comp = bt_port_borrow_component_inline( + (const void *) output_port); + BT_ASSERT_PRE(output_port_comp, + "Output port has no component: %!+p", output_port); + BT_ASSERT_PRE(bt_component_borrow_graph(output_port_comp) == + (void *) graph, + "Output port is not part of graph: %![graph-]+g, %![port-]+p", + graph, output_port); + BT_ASSERT_PRE(!graph->has_sink, + "Graph already has a sink component: %![graph-]+g"); + + /* Create message iterator */ + BT_LIB_LOGD("Creating message iterator on output port: " + "%![port-]+p, %![comp-]+c", output_port, output_port_comp); + iterator = g_new0(struct bt_port_output_message_iterator, 1); + if (!iterator) { + BT_LOGE_STR("Failed to allocate one output port message iterator."); + goto error; + } + + ret = init_message_iterator((void *) iterator, + BT_MESSAGE_ITERATOR_TYPE_PORT_OUTPUT, + bt_port_output_message_iterator_destroy); + if (ret) { + /* init_message_iterator() logs errors */ + BT_OBJECT_PUT_REF_AND_RESET(iterator); goto end; } - assert(packet); - packet_end_notif = bt_notification_packet_end_create(packet); - if (!packet_end_notif) { - BT_LOGE_STR("Cannot create packet end notification."); + /* Create colander component */ + colander_comp_cls = bt_component_class_sink_colander_get(); + if (!colander_comp_cls) { + BT_LOGW("Cannot get colander sink component class."); goto error; } - add_action_push_notif(iterator, packet_end_notif); - BT_LOGV("Added \"push packet end notification\" action: " - "packet-addr=%p", packet); - goto end; - -error: - ret = -1; + iterator->graph = graph; + bt_object_get_no_null_check(iterator->graph); + colander_data.msgs = (void *) iterator->base.msgs->pdata; + colander_data.count_addr = &iterator->count; -end: - bt_put(packet_end_notif); - return ret; -} + /* Hope that nobody uses this very unique name */ + graph_status = + bt_graph_add_sink_component_with_init_method_data( + (void *) graph, colander_comp_cls, + "colander-36ac3409-b1a8-4d60-ab1f-4fdf341a8fb1", + NULL, &colander_data, (void *) &iterator->colander); + if (graph_status != BT_GRAPH_STATUS_OK) { + BT_LIB_LOGW("Cannot add colander sink component to graph: " + "%1[graph-]+g, status=%s", graph, + bt_graph_status_string(graph_status)); + goto error; + } -static -void add_action_set_stream_state_is_ended( - struct bt_notification_iterator_private_connection *iterator, - struct stream_state *stream_state) -{ - struct action action = { - .type = ACTION_TYPE_SET_STREAM_STATE_IS_ENDED, - .payload.set_stream_state_is_ended = { - .stream_state = stream_state, - }, - }; - - assert(stream_state); - add_action(iterator, &action); - BT_LOGV("Added \"set stream state's ended\" action: " - "stream-state-addr=%p", stream_state); -} + /* + * Connect provided output port to the colander component's + * input port. + */ + colander_in_port = + (void *) bt_component_sink_borrow_input_port_by_index_const( + (void *) iterator->colander, 0); + BT_ASSERT(colander_in_port); + graph_status = bt_graph_connect_ports(graph, + output_port, colander_in_port, NULL); + if (graph_status != BT_GRAPH_STATUS_OK) { + BT_LIB_LOGW("Cannot add colander sink component to graph: " + "%![graph-]+g, %![comp-]+c, status=%s", graph, + iterator->colander, + bt_graph_status_string(graph_status)); + goto error; + } -static -void add_action_set_stream_state_cur_packet( - struct bt_notification_iterator_private_connection *iterator, - struct stream_state *stream_state, - struct bt_ctf_packet *packet) -{ - struct action action = { - .type = ACTION_TYPE_SET_STREAM_STATE_CUR_PACKET, - .payload.set_stream_state_cur_packet = { - .stream_state = stream_state, - .packet = bt_get(packet), - }, - }; - - assert(stream_state); - add_action(iterator, &action); - BT_LOGV("Added \"set stream state's current packet\" action: " - "stream-state-addr=%p, packet-addr=%p", - stream_state, packet); -} + /* + * At this point everything went fine. Make the graph + * nonconsumable forever so that only this message iterator + * can consume (thanks to bt_graph_consume_sink_no_check()). + * This avoids leaking the message created by the colander + * sink and moved to the message iterator's message + * member. + */ + bt_graph_set_can_consume(iterator->graph, false); -static -void add_action_update_stream_state_discarded_elements( - struct bt_notification_iterator_private_connection *iterator, - enum action_type type, - struct stream_state *stream_state, - struct bt_ctf_clock_value *cur_begin, - uint64_t cur_count) -{ - struct action action = { - .type = type, - .payload.update_stream_state_discarded_elements = { - .stream_state = stream_state, - .cur_begin = bt_get(cur_begin), - .cur_count = cur_count, - }, - }; - - assert(stream_state); - assert(type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS || - type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS); - add_action(iterator, &action); - if (type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS) { - BT_LOGV("Added \"update stream state's discarded packets\" action: " - "stream-state-addr=%p, cur-begin-addr=%p, cur-count=%" PRIu64, - stream_state, cur_begin, cur_count); - } else if (type == ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS) { - BT_LOGV("Added \"update stream state's discarded events\" action: " - "stream-state-addr=%p, cur-begin-addr=%p, cur-count=%" PRIu64, - stream_state, cur_begin, cur_count); - } -} + /* + * Also set the graph as being configured: it has no active sink + * anyway, so we don't need to call bt_graph_configure(). + */ + graph->config_state = BT_GRAPH_CONFIGURATION_STATE_CONFIGURED; + goto end; -static -int ensure_stream_state_exists( - struct bt_notification_iterator_private_connection *iterator, - struct bt_notification *stream_begin_notif, - struct bt_ctf_stream *notif_stream, - struct stream_state **_stream_state) -{ - int ret = 0; - struct stream_state *stream_state = NULL; +error: + if (iterator && iterator->graph && iterator->colander) { + int ret; - if (!notif_stream) { - /* - * The notification does not reference any stream: no - * need to get or create a stream state. - */ - goto end; - } + /* Remove created colander component from graph if any */ + colander_comp = iterator->colander; + BT_OBJECT_PUT_REF_AND_RESET(iterator->colander); - stream_state = g_hash_table_lookup(iterator->stream_states, - notif_stream); - if (!stream_state) { /* - * This iterator did not bump into this stream yet: - * create a stream state and a "stream begin" - * notification. + * At this point the colander component's reference + * count is 0 because iterator->colander was the only + * owner. We also know that it is not connected because + * this is the last operation before this function + * succeeds. + * + * Since we honor the preconditions here, + * bt_graph_remove_unconnected_component() always + * succeeds. */ - struct action action = { - .type = ACTION_TYPE_ADD_STREAM_STATE, - .payload.add_stream_state = { - .stream = bt_get(notif_stream), - .stream_state = NULL, - }, - }; - - stream_state = create_stream_state(notif_stream); - if (!stream_state) { - BT_LOGE_STR("Cannot create stream state."); - goto error; - } - - action.payload.add_stream_state.stream_state = stream_state; - add_action(iterator, &action); - - if (stream_begin_notif) { - add_action_push_notif(iterator, stream_begin_notif); - } else { - ret = add_action_push_notif_stream_begin(iterator, - notif_stream); - if (ret) { - BT_LOGE_STR("Cannot add \"push stream beginning notification\" action."); - goto error; - } - } + ret = bt_graph_remove_unconnected_component(iterator->graph, + (void *) colander_comp); + BT_ASSERT(ret == 0); } - goto end; -error: - destroy_stream_state(stream_state); - stream_state = NULL; - ret = -1; + BT_OBJECT_PUT_REF_AND_RESET(iterator); end: - *_stream_state = stream_state; - return ret; + bt_object_put_ref(colander_comp_cls); + return (void *) iterator; } -static -struct bt_ctf_field *get_struct_field_uint(struct bt_ctf_field *struct_field, - const char *field_name) +bt_bool bt_self_component_port_input_message_iterator_can_seek_ns_from_origin( + struct bt_self_component_port_input_message_iterator *iterator, + int64_t ns_from_origin) { - struct bt_ctf_field *field = NULL; - struct bt_ctf_field_type *ft = NULL; - - field = bt_ctf_field_structure_get_field_by_name(struct_field, - field_name); - if (!field) { - BT_LOGV_STR("`%s` field does not exist."); + bt_bool can = BT_FALSE; + + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator); + BT_ASSERT_PRE( + bt_component_borrow_graph(iterator->upstream_component)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(iterator->upstream_component)); + + if (iterator->methods.can_seek_ns_from_origin) { + can = iterator->methods.can_seek_ns_from_origin(iterator, + ns_from_origin); goto end; } - if (!bt_ctf_field_is_integer(field)) { - BT_LOGV("Skipping `%s` field because its type is not an integer field type: " - "field-addr=%p, ft-addr=%p, ft-id=%s", field_name, - field, ft, bt_ctf_field_type_id_string( - bt_ctf_field_type_get_type_id(ft))); - BT_PUT(field); - goto end; + /* + * Automatic seeking fall back: if we can seek to the beginning, + * then we can automatically seek to any message. + */ + if (iterator->methods.can_seek_beginning) { + can = iterator->methods.can_seek_beginning(iterator); } - ft = bt_ctf_field_get_type(field); - assert(ft); +end: + return can; +} + +bt_bool bt_self_component_port_input_message_iterator_can_seek_beginning( + struct bt_self_component_port_input_message_iterator *iterator) +{ + bt_bool can = BT_FALSE; - if (bt_ctf_field_type_integer_is_signed(ft)) { - BT_LOGV("Skipping `%s` integer field because its type is signed: " - "field-addr=%p, ft-addr=%p", field_name, field, ft); - BT_PUT(field); - goto end; + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator); + BT_ASSERT_PRE( + bt_component_borrow_graph(iterator->upstream_component)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(iterator->upstream_component)); + + if (iterator->methods.can_seek_beginning) { + can = iterator->methods.can_seek_beginning(iterator); } -end: - bt_put(ft); - return field; + return can; } -static -uint64_t get_packet_context_events_discarded(struct bt_ctf_packet *packet) +static inline +void _set_iterator_state_after_seeking( + struct bt_self_component_port_input_message_iterator *iterator, + enum bt_message_iterator_status status) { - struct bt_ctf_field *packet_context = NULL; - struct bt_ctf_field *field = NULL; - uint64_t retval = -1ULL; - int ret; + enum bt_self_component_port_input_message_iterator_state new_state = 0; - packet_context = bt_ctf_packet_get_context(packet); - if (!packet_context) { - goto end; + /* Set iterator's state depending on seeking status */ + switch (status) { + case BT_MESSAGE_ITERATOR_STATUS_OK: + new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE; + break; + case BT_MESSAGE_ITERATOR_STATUS_AGAIN: + new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_AGAIN; + break; + case BT_MESSAGE_ITERATOR_STATUS_ERROR: + case BT_MESSAGE_ITERATOR_STATUS_NOMEM: + new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_LAST_SEEKING_RETURNED_ERROR; + break; + case BT_MESSAGE_ITERATOR_STATUS_END: + new_state = BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ENDED; + break; + default: + abort(); } - field = get_struct_field_uint(packet_context, "events_discarded"); - if (!field) { - BT_LOGV("`events_discarded` field does not exist in packet's context field: " - "packet-addr=%p, packet-context-field-addr=%p", - packet, packet_context); - goto end; - } + set_self_comp_port_input_msg_iterator_state(iterator, new_state); +} - assert(bt_ctf_field_is_integer(field)); - ret = bt_ctf_field_unsigned_integer_get_value(field, &retval); - if (ret) { - BT_LOGV("Cannot get raw value of packet's context field's `events_discarded` integer field: " - "packet-addr=%p, field-addr=%p", - packet, field); - retval = -1ULL; - goto end; - } +#ifdef BT_DEV_MODE +# define set_iterator_state_after_seeking _set_iterator_state_after_seeking +#else +# define set_iterator_state_after_seeking(_iter, _status) ((void) _iter); ((void) _status); +#endif -end: - bt_put(packet_context); - bt_put(field); - return retval; +enum bt_message_iterator_status +bt_self_component_port_input_message_iterator_seek_beginning( + struct bt_self_component_port_input_message_iterator *iterator) +{ + int status; + + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator); + BT_ASSERT_PRE( + bt_component_borrow_graph(iterator->upstream_component)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(iterator->upstream_component)); + BT_ASSERT_PRE( + bt_self_component_port_input_message_iterator_can_seek_beginning( + iterator), + "Message iterator cannot seek beginning: %!+i", iterator); + BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i", iterator); + set_self_comp_port_input_msg_iterator_state(iterator, + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING); + status = iterator->methods.seek_beginning(iterator); + BT_LOGD("User method returned: status=%s", + bt_message_iterator_status_string(status)); + BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK || + status == BT_MESSAGE_ITERATOR_STATUS_ERROR || + status == BT_MESSAGE_ITERATOR_STATUS_NOMEM || + status == BT_MESSAGE_ITERATOR_STATUS_AGAIN, + "Unexpected status: %![iter-]+i, status=%s", + iterator, bt_common_self_message_iterator_status_string(status)); + set_iterator_state_after_seeking(iterator, status); + return status; } -static -uint64_t get_packet_context_packet_seq_num(struct bt_ctf_packet *packet) +static inline +enum bt_message_iterator_status auto_seek_handle_message( + struct bt_self_component_port_input_message_iterator *iterator, + int64_t ns_from_origin, const struct bt_message *msg, + bool *got_first) { - struct bt_ctf_field *packet_context = NULL; - struct bt_ctf_field *field = NULL; - uint64_t retval = -1ULL; + enum bt_message_iterator_status status = BT_MESSAGE_ITERATOR_STATUS_OK; + int64_t msg_ns_from_origin; + const struct bt_clock_snapshot *clk_snapshot = NULL; int ret; - packet_context = bt_ctf_packet_get_context(packet); - if (!packet_context) { - goto end; - } + BT_ASSERT(msg); + BT_ASSERT(got_first); - field = get_struct_field_uint(packet_context, "packet_seq_num"); - if (!field) { - BT_LOGV("`packet_seq_num` field does not exist in packet's context field: " - "packet-addr=%p, packet-context-field-addr=%p", - packet, packet_context); - goto end; - } + switch (msg->type) { + case BT_MESSAGE_TYPE_EVENT: + { + const struct bt_message_event *event_msg = + (const void *) msg; - assert(bt_ctf_field_is_integer(field)); - ret = bt_ctf_field_unsigned_integer_get_value(field, &retval); - if (ret) { - BT_LOGV("Cannot get raw value of packet's context field's `packet_seq_num` integer field: " - "packet-addr=%p, field-addr=%p", - packet, field); - retval = -1ULL; - goto end; + clk_snapshot = event_msg->default_cs; + BT_ASSERT_PRE(clk_snapshot, + "Event message has no default clock snapshot: %!+n", + event_msg); + break; } + case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY: + { + const struct bt_message_message_iterator_inactivity *inactivity_msg = + (const void *) msg; -end: - bt_put(packet_context); - bt_put(field); - return retval; -} - -static -int handle_discarded_packets( - struct bt_notification_iterator_private_connection *iterator, - struct bt_ctf_packet *packet, - struct bt_ctf_clock_value *ts_begin, - struct bt_ctf_clock_value *ts_end, - struct stream_state *stream_state) -{ - struct bt_notification *notif = NULL; - uint64_t diff; - uint64_t prev_count, next_count; - int ret = 0; + clk_snapshot = inactivity_msg->default_cs; + BT_ASSERT(clk_snapshot); + break; + } + case BT_MESSAGE_TYPE_PACKET_BEGINNING: + case BT_MESSAGE_TYPE_PACKET_END: + { + const struct bt_message_packet *packet_msg = + (const void *) msg; - next_count = get_packet_context_packet_seq_num(packet); - if (next_count == -1ULL) { - /* - * Stream does not have seqnum field, skip discarded - * packets feature. - */ - goto end; + clk_snapshot = packet_msg->default_cs; + BT_ASSERT_PRE(clk_snapshot, + "Packet message has no default clock snapshot: %!+n", + packet_msg); + break; } - prev_count = stream_state->discarded_packets_state.cur_count; - - if (prev_count != -1ULL) { - if (next_count < prev_count) { - BT_LOGW("Current value of packet's context field's `packet_seq_num` field is lesser than the previous value for the same stream: " - "not updating the stream state's current value: " - "packet-addr=%p, prev-count=%" PRIu64 ", " - "cur-count=%" PRIu64, - packet, prev_count, next_count); + case BT_MESSAGE_TYPE_DISCARDED_EVENTS: + case BT_MESSAGE_TYPE_DISCARDED_PACKETS: + { + struct bt_message_discarded_items *msg_disc_items = + (void *) msg; + + BT_ASSERT_PRE(msg_disc_items->default_begin_cs && + msg_disc_items->default_end_cs, + "Discarded events/packets message has no default clock snapshots: %!+n", + msg_disc_items); + ret = bt_clock_snapshot_get_ns_from_origin( + msg_disc_items->default_begin_cs, + &msg_ns_from_origin); + if (ret) { + status = BT_MESSAGE_ITERATOR_STATUS_ERROR; goto end; } - if (next_count == prev_count) { - BT_LOGW("Current value of packet's context field's `packet_seq_num` field is equal to the previous value for the same stream: " - "not updating the stream state's current value: " - "packet-addr=%p, prev-count=%" PRIu64 ", " - "cur-count=%" PRIu64, - packet, prev_count, next_count); + + if (msg_ns_from_origin >= ns_from_origin) { + *got_first = true; + goto push_msg; + } + + ret = bt_clock_snapshot_get_ns_from_origin( + msg_disc_items->default_end_cs, + &msg_ns_from_origin); + if (ret) { + status = BT_MESSAGE_ITERATOR_STATUS_ERROR; goto end; } - diff = next_count - prev_count; - if (diff > 1) { + if (msg_ns_from_origin >= ns_from_origin) { /* - * Add a discarded packets notification. The packets - * are considered to be lost between the state's last time - * and the current packet's beginning time. - * The counter is expected to monotonically increase of - * 1 for each packet. Therefore, the number of missing - * packets is 'diff - 1'. + * The discarded items message's beginning time + * is before the requested seeking time, but its + * end time is after. Modify the message so as + * to set its beginning time to the requested + * seeking time, and make its item count unknown + * as we don't know if items were really + * discarded within the new time range. */ - notif = bt_notification_discarded_elements_create( - BT_NOTIFICATION_TYPE_DISCARDED_PACKETS, - stream_state->stream, - stream_state->discarded_packets_state.cur_begin, - ts_begin, diff - 1); - if (!notif) { - BT_LOGE_STR("Cannot create discarded packets notification."); - ret = -1; + uint64_t new_begin_raw_value; + + ret = bt_clock_class_clock_value_from_ns_from_origin( + msg_disc_items->default_end_cs->clock_class, + ns_from_origin, &new_begin_raw_value); + if (ret) { + status = BT_MESSAGE_ITERATOR_STATUS_ERROR; goto end; } - add_action_push_notif(iterator, notif); - } - } - - add_action_update_stream_state_discarded_elements(iterator, - ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_PACKETS, - stream_state, ts_end, next_count); - -end: - bt_put(notif); - return ret; -} - -static -int handle_discarded_events( - struct bt_notification_iterator_private_connection *iterator, - struct bt_ctf_packet *packet, - struct bt_ctf_clock_value *ts_begin, - struct bt_ctf_clock_value *ts_end, - struct stream_state *stream_state) -{ - struct bt_notification *notif = NULL; - uint64_t diff; - uint64_t next_count; - int ret = 0; - - next_count = get_packet_context_events_discarded(packet); - if (next_count == -1ULL) { - next_count = stream_state->discarded_events_state.cur_count; - goto update_state; - } + bt_clock_snapshot_set_raw_value( + msg_disc_items->default_begin_cs, + new_begin_raw_value); + msg_disc_items->count.base.avail = + BT_PROPERTY_AVAILABILITY_NOT_AVAILABLE; - if (next_count < stream_state->discarded_events_state.cur_count) { - BT_LOGW("Current value of packet's context field's `events_discarded` field is lesser than the previous value for the same stream: " - "not updating the stream state's current value: " - "packet-addr=%p, prev-count=%" PRIu64 ", " - "cur-count=%" PRIu64, - packet, stream_state->discarded_events_state.cur_count, - next_count); - goto end; + /* + * It is safe to push it because its beginning + * time is exactly the requested seeking time. + */ + goto push_msg; + } else { + goto skip_msg; + } } + case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING: + { + const struct bt_message_stream_activity *stream_act_msg = + (const void *) msg; - diff = next_count - stream_state->discarded_events_state.cur_count; - if (diff > 0) { - /* - * Add a discarded events notification. The events are - * considered to be lost betweem the state's last time - * and the current packet's end time. - */ - notif = bt_notification_discarded_elements_create( - BT_NOTIFICATION_TYPE_DISCARDED_EVENTS, - stream_state->stream, - stream_state->discarded_events_state.cur_begin, - ts_end, diff); - if (!notif) { - BT_LOGE_STR("Cannot create discarded events notification."); - ret = -1; - goto end; + switch (stream_act_msg->default_cs_state) { + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN: + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE: + /* + * -inf is always less than any requested time, + * and we can't assume any specific time for an + * unknown clock snapshot, so skip this. + */ + goto skip_msg; + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN: + clk_snapshot = stream_act_msg->default_cs; + BT_ASSERT(clk_snapshot); + break; + default: + abort(); } - add_action_push_notif(iterator, notif); + break; } + case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END: + { + const struct bt_message_stream_activity *stream_act_msg = + (const void *) msg; -update_state: - add_action_update_stream_state_discarded_elements(iterator, - ACTION_TYPE_UPDATE_STREAM_STATE_DISCARDED_EVENTS, - stream_state, ts_end, next_count); - -end: - bt_put(notif); - return ret; -} - -static -int get_field_clock_value(struct bt_ctf_field *root_field, - const char *field_name, - struct bt_ctf_clock_value **user_clock_val) -{ - struct bt_ctf_field *field; - struct bt_ctf_field_type *ft = NULL; - struct bt_ctf_clock_class *clock_class = NULL; - struct bt_ctf_clock_value *clock_value = NULL; - uint64_t val; - int ret = 0; + switch (stream_act_msg->default_cs_state) { + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN: + /* + * We can't assume any specific time for an + * unknown clock snapshot, so skip this. + */ + goto skip_msg; + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE: + /* + * +inf is always greater than any requested + * time. + */ + *got_first = true; + goto push_msg; + case BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_KNOWN: + clk_snapshot = stream_act_msg->default_cs; + BT_ASSERT(clk_snapshot); + break; + default: + abort(); + } - field = get_struct_field_uint(root_field, field_name); - if (!field) { - /* Not an error: skip this */ - goto end; + break; } - - ft = bt_ctf_field_get_type(field); - assert(ft); - clock_class = bt_ctf_field_type_integer_get_mapped_clock_class(ft); - if (!clock_class) { - BT_LOGW("Integer field type has no mapped clock class but it's expected to have one: " - "ft-addr=%p", ft); - ret = -1; - goto end; + case BT_MESSAGE_TYPE_STREAM_BEGINNING: + case BT_MESSAGE_TYPE_STREAM_END: + /* Ignore */ + goto skip_msg; + default: + abort(); } - ret = bt_ctf_field_unsigned_integer_get_value(field, &val); + BT_ASSERT(clk_snapshot); + ret = bt_clock_snapshot_get_ns_from_origin(clk_snapshot, + &msg_ns_from_origin); if (ret) { - BT_LOGW("Cannot get integer field's raw value: " - "field-addr=%p", field); - ret = -1; + status = BT_MESSAGE_ITERATOR_STATUS_ERROR; goto end; } - clock_value = bt_ctf_clock_value_create(clock_class, val); - if (!clock_value) { - BT_LOGE_STR("Cannot create clock value from clock class."); - ret = -1; - goto end; + if (msg_ns_from_origin >= ns_from_origin) { + *got_first = true; + goto push_msg; } - /* Move clock value to user */ - *user_clock_val = clock_value; - clock_value = NULL; +skip_msg: + bt_object_put_no_null_check(msg); + msg = NULL; + goto end; + +push_msg: + g_queue_push_head(iterator->auto_seek_msgs, (void *) msg); + msg = NULL; end: - bt_put(field); - bt_put(ft); - bt_put(clock_class); - bt_put(clock_value); - return ret; + BT_ASSERT(!msg || status != BT_MESSAGE_ITERATOR_STATUS_OK); + return status; } static -int get_ts_begin_ts_end_from_packet(struct bt_ctf_packet *packet, - struct bt_ctf_clock_value **user_ts_begin, - struct bt_ctf_clock_value **user_ts_end) +enum bt_message_iterator_status find_message_ge_ns_from_origin( + struct bt_self_component_port_input_message_iterator *iterator, + int64_t ns_from_origin) { - struct bt_ctf_field *packet_context = NULL; - struct bt_ctf_clock_value *ts_begin = NULL; - struct bt_ctf_clock_value *ts_end = NULL; - int ret = 0; + int status; + enum bt_self_component_port_input_message_iterator_state init_state = + iterator->state; + const struct bt_message *messages[MSG_BATCH_SIZE]; + uint64_t user_count = 0; + uint64_t i; + bool got_first = false; - packet_context = bt_ctf_packet_get_context(packet); - if (!packet_context) { - goto end; - } + BT_ASSERT(iterator); + memset(&messages[0], 0, sizeof(messages[0]) * MSG_BATCH_SIZE); - ret = get_field_clock_value(packet_context, "timestamp_begin", - &ts_begin); - if (ret) { - BT_LOGW("Cannot create clock value for packet context's `timestamp_begin` field: " - "packet-addr=%p, packet-context-field-addr=%p", - packet, packet_context); - goto end; - } + /* + * Make this iterator temporarily active (not seeking) to call + * the "next" method. + */ + set_self_comp_port_input_msg_iterator_state(iterator, + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE); - ret = get_field_clock_value(packet_context, "timestamp_end", - &ts_end); - if (ret) { - BT_LOGW("Cannot create clock value for packet context's `timestamp_begin` field: " - "packet-addr=%p, packet-context-field-addr=%p", - packet, packet_context); - goto end; - } + BT_ASSERT(iterator->methods.next); - /* Move clock values to user */ - *user_ts_begin = ts_begin; - ts_begin = NULL; - *user_ts_end = ts_end; - ts_end = NULL; + while (!got_first) { + /* + * Call the user's "next" method to get the next + * messages and status. + */ + BT_LOGD_STR("Calling user's \"next\" method."); + status = iterator->methods.next(iterator, + &messages[0], MSG_BATCH_SIZE, &user_count); + BT_LOGD("User method returned: status=%s", + bt_message_iterator_status_string(status)); -end: - bt_put(packet_context); - bt_put(ts_begin); - bt_put(ts_end); - return ret; -} - -static -int handle_discarded_elements( - struct bt_notification_iterator_private_connection *iterator, - struct bt_ctf_packet *packet, struct stream_state *stream_state) -{ - struct bt_ctf_clock_value *ts_begin = NULL; - struct bt_ctf_clock_value *ts_end = NULL; - int ret; - - ret = get_ts_begin_ts_end_from_packet(packet, &ts_begin, &ts_end); - if (ret) { - BT_LOGW("Cannot get packet's beginning or end clock values: " - "packet-addr=%p, ret=%d", packet, ret); - ret = -1; - goto end; - } - - ret = handle_discarded_packets(iterator, packet, ts_begin, ts_end, - stream_state); - if (ret) { - BT_LOGW("Cannot handle discarded packets for packet: " - "packet-addr=%p, ret=%d", packet, ret); - ret = -1; - goto end; - } - - ret = handle_discarded_events(iterator, packet, ts_begin, ts_end, - stream_state); - if (ret) { - BT_LOGW("Cannot handle discarded events for packet: " - "packet-addr=%p, ret=%d", packet, ret); - ret = -1; - goto end; - } - -end: - bt_put(ts_begin); - bt_put(ts_end); - return ret; -} - -static -int handle_packet_switch( - struct bt_notification_iterator_private_connection *iterator, - struct bt_notification *packet_begin_notif, - struct bt_ctf_packet *new_packet, - struct stream_state *stream_state) -{ - int ret = 0; - - if (stream_state->cur_packet == new_packet) { - goto end; - } - - BT_LOGV("Handling packet switch: " - "cur-packet-addr=%p, new-packet-addr=%p", - stream_state->cur_packet, new_packet); - - if (stream_state->cur_packet) { - /* End of the current packet */ - ret = add_action_push_notif_packet_end(iterator, - stream_state->cur_packet); - if (ret) { - BT_LOGE_STR("Cannot add \"push packet end notification\" action."); - goto error; +#ifdef BT_DEV_MODE + /* + * The user's "next" method must not do any action which + * would change the iterator's state. + */ + BT_ASSERT(iterator->state == + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE); +#endif + + switch (status) { + case BT_MESSAGE_ITERATOR_STATUS_OK: + BT_ASSERT_PRE(user_count <= MSG_BATCH_SIZE, + "Invalid returned message count: greater than " + "batch size: count=%" PRIu64 ", batch-size=%u", + user_count, MSG_BATCH_SIZE); + break; + case BT_MESSAGE_ITERATOR_STATUS_AGAIN: + case BT_MESSAGE_ITERATOR_STATUS_ERROR: + case BT_MESSAGE_ITERATOR_STATUS_NOMEM: + case BT_MESSAGE_ITERATOR_STATUS_END: + goto end; + default: + abort(); } - } - /* - * Check the new packet's context fields for discarded packets - * and events to emit those automatic notifications. - */ - ret = handle_discarded_elements(iterator, new_packet, stream_state); - if (ret) { - BT_LOGE_STR("Cannot handle discarded elements for new packet."); - goto error; - } + for (i = 0; i < user_count; i++) { + if (got_first) { + g_queue_push_head(iterator->auto_seek_msgs, + (void *) messages[i]); + messages[i] = NULL; + continue; + } - /* Beginning of the new packet */ - if (packet_begin_notif) { - add_action_push_notif(iterator, packet_begin_notif); - } else if (new_packet) { - ret = add_action_push_notif_packet_begin(iterator, - new_packet); - if (ret) { - BT_LOGE_STR("Cannot add \"push packet beginning notification\" action."); - goto error; + status = auto_seek_handle_message(iterator, + ns_from_origin, messages[i], &got_first); + if (status == BT_MESSAGE_ITERATOR_STATUS_OK) { + /* Message was either pushed or moved */ + messages[i] = NULL; + } else { + goto end; + } } } - add_action_set_stream_state_cur_packet(iterator, stream_state, - new_packet); - goto end; - -error: - ret = -1; - -end: - return ret; -} - -static -int handle_notif_stream_begin( - struct bt_notification_iterator_private_connection *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream) -{ - int ret = 0; - struct stream_state *stream_state; - - assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_BEGIN); - assert(notif_stream); - ret = ensure_stream_state_exists(iterator, notif, notif_stream, - &stream_state); - if (ret) { - BT_LOGE_STR("Cannot ensure that stream state exists."); - goto error; - } - - goto end; - -error: - ret = -1; - end: - return ret; -} - -static -int handle_notif_stream_end( - struct bt_notification_iterator_private_connection *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream) -{ - int ret = 0; - struct stream_state *stream_state; - - assert(notif->type == BT_NOTIFICATION_TYPE_STREAM_END); - assert(notif_stream); - ret = ensure_stream_state_exists(iterator, NULL, notif_stream, - &stream_state); - if (ret) { - BT_LOGE_STR("Cannot ensure that stream state exists."); - goto error; - } - - ret = handle_packet_switch(iterator, NULL, NULL, stream_state); - if (ret) { - BT_LOGE_STR("Cannot handle packet switch."); - goto error; - } - - add_action_push_notif(iterator, notif); - add_action_set_stream_state_is_ended(iterator, stream_state); - goto end; - -error: - ret = -1; - -end: - return ret; -} - -static -int handle_notif_discarded_elements( - struct bt_notification_iterator_private_connection *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream) -{ - int ret = 0; - struct stream_state *stream_state; - - assert(notif->type == BT_NOTIFICATION_TYPE_DISCARDED_EVENTS || - notif->type == BT_NOTIFICATION_TYPE_DISCARDED_PACKETS); - assert(notif_stream); - ret = ensure_stream_state_exists(iterator, NULL, notif_stream, - &stream_state); - if (ret) { - BT_LOGE_STR("Cannot ensure that stream state exists."); - goto error; - } - - add_action_push_notif(iterator, notif); - goto end; - -error: - ret = -1; - -end: - return ret; -} - -static -int handle_notif_packet_begin( - struct bt_notification_iterator_private_connection *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream, - struct bt_ctf_packet *notif_packet) -{ - int ret = 0; - struct stream_state *stream_state; - - assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_BEGIN); - assert(notif_packet); - ret = ensure_stream_state_exists(iterator, NULL, notif_stream, - &stream_state); - if (ret) { - BT_LOGE_STR("Cannot ensure that stream state exists."); - goto error; - } - - ret = handle_packet_switch(iterator, notif, notif_packet, stream_state); - if (ret) { - BT_LOGE_STR("Cannot handle packet switch."); - goto error; - } - - goto end; - -error: - ret = -1; - -end: - return ret; -} - -static -int handle_notif_packet_end( - struct bt_notification_iterator_private_connection *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream, - struct bt_ctf_packet *notif_packet) -{ - int ret = 0; - struct stream_state *stream_state; - - assert(notif->type == BT_NOTIFICATION_TYPE_PACKET_END); - assert(notif_packet); - ret = ensure_stream_state_exists(iterator, NULL, notif_stream, - &stream_state); - if (ret) { - BT_LOGE_STR("Cannot ensure that stream state exists."); - goto error; - } - - ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state); - if (ret) { - BT_LOGE_STR("Cannot handle packet switch."); - goto error; - } - - /* End of the current packet */ - add_action_push_notif(iterator, notif); - add_action_set_stream_state_cur_packet(iterator, stream_state, NULL); - goto end; - -error: - ret = -1; - -end: - return ret; -} - -static -int handle_notif_event( - struct bt_notification_iterator_private_connection *iterator, - struct bt_notification *notif, - struct bt_ctf_stream *notif_stream, - struct bt_ctf_packet *notif_packet) -{ - int ret = 0; - struct stream_state *stream_state; - - assert(notif->type == BT_NOTIFICATION_TYPE_EVENT); - assert(notif_packet); - ret = ensure_stream_state_exists(iterator, NULL, notif_stream, - &stream_state); - if (ret) { - BT_LOGE_STR("Cannot ensure that stream state exists."); - goto error; - } - - ret = handle_packet_switch(iterator, NULL, notif_packet, stream_state); - if (ret) { - BT_LOGE_STR("Cannot handle packet switch."); - goto error; - } - - add_action_push_notif(iterator, notif); - goto end; - -error: - ret = -1; - -end: - return ret; -} - -static -int enqueue_notification_and_automatic( - struct bt_notification_iterator_private_connection *iterator, - struct bt_notification *notif) -{ - int ret = 0; - struct bt_ctf_event *notif_event = NULL; - struct bt_ctf_stream *notif_stream = NULL; - struct bt_ctf_packet *notif_packet = NULL; - - assert(notif); - - BT_LOGV("Enqueuing user notification and automatic notifications: " - "iter-addr=%p, notif-addr=%p", iterator, notif); - - // TODO: Skip most of this if the iterator is only subscribed - // to event/inactivity notifications. - - /* Get the stream and packet referred by the notification */ - switch (notif->type) { - case BT_NOTIFICATION_TYPE_EVENT: - notif_event = bt_notification_event_borrow_event(notif); - assert(notif_event); - notif_packet = bt_ctf_event_borrow_packet(notif_event); - assert(notif_packet); - break; - case BT_NOTIFICATION_TYPE_STREAM_BEGIN: - notif_stream = - bt_notification_stream_begin_borrow_stream(notif); - assert(notif_stream); - break; - case BT_NOTIFICATION_TYPE_STREAM_END: - notif_stream = bt_notification_stream_end_borrow_stream(notif); - assert(notif_stream); - break; - case BT_NOTIFICATION_TYPE_PACKET_BEGIN: - notif_packet = - bt_notification_packet_begin_borrow_packet(notif); - assert(notif_packet); - break; - case BT_NOTIFICATION_TYPE_PACKET_END: - notif_packet = bt_notification_packet_end_borrow_packet(notif); - assert(notif_packet); - break; - case BT_NOTIFICATION_TYPE_INACTIVITY: - /* Always valid */ - goto handle_notif; - default: - /* - * Invalid type of notification. Only the notification - * types above are allowed to be returned by a user - * component. - */ - BT_LOGF("Unexpected notification type at this point: " - "notif-addr=%p, notif-type=%s", notif, - bt_notification_type_string(notif->type)); - abort(); - } - - if (notif_packet) { - notif_stream = bt_ctf_packet_borrow_stream(notif_packet); - assert(notif_stream); - } - - if (!notif_stream) { - /* - * The notification has no reference to a stream: it - * cannot cause the creation of automatic notifications. - */ - BT_LOGV_STR("Notification has no reference to any stream: skipping automatic notification generation."); - goto end; - } - - if (!validate_notification(iterator, notif, notif_stream, - notif_packet)) { - BT_LOGW_STR("Invalid notification."); - goto error; - } - -handle_notif: - switch (notif->type) { - case BT_NOTIFICATION_TYPE_EVENT: - ret = handle_notif_event(iterator, notif, notif_stream, - notif_packet); - break; - case BT_NOTIFICATION_TYPE_STREAM_BEGIN: - ret = handle_notif_stream_begin(iterator, notif, notif_stream); - break; - case BT_NOTIFICATION_TYPE_STREAM_END: - ret = handle_notif_stream_end(iterator, notif, notif_stream); - break; - case BT_NOTIFICATION_TYPE_PACKET_BEGIN: - ret = handle_notif_packet_begin(iterator, notif, notif_stream, - notif_packet); - break; - case BT_NOTIFICATION_TYPE_PACKET_END: - ret = handle_notif_packet_end(iterator, notif, notif_stream, - notif_packet); - break; - case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS: - case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS: - ret = handle_notif_discarded_elements(iterator, notif, - notif_stream); - break; - case BT_NOTIFICATION_TYPE_INACTIVITY: - add_action_push_notif(iterator, notif); - break; - default: - break; - } - - if (ret) { - BT_LOGW_STR("Failed to handle notification for automatic notification generation."); - goto error; + for (i = 0; i < user_count; i++) { + if (messages[i]) { + bt_object_put_no_null_check(messages[i]); + } } - apply_actions(iterator); - BT_LOGV("Enqueued user notification and automatic notifications: " - "iter-addr=%p, notif-addr=%p", iterator, notif); - goto end; - -error: - ret = -1; - -end: - return ret; + set_self_comp_port_input_msg_iterator_state(iterator, init_state); + return status; } static -int handle_end(struct bt_notification_iterator_private_connection *iterator) +enum bt_self_message_iterator_status post_auto_seek_next( + struct bt_self_component_port_input_message_iterator *iterator, + bt_message_array_const msgs, uint64_t capacity, + uint64_t *count) { - GHashTableIter stream_state_iter; - gpointer stream_gptr, stream_state_gptr; - int ret = 0; - - BT_LOGV("Handling end of iteration: addr=%p", iterator); + BT_ASSERT(!g_queue_is_empty(iterator->auto_seek_msgs)); + *count = 0; /* - * Emit a "stream end" notification for each non-ended stream - * known by this iterator and mark them as ended. + * Move auto-seek messages to the output array (which is this + * iterator's base message array). */ - g_hash_table_iter_init(&stream_state_iter, iterator->stream_states); + while (capacity > 0 && !g_queue_is_empty(iterator->auto_seek_msgs)) { + msgs[*count] = g_queue_pop_tail(iterator->auto_seek_msgs); + capacity--; + (*count)++; + } - while (g_hash_table_iter_next(&stream_state_iter, &stream_gptr, - &stream_state_gptr)) { - struct stream_state *stream_state = stream_state_gptr; + BT_ASSERT(*count > 0); - assert(stream_state_gptr); + if (g_queue_is_empty(iterator->auto_seek_msgs)) { + /* No more auto-seek messages */ + switch (iterator->upstream_component->class->type) { + case BT_COMPONENT_CLASS_TYPE_SOURCE: + { + struct bt_component_class_source *src_comp_cls = + (void *) iterator->upstream_component->class; - if (stream_state->is_ended) { - continue; + iterator->methods.next = + (bt_self_component_port_input_message_iterator_next_method) + src_comp_cls->methods.msg_iter_next; + break; } + case BT_COMPONENT_CLASS_TYPE_FILTER: + { + struct bt_component_class_filter *flt_comp_cls = + (void *) iterator->upstream_component->class; - ret = handle_packet_switch(iterator, NULL, NULL, stream_state); - if (ret) { - BT_LOGE_STR("Cannot handle packet switch."); - goto error; + iterator->methods.next = + (bt_self_component_port_input_message_iterator_next_method) + flt_comp_cls->methods.msg_iter_next; + break; } - - ret = add_action_push_notif_stream_end(iterator, stream_gptr); - if (ret) { - BT_LOGE_STR("Cannot add \"push stream end notification\" action."); - goto error; + default: + abort(); } - - add_action_set_stream_state_is_ended(iterator, stream_state); } - apply_actions(iterator); - BT_LOGV("Handled end of iteration: addr=%p", iterator); - goto end; - -error: - ret = -1; - -end: - return ret; + return BT_SELF_MESSAGE_ITERATOR_STATUS_OK; } -static -enum bt_notification_iterator_status ensure_queue_has_notifications( - struct bt_notification_iterator_private_connection *iterator) +enum bt_message_iterator_status +bt_self_component_port_input_message_iterator_seek_ns_from_origin( + struct bt_self_component_port_input_message_iterator *iterator, + int64_t ns_from_origin) { - struct bt_private_connection_private_notification_iterator *priv_iterator = - bt_private_connection_private_notification_iterator_from_notification_iterator(iterator); - bt_component_class_notification_iterator_next_method next_method = NULL; - struct bt_notification_iterator_next_method_return next_return = { - .status = BT_NOTIFICATION_ITERATOR_STATUS_OK, - .notification = NULL, - }; - enum bt_notification_iterator_status status = - BT_NOTIFICATION_ITERATOR_STATUS_OK; - int ret; - - assert(iterator); - BT_LOGD("Ensuring that notification iterator's queue has at least one notification: " - "iter-addr=%p, queue-size=%u, iter-state=%s", - iterator, iterator->queue->length, - bt_private_connection_notification_iterator_state_string(iterator->state)); - - if (iterator->queue->length > 0) { - /* - * We already have enough. Even if this notification - * iterator is finalized, its user can still flush its - * current queue's content by calling its "next" method - * since this content is local and has no impact on what - * used to be the iterator's upstream component. - */ - BT_LOGD_STR("Queue already has at least one notification."); - goto end; - } - - switch (iterator->state) { - case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED: - case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED: - BT_LOGD_STR("Notification iterator's \"next\" called, but it is finalized."); - status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED; - goto end; - case BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ENDED: - BT_LOGD_STR("Notification iterator is ended."); - status = BT_NOTIFICATION_ITERATOR_STATUS_END; - goto end; - default: - break; - } - - assert(iterator->upstream_component); - assert(iterator->upstream_component->class); - - /* Pick the appropriate "next" method */ - switch (iterator->upstream_component->class->type) { - case BT_COMPONENT_CLASS_TYPE_SOURCE: - { - struct bt_component_class_source *source_class = - container_of(iterator->upstream_component->class, - struct bt_component_class_source, parent); - - assert(source_class->methods.iterator.next); - next_method = source_class->methods.iterator.next; - break; - } - case BT_COMPONENT_CLASS_TYPE_FILTER: - { - struct bt_component_class_filter *filter_class = - container_of(iterator->upstream_component->class, - struct bt_component_class_filter, parent); - - assert(filter_class->methods.iterator.next); - next_method = filter_class->methods.iterator.next; - break; - } - default: - abort(); - } - - /* - * Call the user's "next" method to get the next notification - * and status. - */ - assert(next_method); - - while (iterator->queue->length == 0) { - BT_LOGD_STR("Calling user's \"next\" method."); - next_return = next_method(priv_iterator); + int status; + + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + BT_ASSERT_PRE_ITER_HAS_STATE_TO_SEEK(iterator); + BT_ASSERT_PRE( + bt_component_borrow_graph(iterator->upstream_component)->config_state != + BT_GRAPH_CONFIGURATION_STATE_CONFIGURING, + "Graph is not configured: %!+g", + bt_component_borrow_graph(iterator->upstream_component)); + BT_ASSERT_PRE( + bt_self_component_port_input_message_iterator_can_seek_ns_from_origin( + iterator, ns_from_origin), + "Message iterator cannot seek nanoseconds from origin: %!+i, " + "ns-from-origin=%" PRId64, iterator, ns_from_origin); + set_self_comp_port_input_msg_iterator_state(iterator, + BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_SEEKING); + + if (iterator->methods.seek_ns_from_origin) { + BT_LIB_LOGD("Calling user's \"seek nanoseconds from origin\" method: " + "%![iter-]+i, ns=%" PRId64, iterator, ns_from_origin); + status = iterator->methods.seek_ns_from_origin(iterator, + ns_from_origin); BT_LOGD("User method returned: status=%s", - bt_notification_iterator_status_string(next_return.status)); - if (next_return.status < 0) { - BT_LOGW_STR("User method failed."); - status = next_return.status; + bt_message_iterator_status_string(status)); + BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK || + status == BT_MESSAGE_ITERATOR_STATUS_ERROR || + status == BT_MESSAGE_ITERATOR_STATUS_NOMEM || + status == BT_MESSAGE_ITERATOR_STATUS_AGAIN, + "Unexpected status: %![iter-]+i, status=%s", + iterator, + bt_common_self_message_iterator_status_string(status)); + } else { + /* Start automatic seeking: seek beginning first */ + BT_ASSERT(iterator->methods.can_seek_beginning(iterator)); + BT_ASSERT(iterator->methods.seek_beginning); + BT_LIB_LOGD("Calling user's \"seek beginning\" method: %!+i", + iterator); + status = iterator->methods.seek_beginning(iterator); + BT_LOGD("User method returned: status=%s", + bt_message_iterator_status_string(status)); + BT_ASSERT_PRE(status == BT_MESSAGE_ITERATOR_STATUS_OK || + status == BT_MESSAGE_ITERATOR_STATUS_ERROR || + status == BT_MESSAGE_ITERATOR_STATUS_NOMEM || + status == BT_MESSAGE_ITERATOR_STATUS_AGAIN, + "Unexpected status: %![iter-]+i, status=%s", + iterator, + bt_common_self_message_iterator_status_string(status)); + switch (status) { + case BT_MESSAGE_ITERATOR_STATUS_OK: + break; + case BT_MESSAGE_ITERATOR_STATUS_ERROR: + case BT_MESSAGE_ITERATOR_STATUS_NOMEM: + case BT_MESSAGE_ITERATOR_STATUS_AGAIN: goto end; + default: + abort(); } - if (iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED || - iterator->state == BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_FINALIZED_AND_ENDED) { - /* - * The user's "next" method, somehow, cancelled - * its own notification iterator. This can - * happen, for example, when the user's method - * removes the port on which there's the - * connection from which the iterator was - * created. In this case, said connection is - * ended, and all its notification iterators are - * finalized. - * - * Only bt_put() the returned notification if - * the status is - * BT_NOTIFICATION_ITERATOR_STATUS_OK because - * otherwise this field could be garbage. - */ - if (next_return.status == - BT_NOTIFICATION_ITERATOR_STATUS_OK) { - bt_put(next_return.notification); - } - - status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED; - goto end; + /* + * Find the first message which has a default clock + * snapshot greater than or equal to the requested + * seeking time, and move the received messages from + * this point in the batch to this iterator's auto-seek + * message queue. + */ + while (!g_queue_is_empty(iterator->auto_seek_msgs)) { + bt_object_put_no_null_check( + g_queue_pop_tail(iterator->auto_seek_msgs)); } - switch (next_return.status) { - case BT_NOTIFICATION_ITERATOR_STATUS_END: - ret = handle_end(iterator); - if (ret) { - BT_LOGW_STR("Cannot handle end of iteration."); - status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; - goto end; - } - - assert(iterator->state == - BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ACTIVE); - iterator->state = BT_PRIVATE_CONNECTION_NOTIFICATION_ITERATOR_STATE_ENDED; - - if (iterator->queue->length == 0) { - status = BT_NOTIFICATION_ITERATOR_STATUS_END; - } - - BT_LOGD("Set new status: status=%s", - bt_notification_iterator_status_string(status)); - goto end; - case BT_NOTIFICATION_ITERATOR_STATUS_AGAIN: - status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN; - goto end; - case BT_NOTIFICATION_ITERATOR_STATUS_OK: - if (!next_return.notification) { - BT_LOGW_STR("User method returned BT_NOTIFICATION_ITERATOR_STATUS_OK, but notification is NULL."); - status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; - goto end; - } - + status = find_message_ge_ns_from_origin(iterator, + ns_from_origin); + switch (status) { + case BT_MESSAGE_ITERATOR_STATUS_OK: + case BT_MESSAGE_ITERATOR_STATUS_END: /* - * Ignore some notifications which are always - * automatically generated by the notification - * iterator to make sure they have valid values. + * If there are messages in the auto-seek + * message queue, replace the user's "next" + * method with a custom, temporary "next" method + * which returns them. */ - switch (next_return.notification->type) { - case BT_NOTIFICATION_TYPE_DISCARDED_PACKETS: - case BT_NOTIFICATION_TYPE_DISCARDED_EVENTS: - BT_LOGV("Ignoring discarded elements notification returned by notification iterator's \"next\" method: " - "notif-type=%s", - bt_notification_type_string(next_return.notification->type)); - BT_PUT(next_return.notification); - continue; - default: - break; + if (!g_queue_is_empty(iterator->auto_seek_msgs)) { + iterator->methods.next = + (bt_self_component_port_input_message_iterator_next_method) + post_auto_seek_next; } /* - * We know the notification is valid. Before we - * push it to the head of the queue, push the - * appropriate automatic notifications if any. + * `BT_MESSAGE_ITERATOR_STATUS_END` becomes + * `BT_MESSAGE_ITERATOR_STATUS_OK`: the next + * time this iterator's "next" method is called, + * it will return + * `BT_MESSAGE_ITERATOR_STATUS_END`. */ - ret = enqueue_notification_and_automatic(iterator, - next_return.notification); - BT_PUT(next_return.notification); - if (ret) { - BT_LOGW("Cannot enqueue notification and automatic notifications."); - status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; - goto end; - } + status = BT_MESSAGE_ITERATOR_STATUS_OK; break; + case BT_MESSAGE_ITERATOR_STATUS_ERROR: + case BT_MESSAGE_ITERATOR_STATUS_NOMEM: + case BT_MESSAGE_ITERATOR_STATUS_AGAIN: + goto end; default: - /* Unknown non-error status */ abort(); } } end: + set_iterator_state_after_seeking(iterator, status); return status; } -enum bt_notification_iterator_status -bt_notification_iterator_next(struct bt_notification_iterator *iterator) +static inline +bt_self_component_port_input_message_iterator * +borrow_output_port_message_iterator_upstream_iterator( + struct bt_port_output_message_iterator *iterator) { - enum bt_notification_iterator_status status; - - if (!iterator) { - BT_LOGW_STR("Invalid parameter: notification iterator is NULL."); - status = BT_NOTIFICATION_ITERATOR_STATUS_INVALID; - goto end; - } - - BT_LOGD("Notification iterator's \"next\": iter-addr=%p", iterator); - - switch (iterator->type) { - case BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION: - { - struct bt_notification_iterator_private_connection *priv_conn_iter = - (void *) iterator; - struct bt_notification *notif; - - /* - * Make sure that the iterator's queue contains at least - * one notification. - */ - status = ensure_queue_has_notifications(priv_conn_iter); - if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) { - goto end; - } - - /* - * Move the notification at the tail of the queue to the - * iterator's current notification. - */ - assert(priv_conn_iter->queue->length > 0); - notif = g_queue_pop_tail(priv_conn_iter->queue); - bt_notification_iterator_replace_current_notification( - iterator, notif); - bt_put(notif); - break; - } - case BT_NOTIFICATION_ITERATOR_TYPE_OUTPUT_PORT: - { - struct bt_notification_iterator_output_port *out_port_iter = - (void *) iterator; - - /* - * Keep current notification in case there's an error: - * restore this notification so that the current - * notification is not changed from the user's point of - * view. - */ - struct bt_notification *old_notif = - bt_get(bt_notification_iterator_borrow_current_notification(iterator)); - enum bt_graph_status graph_status; + struct bt_component_class_sink_colander_priv_data *colander_data; - /* - * Put current notification since it's possibly - * about to be replaced by a new one by the - * colander sink. - */ - bt_notification_iterator_replace_current_notification( - iterator, NULL); - graph_status = bt_graph_consume_sink_no_check( - out_port_iter->graph, - out_port_iter->colander); - switch (graph_status) { - case BT_GRAPH_STATUS_CANCELED: - status = BT_NOTIFICATION_ITERATOR_STATUS_CANCELED; - break; - case BT_GRAPH_STATUS_AGAIN: - status = BT_NOTIFICATION_ITERATOR_STATUS_AGAIN; - break; - case BT_GRAPH_STATUS_END: - status = BT_NOTIFICATION_ITERATOR_STATUS_END; - break; - case BT_GRAPH_STATUS_NOMEM: - status = BT_NOTIFICATION_ITERATOR_STATUS_NOMEM; - break; - case BT_GRAPH_STATUS_OK: - status = BT_NOTIFICATION_ITERATOR_STATUS_OK; - assert(bt_notification_iterator_borrow_current_notification(iterator)); - break; - default: - /* Other errors */ - status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR; - } - - if (status != BT_NOTIFICATION_ITERATOR_STATUS_OK) { - /* Error/exception: restore old notification */ - bt_notification_iterator_replace_current_notification( - iterator, old_notif); - } - - bt_put(old_notif); - break; - } - default: - BT_LOGF("Unknown notification iterator type: addr=%p, type=%d", - iterator, iterator->type); - abort(); - } - -end: - return status; + BT_ASSERT(iterator); + colander_data = (void *) iterator->colander->parent.user_data; + BT_ASSERT(colander_data); + BT_ASSERT(colander_data->msg_iter); + return colander_data->msg_iter; } -struct bt_component *bt_private_connection_notification_iterator_get_component( - struct bt_notification_iterator *iterator) +bt_bool bt_port_output_message_iterator_can_seek_ns_from_origin( + struct bt_port_output_message_iterator *iterator, + int64_t ns_from_origin) { - struct bt_component *comp = NULL; - struct bt_notification_iterator_private_connection *iter_priv_conn; - - if (!iterator) { - BT_LOGW_STR("Invalid parameter: notification iterator is NULL."); - goto end; - } - - if (iterator->type != BT_NOTIFICATION_ITERATOR_TYPE_PRIVATE_CONNECTION) { - BT_LOGW_STR("Invalid parameter: notification iterator was not created from a private connection."); - goto end; - } - - iter_priv_conn = (void *) iterator; - comp = bt_get(iter_priv_conn->upstream_component); - -end: - return comp; + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return bt_self_component_port_input_message_iterator_can_seek_ns_from_origin( + borrow_output_port_message_iterator_upstream_iterator( + iterator), ns_from_origin); } -struct bt_private_component * -bt_private_connection_private_notification_iterator_get_private_component( - struct bt_private_connection_private_notification_iterator *private_iterator) +bt_bool bt_port_output_message_iterator_can_seek_beginning( + struct bt_port_output_message_iterator *iterator) { - return bt_private_component_from_component( - bt_private_connection_notification_iterator_get_component( - (void *) bt_private_connection_notification_iterator_borrow_from_private(private_iterator))); + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return bt_self_component_port_input_message_iterator_can_seek_beginning( + borrow_output_port_message_iterator_upstream_iterator( + iterator)); } -static -void bt_output_port_notification_iterator_destroy(struct bt_object *obj) +enum bt_message_iterator_status bt_port_output_message_iterator_seek_ns_from_origin( + struct bt_port_output_message_iterator *iterator, + int64_t ns_from_origin) { - struct bt_notification_iterator_output_port *iterator = - (void *) container_of(obj, struct bt_notification_iterator, base); - - BT_LOGD("Destroying output port notification iterator object: addr=%p", - iterator); - BT_LOGD_STR("Putting graph."); - bt_put(iterator->graph); - BT_LOGD_STR("Putting output port."); - bt_put(iterator->output_port); - BT_LOGD_STR("Putting colander sink component."); - bt_put(iterator->colander); - destroy_base_notification_iterator(obj); + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return bt_self_component_port_input_message_iterator_seek_ns_from_origin( + borrow_output_port_message_iterator_upstream_iterator(iterator), + ns_from_origin); } -struct bt_notification_iterator *bt_output_port_notification_iterator_create( - struct bt_port *output_port, - const char *colander_component_name, - const enum bt_notification_type *notification_types) +enum bt_message_iterator_status bt_port_output_message_iterator_seek_beginning( + struct bt_port_output_message_iterator *iterator) { - struct bt_notification_iterator *iterator_base = NULL; - struct bt_notification_iterator_output_port *iterator = NULL; - struct bt_component_class *colander_comp_cls = NULL; - struct bt_component *output_port_comp = NULL; - struct bt_component *colander_comp; - struct bt_graph *graph = NULL; - enum bt_graph_status graph_status; - const char *colander_comp_name; - struct bt_port *colander_in_port = NULL; - struct bt_component_class_sink_colander_data colander_data; - - if (!output_port) { - BT_LOGW_STR("Invalid parameter: port is NULL."); - goto error; - } - - if (bt_port_get_type(output_port) != BT_PORT_TYPE_OUTPUT) { - BT_LOGW_STR("Invalid parameter: port is not an output port."); - goto error; - } - - output_port_comp = bt_port_get_component(output_port); - if (!output_port_comp) { - BT_LOGW("Cannot get output port's component: " - "port-addr=%p, port-name=\"%s\"", - output_port, bt_port_get_name(output_port)); - goto error; - } - - graph = bt_component_get_graph(output_port_comp); - assert(graph); - - /* Create notification iterator */ - BT_LOGD("Creating notification iterator on output port: " - "comp-addr=%p, comp-name\"%s\", port-addr=%p, port-name=\"%s\"", - output_port_comp, bt_component_get_name(output_port_comp), - output_port, bt_port_get_name(output_port)); - iterator = g_new0(struct bt_notification_iterator_output_port, 1); - if (!iterator) { - BT_LOGE_STR("Failed to allocate one output port notification iterator."); - goto error; - } - - init_notification_iterator((void *) iterator, - BT_NOTIFICATION_ITERATOR_TYPE_OUTPUT_PORT, - bt_output_port_notification_iterator_destroy); - - /* Create colander component */ - colander_comp_cls = bt_component_class_sink_colander_get(); - if (!colander_comp_cls) { - BT_LOGW("Cannot get colander sink component class."); - goto error; - } - - BT_MOVE(iterator->graph, graph); - iterator_base = (void *) iterator; - colander_comp_name = - colander_component_name ? colander_component_name : "colander"; - colander_data.notification = &iterator_base->current_notification; - colander_data.notification_types = notification_types; - graph_status = bt_graph_add_component_with_init_method_data( - iterator->graph, colander_comp_cls, colander_comp_name, - NULL, &colander_data, &iterator->colander); - if (graph_status != BT_GRAPH_STATUS_OK) { - BT_LOGW("Cannot add colander sink component to graph: " - "graph-addr=%p, name=\"%s\", graph-status=%s", - iterator->graph, colander_comp_name, - bt_graph_status_string(graph_status)); - goto error; - } - - /* - * Connect provided output port to the colander component's - * input port. - */ - colander_in_port = bt_component_sink_get_input_port_by_index( - iterator->colander, 0); - assert(colander_in_port); - graph_status = bt_graph_connect_ports(iterator->graph, - output_port, colander_in_port, NULL); - if (graph_status != BT_GRAPH_STATUS_OK) { - BT_LOGW("Cannot add colander sink component to graph: " - "graph-addr=%p, name=\"%s\", graph-status=%s", - iterator->graph, colander_comp_name, - bt_graph_status_string(graph_status)); - goto error; - } - - /* - * At this point everything went fine. Make the graph - * nonconsumable forever so that only this notification iterator - * can consume (thanks to bt_graph_consume_sink_no_check()). - * This avoids leaking the notification created by the colander - * sink and moved to the base notification iterator's current - * notification member. - */ - bt_graph_set_can_consume(iterator->graph, BT_FALSE); - goto end; - -error: - if (iterator && iterator->graph && iterator->colander) { - int ret; - - /* Remove created colander component from graph if any */ - colander_comp = iterator->colander; - BT_PUT(iterator->colander); + BT_ASSERT_PRE_NON_NULL(iterator, "Message iterator"); + return bt_self_component_port_input_message_iterator_seek_beginning( + borrow_output_port_message_iterator_upstream_iterator( + iterator)); +} - /* - * At this point the colander component's reference - * count is 0 because iterator->colander was the only - * owner. We also know that it is not connected because - * this is the last operation before this function - * succeeds. - * - * Since we honor the preconditions here, - * bt_graph_remove_unconnected_component() always - * succeeds. - */ - ret = bt_graph_remove_unconnected_component(iterator->graph, - colander_comp); - assert(ret == 0); - } +void bt_port_output_message_iterator_get_ref( + const struct bt_port_output_message_iterator *iterator) +{ + bt_object_get_ref(iterator); +} - BT_PUT(iterator); +void bt_port_output_message_iterator_put_ref( + const struct bt_port_output_message_iterator *iterator) +{ + bt_object_put_ref(iterator); +} -end: - bt_put(colander_in_port); - bt_put(colander_comp_cls); - bt_put(output_port_comp); - bt_put(graph); - return (void *) iterator; +void bt_self_component_port_input_message_iterator_get_ref( + const struct bt_self_component_port_input_message_iterator *iterator) +{ + bt_object_get_ref(iterator); } -struct bt_notification_iterator * -bt_private_connection_notification_iterator_from_private( - struct bt_private_connection_private_notification_iterator *private_notification_iterator) +void bt_self_component_port_input_message_iterator_put_ref( + const struct bt_self_component_port_input_message_iterator *iterator) { - return bt_get( - bt_private_connection_notification_iterator_borrow_from_private( - private_notification_iterator)); + bt_object_put_ref(iterator); }