+ case BT_MESSAGE_TYPE_STREAM_BEGINNING:
+ case BT_MESSAGE_TYPE_STREAM_END:
+ /* Ignore */
+ goto skip_msg;
+ default:
+ abort();
+ }
+
+ BT_ASSERT(clk_snapshot);
+ ret = bt_clock_snapshot_get_ns_from_origin(clk_snapshot,
+ &msg_ns_from_origin);
+ if (ret) {
+ status = BT_MESSAGE_ITERATOR_STATUS_ERROR;
+ goto end;
+ }
+
+ if (msg_ns_from_origin >= ns_from_origin) {
+ *got_first = true;
+ goto push_msg;
+ }
+
+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_ASSERT(!msg || status != BT_MESSAGE_ITERATOR_STATUS_OK);
+ return status;
+}
+
+static
+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)
+{
+ 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;
+
+ BT_ASSERT(iterator);
+ memset(&messages[0], 0, sizeof(messages[0]) * MSG_BATCH_SIZE);
+
+ /*
+ * 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);
+
+ BT_ASSERT(iterator->methods.next);
+
+ 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));
+
+#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();
+ }
+
+ 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;
+ }
+
+ 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;
+ }
+ }
+ }
+
+end:
+ for (i = 0; i < user_count; i++) {
+ if (messages[i]) {
+ bt_object_put_no_null_check(messages[i]);
+ }
+ }
+
+ set_self_comp_port_input_msg_iterator_state(iterator, init_state);
+ return status;
+}
+
+static
+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)
+{
+ BT_ASSERT(!g_queue_is_empty(iterator->auto_seek_msgs));
+ *count = 0;
+
+ /*
+ * Move auto-seek messages to the output array (which is this
+ * iterator's base message array).
+ */
+ while (capacity > 0 && !g_queue_is_empty(iterator->auto_seek_msgs)) {
+ msgs[*count] = g_queue_pop_tail(iterator->auto_seek_msgs);
+ capacity--;
+ (*count)++;
+ }
+
+ BT_ASSERT(*count > 0);
+
+ 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;
+
+ 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;
+
+ iterator->methods.next =
+ (bt_self_component_port_input_message_iterator_next_method)
+ flt_comp_cls->methods.msg_iter_next;
+ break;
+ }
+ default:
+ abort();
+ }
+ }