[babeltrace.git] / src / plugins / utils / trimmer / trimmer.c

/*
 * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
 * Copyright 2019 Philippe Proulx <pproulx@efficios.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

#define BT_LOG_TAG "PLUGIN/FLT.UTILS.TRIMMER"
#include "logging.h"

#include "compat/utc.h"
#include "compat/time.h"
#include <babeltrace2/babeltrace.h>
#include "common/common.h"
#include "common/assert.h"
#include <stdint.h>
#include <inttypes.h>
#include <glib.h>

#include "trimmer.h"

#define NS_PER_S	INT64_C(1000000000)

static const char * const in_port_name = "in";

struct trimmer_time {
	unsigned int hour, minute, second, ns;
};

struct trimmer_bound {
	/*
	 * Nanoseconds from origin, valid if `is_set` is set and
	 * `is_infinite` is false.
	 */
	int64_t ns_from_origin;

	/* True if this bound's full time (`ns_from_origin`) is set */
	bool is_set;

	/*
	 * True if this bound represents the infinity (negative or
	 * positive depending on which bound it is). If this is true,
	 * then we don't care about `ns_from_origin` above.
	 */
	bool is_infinite;

	/*
	 * This bound's time without the date; this time is used to set
	 * `ns_from_origin` once we know the date.
	 */
	struct trimmer_time time;
};

struct trimmer_comp {
	struct trimmer_bound begin, end;
	bool is_gmt;
};

enum trimmer_iterator_state {
	/*
	 * Find the first message's date and set the bounds's times
	 * accordingly.
	 */
	TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN,

	/*
	 * Initially seek to the trimming range's beginning time.
	 */
	TRIMMER_ITERATOR_STATE_SEEK_INITIALLY,

	/*
	 * Fill the output message queue for as long as received input
	 * messages are within the trimming time range.
	 */
	TRIMMER_ITERATOR_STATE_TRIM,

	/* Flush the remaining messages in the output message queue */
	TRIMMER_ITERATOR_STATE_ENDING,

	/* Trimming operation and message iterator is ended */
	TRIMMER_ITERATOR_STATE_ENDED,
};

struct trimmer_iterator {
	/* Weak */
	struct trimmer_comp *trimmer_comp;

	/* Weak */
	bt_self_message_iterator *self_msg_iter;

	enum trimmer_iterator_state state;

	/* Owned by this */
	bt_self_component_port_input_message_iterator *upstream_iter;
	struct trimmer_bound begin, end;

	/*
	 * Queue of `const bt_message *` (owned by the queue).
	 *
	 * This is where the trimming operation pushes the messages to
	 * output by this message iterator.
	 */
	GQueue *output_messages;

	/*
	 * Hash table of `bt_stream *` (weak) to
	 * `struct trimmer_iterator_stream_state *` (owned by the HT).
	 */
	GHashTable *stream_states;
};

struct trimmer_iterator_stream_state {
	/*
	 * True if both stream beginning and initial stream activity
	 * beginning messages were pushed for this stream.
	 */
	bool inited;

	/*
	 * True if the last pushed message for this stream was a stream
	 * activity end message.
	 */
	bool last_msg_is_stream_activity_end;

	/*
	 * Time to use for a generated stream end activity message when
	 * ending the stream.
	 */
	int64_t stream_act_end_ns_from_origin;

	/* Weak */
	const bt_stream *stream;

	/* Owned by this (`NULL` initially and between packets) */
	const bt_packet *cur_packet;

	/* Owned by this */
	const bt_message *stream_beginning_msg;
};

static
void destroy_trimmer_comp(struct trimmer_comp *trimmer_comp)
{
	BT_ASSERT(trimmer_comp);
	g_free(trimmer_comp);
}

static
struct trimmer_comp *create_trimmer_comp(void)
{
	return g_new0(struct trimmer_comp, 1);
}

BT_HIDDEN
void trimmer_finalize(bt_self_component_filter *self_comp)
{
	struct trimmer_comp *trimmer_comp =
		bt_self_component_get_data(
			bt_self_component_filter_as_self_component(self_comp));

	if (trimmer_comp) {
		destroy_trimmer_comp(trimmer_comp);
	}
}

/*
 * Sets the time (in ns from origin) of a trimmer bound from date and
 * time components.
 *
 * Returns a negative value if anything goes wrong.
 */
static
int set_bound_ns_from_origin(struct trimmer_bound *bound,
		unsigned int year, unsigned int month, unsigned int day,
		unsigned int hour, unsigned int minute, unsigned int second,
		unsigned int ns, bool is_gmt)
{
	int ret = 0;
	time_t result;
	struct tm tm = {
		.tm_sec = second,
		.tm_min = minute,
		.tm_hour = hour,
		.tm_mday = day,
		.tm_mon = month - 1,
		.tm_year = year - 1900,
		.tm_isdst = -1,
	};

	if (is_gmt) {
		result = bt_timegm(&tm);
	} else {
		result = mktime(&tm);
	}

	if (result < 0) {
		ret = -1;
		goto end;
	}

	BT_ASSERT(bound);
	bound->ns_from_origin = (int64_t) result;
	bound->ns_from_origin *= NS_PER_S;
	bound->ns_from_origin += ns;
	bound->is_set = true;

end:
	return ret;
}

/*
 * Parses a timestamp, figuring out its format.
 *
 * Returns a negative value if anything goes wrong.
 *
 * Expected formats:
 *
 *     YYYY-MM-DD hh:mm[:ss[.ns]]
 *     [hh:mm:]ss[.ns]
 *     [-]s[.ns]
 *
 * TODO: Check overflows.
 */
static
int set_bound_from_str(const char *str, struct trimmer_bound *bound,
		bool is_gmt)
{
	int ret = 0;
	int s_ret;
	unsigned int year, month, day, hour, minute, second, ns;
	char dummy;

	/* Try `YYYY-MM-DD hh:mm:ss.ns` format */
	s_ret = sscanf(str, "%u-%u-%u %u:%u:%u.%u%c", &year, &month, &day,
		&hour, &minute, &second, &ns, &dummy);
	if (s_ret == 7) {
		ret = set_bound_ns_from_origin(bound, year, month, day,
			hour, minute, second, ns, is_gmt);
		goto end;
	}

	/* Try `YYYY-MM-DD hh:mm:ss` format */
	s_ret = sscanf(str, "%u-%u-%u %u:%u:%u%c", &year, &month, &day,
		&hour, &minute, &second, &dummy);
	if (s_ret == 6) {
		ret = set_bound_ns_from_origin(bound, year, month, day,
			hour, minute, second, 0, is_gmt);
		goto end;
	}

	/* Try `YYYY-MM-DD hh:mm` format */
	s_ret = sscanf(str, "%u-%u-%u %u:%u%c", &year, &month, &day,
		&hour, &minute, &dummy);
	if (s_ret == 5) {
		ret = set_bound_ns_from_origin(bound, year, month, day,
			hour, minute, 0, 0, is_gmt);
		goto end;
	}

	/* Try `YYYY-MM-DD` format */
	s_ret = sscanf(str, "%u-%u-%u%c", &year, &month, &day, &dummy);
	if (s_ret == 3) {
		ret = set_bound_ns_from_origin(bound, year, month, day,
			0, 0, 0, 0, is_gmt);
		goto end;
	}

	/* Try `hh:mm:ss.ns` format */
	s_ret = sscanf(str, "%u:%u:%u.%u%c", &hour, &minute, &second, &ns,
		&dummy);
	if (s_ret == 4) {
		bound->time.hour = hour;
		bound->time.minute = minute;
		bound->time.second = second;
		bound->time.ns = ns;
		goto end;
	}

	/* Try `hh:mm:ss` format */
	s_ret = sscanf(str, "%u:%u:%u%c", &hour, &minute, &second, &dummy);
	if (s_ret == 3) {
		bound->time.hour = hour;
		bound->time.minute = minute;
		bound->time.second = second;
		bound->time.ns = 0;
		goto end;
	}

	/* Try `-s.ns` format */
	s_ret = sscanf(str, "-%u.%u%c", &second, &ns, &dummy);
	if (s_ret == 2) {
		bound->ns_from_origin = -((int64_t) second) * NS_PER_S;
		bound->ns_from_origin -= (int64_t) ns;
		bound->is_set = true;
		goto end;
	}

	/* Try `s.ns` format */
	s_ret = sscanf(str, "%u.%u%c", &second, &ns, &dummy);
	if (s_ret == 2) {
		bound->ns_from_origin = ((int64_t) second) * NS_PER_S;
		bound->ns_from_origin += (int64_t) ns;
		bound->is_set = true;
		goto end;
	}

	/* Try `-s` format */
	s_ret = sscanf(str, "-%u%c", &second, &dummy);
	if (s_ret == 1) {
		bound->ns_from_origin = -((int64_t) second) * NS_PER_S;
		bound->is_set = true;
		goto end;
	}

	/* Try `s` format */
	s_ret = sscanf(str, "%u%c", &second, &dummy);
	if (s_ret == 1) {
		bound->ns_from_origin = (int64_t) second * NS_PER_S;
		bound->is_set = true;
		goto end;
	}

	BT_LOGE("Invalid date/time format: param=\"%s\"", str);
	ret = -1;

end:
	return ret;
}

/*
 * Sets a trimmer bound's properties from a parameter string/integer
 * value.
 *
 * Returns a negative value if anything goes wrong.
 */
static
int set_bound_from_param(const char *param_name, const bt_value *param,
		struct trimmer_bound *bound, bool is_gmt)
{
	int ret;
	const char *arg;
	char tmp_arg[64];

	if (bt_value_is_signed_integer(param)) {
		int64_t value = bt_value_signed_integer_get(param);

		/*
		 * Just convert it to a temporary string to handle
		 * everything the same way.
		 */
		sprintf(tmp_arg, "%" PRId64, value);
		arg = tmp_arg;
	} else if (bt_value_is_string(param)) {
		arg = bt_value_string_get(param);
	} else {
		BT_LOGE("`%s` parameter must be an integer or a string value.",
			param_name);
		ret = -1;
		goto end;
	}

	ret = set_bound_from_str(arg, bound, is_gmt);

end:
	return ret;
}

static
int validate_trimmer_bounds(struct trimmer_bound *begin,
		struct trimmer_bound *end)
{
	int ret = 0;

	BT_ASSERT(begin->is_set);
	BT_ASSERT(end->is_set);

	if (!begin->is_infinite && !end->is_infinite &&
			begin->ns_from_origin > end->ns_from_origin) {
		BT_LOGE("Trimming time range's beginning time is greater than end time: "
			"begin-ns-from-origin=%" PRId64 ", "
			"end-ns-from-origin=%" PRId64,
			begin->ns_from_origin,
			end->ns_from_origin);
		ret = -1;
		goto end;
	}

	if (!begin->is_infinite && begin->ns_from_origin == INT64_MIN) {
		BT_LOGE("Invalid trimming time range's beginning time: "
			"ns-from-origin=%" PRId64,
			begin->ns_from_origin);
		ret = -1;
		goto end;
	}

	if (!end->is_infinite && end->ns_from_origin == INT64_MIN) {
		BT_LOGE("Invalid trimming time range's end time: "
			"ns-from-origin=%" PRId64,
			end->ns_from_origin);
		ret = -1;
		goto end;
	}

end:
	return ret;
}

static
int init_trimmer_comp_from_params(struct trimmer_comp *trimmer_comp,
		const bt_value *params)
{
	const bt_value *value;
	int ret = 0;

	BT_ASSERT(params);
        value = bt_value_map_borrow_entry_value_const(params, "gmt");
	if (value) {
		trimmer_comp->is_gmt = (bool) bt_value_bool_get(value);
	}

        value = bt_value_map_borrow_entry_value_const(params, "begin");
	if (value) {
		if (set_bound_from_param("begin", value,
				&trimmer_comp->begin, trimmer_comp->is_gmt)) {
			/* set_bound_from_param() logs errors */
			ret = BT_SELF_COMPONENT_STATUS_ERROR;
			goto end;
		}
	} else {
		trimmer_comp->begin.is_infinite = true;
		trimmer_comp->begin.is_set = true;
	}

        value = bt_value_map_borrow_entry_value_const(params, "end");
	if (value) {
		if (set_bound_from_param("end", value,
				&trimmer_comp->end, trimmer_comp->is_gmt)) {
			/* set_bound_from_param() logs errors */
			ret = BT_SELF_COMPONENT_STATUS_ERROR;
			goto end;
		}
	} else {
		trimmer_comp->end.is_infinite = true;
		trimmer_comp->end.is_set = true;
	}

end:
	if (trimmer_comp->begin.is_set && trimmer_comp->end.is_set) {
		/* validate_trimmer_bounds() logs errors */
		ret = validate_trimmer_bounds(&trimmer_comp->begin,
			&trimmer_comp->end);
	}

	return ret;
}

bt_self_component_status trimmer_init(bt_self_component_filter *self_comp,
		const bt_value *params, void *init_data)
{
	int ret;
	bt_self_component_status status;
	struct trimmer_comp *trimmer_comp = create_trimmer_comp();

	if (!trimmer_comp) {
		status = BT_SELF_COMPONENT_STATUS_NOMEM;
		goto error;
	}

	status = bt_self_component_filter_add_input_port(
		self_comp, in_port_name, NULL, NULL);
	if (status != BT_SELF_COMPONENT_STATUS_OK) {
		goto error;
	}

	status = bt_self_component_filter_add_output_port(
		self_comp, "out", NULL, NULL);
	if (status != BT_SELF_COMPONENT_STATUS_OK) {
		goto error;
	}

	ret = init_trimmer_comp_from_params(trimmer_comp, params);
	if (ret) {
		status = BT_SELF_COMPONENT_STATUS_ERROR;
		goto error;
	}

	bt_self_component_set_data(
		bt_self_component_filter_as_self_component(self_comp),
		trimmer_comp);
	goto end;

error:
	if (status == BT_SELF_COMPONENT_STATUS_OK) {
		status = BT_SELF_COMPONENT_STATUS_ERROR;
	}

	if (trimmer_comp) {
		destroy_trimmer_comp(trimmer_comp);
	}

end:
	return status;
}

static
void destroy_trimmer_iterator(struct trimmer_iterator *trimmer_it)
{
	BT_ASSERT(trimmer_it);
	bt_self_component_port_input_message_iterator_put_ref(
		trimmer_it->upstream_iter);

	if (trimmer_it->output_messages) {
		g_queue_free(trimmer_it->output_messages);
	}

	if (trimmer_it->stream_states) {
		g_hash_table_destroy(trimmer_it->stream_states);
	}

	g_free(trimmer_it);
}

static
void destroy_trimmer_iterator_stream_state(
		struct trimmer_iterator_stream_state *sstate)
{
	BT_ASSERT(sstate);
	BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
	BT_MESSAGE_PUT_REF_AND_RESET(sstate->stream_beginning_msg);
	g_free(sstate);
}

BT_HIDDEN
bt_self_message_iterator_status trimmer_msg_iter_init(
		bt_self_message_iterator *self_msg_iter,
		bt_self_component_filter *self_comp,
		bt_self_component_port_output *port)
{
	bt_self_message_iterator_status status =
		BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
	struct trimmer_iterator *trimmer_it;

	trimmer_it = g_new0(struct trimmer_iterator, 1);
	if (!trimmer_it) {
		status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
		goto end;
	}

	trimmer_it->trimmer_comp = bt_self_component_get_data(
		bt_self_component_filter_as_self_component(self_comp));
	BT_ASSERT(trimmer_it->trimmer_comp);

	if (trimmer_it->trimmer_comp->begin.is_set &&
			trimmer_it->trimmer_comp->end.is_set) {
		/*
		 * Both trimming time range's bounds are set, so skip
		 * the
		 * `TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN`
		 * phase.
		 */
		trimmer_it->state = TRIMMER_ITERATOR_STATE_SEEK_INITIALLY;
	}

	trimmer_it->begin = trimmer_it->trimmer_comp->begin;
	trimmer_it->end = trimmer_it->trimmer_comp->end;
	trimmer_it->upstream_iter =
		bt_self_component_port_input_message_iterator_create(
			bt_self_component_filter_borrow_input_port_by_name(
				self_comp, in_port_name));
	if (!trimmer_it->upstream_iter) {
		status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
		goto end;
	}

	trimmer_it->output_messages = g_queue_new();
	if (!trimmer_it->output_messages) {
		status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
		goto end;
	}

	trimmer_it->stream_states = g_hash_table_new_full(g_direct_hash,
		g_direct_equal, NULL,
		(GDestroyNotify) destroy_trimmer_iterator_stream_state);
	if (!trimmer_it->stream_states) {
		status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
		goto end;
	}

	trimmer_it->self_msg_iter = self_msg_iter;
	bt_self_message_iterator_set_data(self_msg_iter, trimmer_it);

end:
	if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK && trimmer_it) {
		destroy_trimmer_iterator(trimmer_it);
	}

	return status;
}

static inline
int get_msg_ns_from_origin(const bt_message *msg, int64_t *ns_from_origin,
		bool *skip)
{
	const bt_clock_class *clock_class = NULL;
	const bt_clock_snapshot *clock_snapshot = NULL;
	bt_message_stream_activity_clock_snapshot_state sa_cs_state;
	int ret = 0;

	BT_ASSERT(msg);
	BT_ASSERT(ns_from_origin);
	BT_ASSERT(skip);

	switch (bt_message_get_type(msg)) {
	case BT_MESSAGE_TYPE_EVENT:
		clock_class =
			bt_message_event_borrow_stream_class_default_clock_class_const(
				msg);
		if (G_UNLIKELY(!clock_class)) {
			goto error;
		}

		clock_snapshot = bt_message_event_borrow_default_clock_snapshot_const(
			msg);
		break;
	case BT_MESSAGE_TYPE_PACKET_BEGINNING:
		clock_class =
			bt_message_packet_beginning_borrow_stream_class_default_clock_class_const(
				msg);
		if (G_UNLIKELY(!clock_class)) {
			goto error;
		}

		clock_snapshot = bt_message_packet_beginning_borrow_default_clock_snapshot_const(
			msg);
		break;
	case BT_MESSAGE_TYPE_PACKET_END:
		clock_class =
			bt_message_packet_end_borrow_stream_class_default_clock_class_const(
				msg);
		if (G_UNLIKELY(!clock_class)) {
			goto error;
		}

		clock_snapshot = bt_message_packet_end_borrow_default_clock_snapshot_const(
			msg);
		break;
	case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
		clock_class =
			bt_message_discarded_events_borrow_stream_class_default_clock_class_const(
				msg);
		if (G_UNLIKELY(!clock_class)) {
			goto error;
		}

		clock_snapshot = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
			msg);
		break;
	case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
		clock_class =
			bt_message_discarded_packets_borrow_stream_class_default_clock_class_const(
				msg);
		if (G_UNLIKELY(!clock_class)) {
			goto error;
		}

		clock_snapshot = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
			msg);
		break;
	case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
		clock_class =
			bt_message_stream_activity_beginning_borrow_stream_class_default_clock_class_const(
				msg);
		if (G_UNLIKELY(!clock_class)) {
			goto error;
		}

		sa_cs_state = bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
			msg, &clock_snapshot);
		if (sa_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN ||
				sa_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE) {
			/* Lowest possible time to always include them */
			*ns_from_origin = INT64_MIN;
			goto no_clock_snapshot;
		}

		break;
	case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
		clock_class =
			bt_message_stream_activity_end_borrow_stream_class_default_clock_class_const(
				msg);
		if (G_UNLIKELY(!clock_class)) {
			goto error;
		}

		sa_cs_state = bt_message_stream_activity_end_borrow_default_clock_snapshot_const(
			msg, &clock_snapshot);
		if (sa_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_UNKNOWN) {
			/* Lowest time to always include it */
			*ns_from_origin = INT64_MIN;
			goto no_clock_snapshot;
		} else if (sa_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE) {
			/* Greatest time to always exclude it */
			*ns_from_origin = INT64_MAX;
			goto no_clock_snapshot;
		}

		break;
	case BT_MESSAGE_TYPE_MESSAGE_ITERATOR_INACTIVITY:
		clock_snapshot =
			bt_message_message_iterator_inactivity_borrow_default_clock_snapshot_const(
				msg);
		break;
	default:
		goto no_clock_snapshot;
	}

	ret = bt_clock_snapshot_get_ns_from_origin(clock_snapshot,
		ns_from_origin);
	if (G_UNLIKELY(ret)) {
		goto error;
	}

	goto end;

no_clock_snapshot:
	*skip = true;
	goto end;

error:
	ret = -1;

end:
	return ret;
}

static inline
void put_messages(bt_message_array_const msgs, uint64_t count)
{
	uint64_t i;

	for (i = 0; i < count; i++) {
		BT_MESSAGE_PUT_REF_AND_RESET(msgs[i]);
	}
}

static inline
int set_trimmer_iterator_bound(struct trimmer_bound *bound,
		int64_t ns_from_origin, bool is_gmt)
{
	struct tm tm;
	time_t time_seconds = (time_t) (ns_from_origin / NS_PER_S);
	int ret = 0;

	BT_ASSERT(!bound->is_set);
	errno = 0;

	/* We only need to extract the date from this time */
	if (is_gmt) {
		bt_gmtime_r(&time_seconds, &tm);
	} else {
		bt_localtime_r(&time_seconds, &tm);
	}

	if (errno) {
		BT_LOGE_ERRNO("Cannot convert timestamp to date and time",
			"ts=%" PRId64, (int64_t) time_seconds);
		ret = -1;
		goto end;
	}

	ret = set_bound_ns_from_origin(bound, tm.tm_year + 1900, tm.tm_mon + 1,
		tm.tm_mday, bound->time.hour, bound->time.minute,
		bound->time.second, bound->time.ns, is_gmt);

end:
	return ret;
}

static
bt_self_message_iterator_status state_set_trimmer_iterator_bounds(
		struct trimmer_iterator *trimmer_it)
{
	bt_message_iterator_status upstream_iter_status =
		BT_MESSAGE_ITERATOR_STATUS_OK;
	struct trimmer_comp *trimmer_comp = trimmer_it->trimmer_comp;
	bt_message_array_const msgs;
	uint64_t count = 0;
	int64_t ns_from_origin = INT64_MIN;
	uint64_t i;
	int ret;

	BT_ASSERT(!trimmer_it->begin.is_set ||
		!trimmer_it->end.is_set);

	while (true) {
		upstream_iter_status =
			bt_self_component_port_input_message_iterator_next(
				trimmer_it->upstream_iter, &msgs, &count);
		if (upstream_iter_status != BT_MESSAGE_ITERATOR_STATUS_OK) {
			goto end;
		}

		for (i = 0; i < count; i++) {
			const bt_message *msg = msgs[i];
			bool skip = false;
			int ret;

			ret = get_msg_ns_from_origin(msg, &ns_from_origin,
				&skip);
			if (ret) {
				goto error;
			}

			if (skip) {
				continue;
			}

			BT_ASSERT(ns_from_origin != INT64_MIN &&
				ns_from_origin != INT64_MAX);
			put_messages(msgs, count);
			goto found;
		}

		put_messages(msgs, count);
	}

found:
	if (!trimmer_it->begin.is_set) {
		BT_ASSERT(!trimmer_it->begin.is_infinite);
		ret = set_trimmer_iterator_bound(&trimmer_it->begin,
			ns_from_origin, trimmer_comp->is_gmt);
		if (ret) {
			goto error;
		}
	}

	if (!trimmer_it->end.is_set) {
		BT_ASSERT(!trimmer_it->end.is_infinite);
		ret = set_trimmer_iterator_bound(&trimmer_it->end,
			ns_from_origin, trimmer_comp->is_gmt);
		if (ret) {
			goto error;
		}
	}

	ret = validate_trimmer_bounds(&trimmer_it->begin,
		&trimmer_it->end);
	if (ret) {
		goto error;
	}

	goto end;

error:
	put_messages(msgs, count);
	upstream_iter_status = BT_MESSAGE_ITERATOR_STATUS_ERROR;

end:
	return (int) upstream_iter_status;
}

static
bt_self_message_iterator_status state_seek_initially(
		struct trimmer_iterator *trimmer_it)
{
	bt_self_message_iterator_status status =
		BT_SELF_MESSAGE_ITERATOR_STATUS_OK;

	BT_ASSERT(trimmer_it->begin.is_set);

	if (trimmer_it->begin.is_infinite) {
		if (!bt_self_component_port_input_message_iterator_can_seek_beginning(
				trimmer_it->upstream_iter)) {
			BT_LOGE_STR("Cannot make upstream message iterator initially seek its beginning.");
			status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
			goto end;
		}

		status = (int) bt_self_component_port_input_message_iterator_seek_beginning(
			trimmer_it->upstream_iter);
	} else {
		if (!bt_self_component_port_input_message_iterator_can_seek_ns_from_origin(
				trimmer_it->upstream_iter,
				trimmer_it->begin.ns_from_origin)) {
			BT_LOGE("Cannot make upstream message iterator initially seek: "
				"seek-ns-from-origin=%" PRId64,
				trimmer_it->begin.ns_from_origin);
			status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
			goto end;
		}

		status = (int) bt_self_component_port_input_message_iterator_seek_ns_from_origin(
			trimmer_it->upstream_iter, trimmer_it->begin.ns_from_origin);
	}

	if (status == BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
		trimmer_it->state = TRIMMER_ITERATOR_STATE_TRIM;
	}

end:
	return status;
}

static inline
void push_message(struct trimmer_iterator *trimmer_it, const bt_message *msg)
{
	g_queue_push_head(trimmer_it->output_messages, (void *) msg);
}

static inline
const bt_message *pop_message(struct trimmer_iterator *trimmer_it)
{
	return g_queue_pop_tail(trimmer_it->output_messages);
}

static inline
int clock_raw_value_from_ns_from_origin(const bt_clock_class *clock_class,
		int64_t ns_from_origin, uint64_t *raw_value)
{

	int64_t cc_offset_s;
	uint64_t cc_offset_cycles;
	uint64_t cc_freq;

	bt_clock_class_get_offset(clock_class, &cc_offset_s, &cc_offset_cycles);
	cc_freq = bt_clock_class_get_frequency(clock_class);
	return bt_common_clock_value_from_ns_from_origin(cc_offset_s,
		cc_offset_cycles, cc_freq, ns_from_origin, raw_value);
}

static inline
bt_self_message_iterator_status end_stream(struct trimmer_iterator *trimmer_it,
		struct trimmer_iterator_stream_state *sstate)
{
	bt_self_message_iterator_status status =
		BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
	uint64_t raw_value;
	const bt_clock_class *clock_class;
	int ret;
	bt_message *msg = NULL;

	BT_ASSERT(!trimmer_it->end.is_infinite);

	if (!sstate->stream) {
		goto end;
	}

	if (sstate->cur_packet) {
		/*
		 * The last message could not have been a stream
		 * activity end message if we have a current packet.
		 */
		BT_ASSERT(!sstate->last_msg_is_stream_activity_end);

		/*
		 * Create and push a packet end message, making its time
		 * the trimming range's end time.
		 */
		clock_class = bt_stream_class_borrow_default_clock_class_const(
			bt_stream_borrow_class_const(sstate->stream));
		BT_ASSERT(clock_class);
		ret = clock_raw_value_from_ns_from_origin(clock_class,
			trimmer_it->end.ns_from_origin, &raw_value);
		if (ret) {
			status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
			goto end;
		}

		msg = bt_message_packet_end_create_with_default_clock_snapshot(
			trimmer_it->self_msg_iter, sstate->cur_packet,
			raw_value);
		if (!msg) {
			status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
			goto end;
		}

		push_message(trimmer_it, msg);
		msg = NULL;
		BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);

		/*
		 * Because we generated a packet end message, set the
		 * stream activity end message's time to use to the
		 * trimming range's end time (this packet end message's
		 * time).
		 */
		sstate->stream_act_end_ns_from_origin =
			trimmer_it->end.ns_from_origin;
	}

	if (!sstate->last_msg_is_stream_activity_end) {
		/* Create and push a stream activity end message */
		msg = bt_message_stream_activity_end_create(
			trimmer_it->self_msg_iter, sstate->stream);
		if (!msg) {
			status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
			goto end;
		}

		clock_class = bt_stream_class_borrow_default_clock_class_const(
			bt_stream_borrow_class_const(sstate->stream));
		BT_ASSERT(clock_class);

		if (sstate->stream_act_end_ns_from_origin == INT64_MIN) {
			/*
			 * We received at least what is necessary to
			 * have a stream state (stream beginning and
			 * stream activity beginning messages), but
			 * nothing else: use the trimmer range's end
			 * time.
			 */
			sstate->stream_act_end_ns_from_origin =
				trimmer_it->end.ns_from_origin;
		}

		ret = clock_raw_value_from_ns_from_origin(clock_class,
			sstate->stream_act_end_ns_from_origin, &raw_value);
		if (ret) {
			status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
			goto end;
		}

		bt_message_stream_activity_end_set_default_clock_snapshot(
			msg, raw_value);
		push_message(trimmer_it, msg);
		msg = NULL;
	}

	/* Create and push a stream end message */
	msg = bt_message_stream_end_create(trimmer_it->self_msg_iter,
		sstate->stream);
	if (!msg) {
		status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
		goto end;
	}

	push_message(trimmer_it, msg);
	msg = NULL;

	/*
	 * Just to make sure that we don't use this stream state again
	 * in the future without an obvious error.
	 */
	sstate->stream = NULL;

end:
	bt_message_put_ref(msg);
	return status;
}

static inline
bt_self_message_iterator_status end_iterator_streams(
		struct trimmer_iterator *trimmer_it)
{
	bt_self_message_iterator_status status =
		BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
	GHashTableIter iter;
	gpointer key, sstate;

	if (trimmer_it->end.is_infinite) {
		/*
		 * An infinite trimming range's end time guarantees that
		 * we received (and pushed) all the appropriate end
		 * messages.
		 */
		goto remove_all;
	}

	/*
	 * End each stream and then remove them from the hash table of
	 * stream states to release unneeded references.
	 */
	g_hash_table_iter_init(&iter, trimmer_it->stream_states);

	while (g_hash_table_iter_next(&iter, &key, &sstate)) {
		status = end_stream(trimmer_it, sstate);
		if (status) {
			goto end;
		}
	}

remove_all:
	g_hash_table_remove_all(trimmer_it->stream_states);

end:
	return status;
}

static inline
bt_self_message_iterator_status create_stream_beginning_activity_message(
		struct trimmer_iterator *trimmer_it,
		const bt_stream *stream,
		const bt_clock_class *clock_class, bt_message **msg)
{
	bt_message *local_msg;
	bt_self_message_iterator_status status =
		BT_SELF_MESSAGE_ITERATOR_STATUS_OK;

	BT_ASSERT(msg);
	BT_ASSERT(!trimmer_it->begin.is_infinite);

	local_msg = bt_message_stream_activity_beginning_create(
		trimmer_it->self_msg_iter, stream);
	if (!local_msg) {
		status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
		goto end;
	}

	if (clock_class) {
		int ret;
		uint64_t raw_value;

		ret = clock_raw_value_from_ns_from_origin(clock_class,
			trimmer_it->begin.ns_from_origin, &raw_value);
		if (ret) {
			status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
			bt_message_put_ref(local_msg);
			goto end;
		}

		bt_message_stream_activity_beginning_set_default_clock_snapshot(
			local_msg, raw_value);
	}

	BT_MESSAGE_MOVE_REF(*msg, local_msg);

end:
	return status;
}

/*
 * Makes sure to initialize a stream state, pushing the appropriate
 * initial messages.
 *
 * `stream_act_beginning_msg` is an initial stream activity beginning
 * message to potentially use, depending on its clock snapshot state.
 * This function consumes `stream_act_beginning_msg` unconditionally.
 */
static inline
bt_self_message_iterator_status ensure_stream_state_is_inited(
		struct trimmer_iterator *trimmer_it,
		struct trimmer_iterator_stream_state *sstate,
		const bt_message *stream_act_beginning_msg)
{
	bt_self_message_iterator_status status =
		BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
	bt_message *new_msg = NULL;
	const bt_clock_class *clock_class =
		bt_stream_class_borrow_default_clock_class_const(
			bt_stream_borrow_class_const(sstate->stream));

	BT_ASSERT(!sstate->inited);

	if (!sstate->stream_beginning_msg) {
		/* No initial stream beginning message: create one */
		sstate->stream_beginning_msg =
			bt_message_stream_beginning_create(
				trimmer_it->self_msg_iter, sstate->stream);
		if (!sstate->stream_beginning_msg) {
			status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
			goto end;
		}
	}

	/* Push initial stream beginning message */
	BT_ASSERT(sstate->stream_beginning_msg);
	push_message(trimmer_it, sstate->stream_beginning_msg);
	sstate->stream_beginning_msg = NULL;

	if (stream_act_beginning_msg) {
		/*
		 * Initial stream activity beginning message exists: if
		 * its time is -inf, then create and push a new one
		 * having the trimming range's beginning time. Otherwise
		 * push it as is (known and unknown).
		 */
		const bt_clock_snapshot *cs;
		bt_message_stream_activity_clock_snapshot_state sa_cs_state;

		sa_cs_state = bt_message_stream_activity_beginning_borrow_default_clock_snapshot_const(
			stream_act_beginning_msg, &cs);
		if (sa_cs_state == BT_MESSAGE_STREAM_ACTIVITY_CLOCK_SNAPSHOT_STATE_INFINITE &&
				!trimmer_it->begin.is_infinite) {
			/*
			 * -inf time: use trimming range's beginning
			 * time (which is not -inf).
			 */
			status = create_stream_beginning_activity_message(
				trimmer_it, sstate->stream, clock_class,
				&new_msg);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}

			push_message(trimmer_it, new_msg);
			new_msg = NULL;
		} else {
			/* Known/unknown: push as is */
			push_message(trimmer_it, stream_act_beginning_msg);
			stream_act_beginning_msg = NULL;
		}
	} else {
		BT_ASSERT(!trimmer_it->begin.is_infinite);

		/*
		 * No stream beginning activity message: create and push
		 * a new message.
	 	 */
		status = create_stream_beginning_activity_message(
			trimmer_it, sstate->stream, clock_class, &new_msg);
		if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
			goto end;
		}

		push_message(trimmer_it, new_msg);
		new_msg = NULL;
	}

	sstate->inited = true;

end:
	bt_message_put_ref(new_msg);
	bt_message_put_ref(stream_act_beginning_msg);
	return status;
}

static inline
bt_self_message_iterator_status ensure_cur_packet_exists(
	struct trimmer_iterator *trimmer_it,
	struct trimmer_iterator_stream_state *sstate,
	const bt_packet *packet)
{
	bt_self_message_iterator_status status =
		BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
	int ret;
	const bt_clock_class *clock_class =
		bt_stream_class_borrow_default_clock_class_const(
			bt_stream_borrow_class_const(sstate->stream));
	bt_message *msg = NULL;
	uint64_t raw_value;

	BT_ASSERT(!trimmer_it->begin.is_infinite);
	BT_ASSERT(!sstate->cur_packet);

	/*
	 * Create and push an initial packet beginning message,
	 * making its time the trimming range's beginning time.
	 */
	ret = clock_raw_value_from_ns_from_origin(clock_class,
		trimmer_it->begin.ns_from_origin, &raw_value);
	if (ret) {
		status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
		goto end;
	}

	msg = bt_message_packet_beginning_create_with_default_clock_snapshot(
		trimmer_it->self_msg_iter, packet, raw_value);
	if (!msg) {
		status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
		goto end;
	}

	push_message(trimmer_it, msg);
	msg = NULL;

	/* Set packet as this stream's current packet */
	sstate->cur_packet = packet;
	bt_packet_get_ref(sstate->cur_packet);

end:
	bt_message_put_ref(msg);
	return status;
}

/*
 * Handles a message which is associated to a given stream state. This
 * _could_ make the iterator's output message queue grow; this could
 * also consume the message without pushing anything to this queue, only
 * modifying the stream state.
 *
 * This function consumes the `msg` reference, _whatever the outcome_.
 *
 * `ns_from_origin` is the message's time, as given by
 * get_msg_ns_from_origin().
 *
 * This function sets `reached_end` if handling this message made the
 * iterator reach the end of the trimming range. Note that the output
 * message queue could contain messages even if this function sets
 * `reached_end`.
 */
static inline
bt_self_message_iterator_status handle_message_with_stream_state(
		struct trimmer_iterator *trimmer_it, const bt_message *msg,
		struct trimmer_iterator_stream_state *sstate,
		int64_t ns_from_origin, bool *reached_end)
{
	bt_self_message_iterator_status status =
		BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
	bt_message_type msg_type = bt_message_get_type(msg);
	int ret;

	switch (msg_type) {
	case BT_MESSAGE_TYPE_EVENT:
		if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
				ns_from_origin > trimmer_it->end.ns_from_origin)) {
			status = end_iterator_streams(trimmer_it);
			*reached_end = true;
			break;
		}

		if (G_UNLIKELY(!sstate->inited)) {
			status = ensure_stream_state_is_inited(trimmer_it,
				sstate, NULL);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}
		}

		if (G_UNLIKELY(!sstate->cur_packet)) {
			const bt_event *event =
				bt_message_event_borrow_event_const(msg);
			const bt_packet *packet = bt_event_borrow_packet_const(
				event);

			status = ensure_cur_packet_exists(trimmer_it, sstate,
				packet);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}
		}

		BT_ASSERT(sstate->cur_packet);
		push_message(trimmer_it, msg);
		msg = NULL;
		break;
	case BT_MESSAGE_TYPE_PACKET_BEGINNING:
		if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
				ns_from_origin > trimmer_it->end.ns_from_origin)) {
			status = end_iterator_streams(trimmer_it);
			*reached_end = true;
			break;
		}

		if (G_UNLIKELY(!sstate->inited)) {
			status = ensure_stream_state_is_inited(trimmer_it,
				sstate, NULL);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}
		}

		BT_ASSERT(!sstate->cur_packet);
		sstate->cur_packet =
			bt_message_packet_beginning_borrow_packet_const(msg);
		bt_packet_get_ref(sstate->cur_packet);
		push_message(trimmer_it, msg);
		msg = NULL;
		break;
	case BT_MESSAGE_TYPE_PACKET_END:
		sstate->stream_act_end_ns_from_origin = ns_from_origin;

		if (G_UNLIKELY(!trimmer_it->end.is_infinite &&
				ns_from_origin > trimmer_it->end.ns_from_origin)) {
			status = end_iterator_streams(trimmer_it);
			*reached_end = true;
			break;
		}

		if (G_UNLIKELY(!sstate->inited)) {
			status = ensure_stream_state_is_inited(trimmer_it,
				sstate, NULL);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}
		}

		if (G_UNLIKELY(!sstate->cur_packet)) {
			const bt_packet *packet =
				bt_message_packet_end_borrow_packet_const(msg);

			status = ensure_cur_packet_exists(trimmer_it, sstate,
				packet);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}
		}

		BT_ASSERT(sstate->cur_packet);
		BT_PACKET_PUT_REF_AND_RESET(sstate->cur_packet);
		push_message(trimmer_it, msg);
		msg = NULL;
		break;
	case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
	case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
	{
		/*
		 * `ns_from_origin` is the message's time range's
		 * beginning time here.
		 */
		int64_t end_ns_from_origin;
		const bt_clock_snapshot *end_cs;

		if (bt_message_get_type(msg) ==
				BT_MESSAGE_TYPE_DISCARDED_EVENTS) {
			/*
			 * Safe to ignore the return value because we
			 * know there's a default clock and it's always
			 * known.
			 */
			end_cs = bt_message_discarded_events_borrow_end_default_clock_snapshot_const(
				msg);
		} else {
			/*
			 * Safe to ignore the return value because we
			 * know there's a default clock and it's always
			 * known.
			 */
			end_cs = bt_message_discarded_packets_borrow_end_default_clock_snapshot_const(
				msg);
		}

		if (bt_clock_snapshot_get_ns_from_origin(end_cs,
				&end_ns_from_origin)) {
			status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
			goto end;
		}

		sstate->stream_act_end_ns_from_origin = end_ns_from_origin;

		if (!trimmer_it->end.is_infinite &&
				ns_from_origin > trimmer_it->end.ns_from_origin) {
			status = end_iterator_streams(trimmer_it);
			*reached_end = true;
			break;
		}

		if (!trimmer_it->end.is_infinite &&
				end_ns_from_origin > trimmer_it->end.ns_from_origin) {
			/*
			 * This message's end time is outside the
			 * trimming time range: replace it with a new
			 * message having an end time equal to the
			 * trimming time range's end and without a
			 * count.
			 */
			const bt_clock_class *clock_class =
				bt_clock_snapshot_borrow_clock_class_const(
					end_cs);
			const bt_clock_snapshot *begin_cs;
			bt_message *new_msg;
			uint64_t end_raw_value;

			ret = clock_raw_value_from_ns_from_origin(clock_class,
				trimmer_it->end.ns_from_origin, &end_raw_value);
			if (ret) {
				status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
				goto end;
			}

			if (msg_type == BT_MESSAGE_TYPE_DISCARDED_EVENTS) {
				begin_cs = bt_message_discarded_events_borrow_beginning_default_clock_snapshot_const(
					msg);
				new_msg = bt_message_discarded_events_create_with_default_clock_snapshots(
					trimmer_it->self_msg_iter,
					sstate->stream,
					bt_clock_snapshot_get_value(begin_cs),
					end_raw_value);
			} else {
				begin_cs = bt_message_discarded_packets_borrow_beginning_default_clock_snapshot_const(
					msg);
				new_msg = bt_message_discarded_packets_create_with_default_clock_snapshots(
					trimmer_it->self_msg_iter,
					sstate->stream,
					bt_clock_snapshot_get_value(begin_cs),
					end_raw_value);
			}

			if (!new_msg) {
				status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
				goto end;
			}

			/* Replace the original message */
			BT_MESSAGE_MOVE_REF(msg, new_msg);
		}

		if (G_UNLIKELY(!sstate->inited)) {
			status = ensure_stream_state_is_inited(trimmer_it,
				sstate, NULL);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}
		}

		push_message(trimmer_it, msg);
		msg = NULL;
		break;
	}
	case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
		if (!trimmer_it->end.is_infinite &&
				ns_from_origin > trimmer_it->end.ns_from_origin) {
			/*
			 * This only happens when the message's time is
			 * known and is greater than the trimming
			 * range's end time. Unknown and -inf times are
			 * always less than
			 * `trimmer_it->end.ns_from_origin`.
			 */
			status = end_iterator_streams(trimmer_it);
			*reached_end = true;
			break;
		}

		if (!sstate->inited) {
			status = ensure_stream_state_is_inited(trimmer_it,
				sstate, msg);
			msg = NULL;
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}
		} else {
			push_message(trimmer_it, msg);
			msg = NULL;
		}

		break;
	case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
		if (trimmer_it->end.is_infinite) {
			push_message(trimmer_it, msg);
			msg = NULL;
			break;
		}

		if (ns_from_origin == INT64_MIN) {
			/* Unknown: push as is if stream state is inited */
			if (sstate->inited) {
				push_message(trimmer_it, msg);
				msg = NULL;
				sstate->last_msg_is_stream_activity_end = true;
			}
		} else if (ns_from_origin == INT64_MAX) {
			/* Infinite: use trimming range's end time */
			sstate->stream_act_end_ns_from_origin =
				trimmer_it->end.ns_from_origin;
		} else {
			/* Known: check if outside of trimming range */
			if (ns_from_origin > trimmer_it->end.ns_from_origin) {
				sstate->stream_act_end_ns_from_origin =
					trimmer_it->end.ns_from_origin;
				status = end_iterator_streams(trimmer_it);
				*reached_end = true;
				break;
			}

			if (!sstate->inited) {
				/*
				 * First message for this stream is a
				 * stream activity end: we can't deduce
				 * anything about the stream activity
				 * beginning's time, and using this
				 * message's time would make a useless
				 * pair of stream activity beginning/end
				 * with the same time. Just skip this
				 * message and wait for something
				 * useful.
				 */
				break;
			}

			push_message(trimmer_it, msg);
			msg = NULL;
			sstate->last_msg_is_stream_activity_end = true;
			sstate->stream_act_end_ns_from_origin = ns_from_origin;
		}

		break;
	case BT_MESSAGE_TYPE_STREAM_BEGINNING:
		/*
		 * We don't know what follows at this point, so just
		 * keep this message until we know what to do with it
		 * (it will be used in ensure_stream_state_is_inited()).
		 */
		BT_ASSERT(!sstate->inited);
		BT_MESSAGE_MOVE_REF(sstate->stream_beginning_msg, msg);
		break;
	case BT_MESSAGE_TYPE_STREAM_END:
		if (sstate->inited) {
			/*
			 * This is the end of an inited stream: end this
			 * stream if its stream activity end message
			 * time is not the trimming range's end time
			 * (which means the final stream activity end
			 * message had an infinite time). end_stream()
			 * will generate its own stream end message.
			 */
			if (trimmer_it->end.is_infinite) {
				push_message(trimmer_it, msg);
				msg = NULL;
				g_hash_table_remove(trimmer_it->stream_states,
					sstate->stream);
			} else if (sstate->stream_act_end_ns_from_origin <
					trimmer_it->end.ns_from_origin) {
				status = end_stream(trimmer_it, sstate);
				if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
					goto end;
				}

				/* We won't need this stream state again */
				g_hash_table_remove(trimmer_it->stream_states,
					sstate->stream);
			}
		} else {
			/* We dont't need this stream state anymore */
			g_hash_table_remove(trimmer_it->stream_states, sstate->stream);
		}

		break;
	default:
		break;
	}

end:
	/* We release the message's reference whatever the outcome */
	bt_message_put_ref(msg);
	return BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
}

/*
 * Handles an input message. This _could_ make the iterator's output
 * message queue grow; this could also consume the message without
 * pushing anything to this queue, only modifying the stream state.
 *
 * This function consumes the `msg` reference, _whatever the outcome_.
 *
 * This function sets `reached_end` if handling this message made the
 * iterator reach the end of the trimming range. Note that the output
 * message queue could contain messages even if this function sets
 * `reached_end`.
 */
static inline
bt_self_message_iterator_status handle_message(
		struct trimmer_iterator *trimmer_it, const bt_message *msg,
		bool *reached_end)
{
	bt_self_message_iterator_status status;
	const bt_stream *stream = NULL;
	int64_t ns_from_origin = INT64_MIN;
	bool skip;
	int ret;
	struct trimmer_iterator_stream_state *sstate = NULL;

	/* Find message's associated stream */
	switch (bt_message_get_type(msg)) {
	case BT_MESSAGE_TYPE_EVENT:
		stream = bt_event_borrow_stream_const(
			bt_message_event_borrow_event_const(msg));
		break;
	case BT_MESSAGE_TYPE_PACKET_BEGINNING:
		stream = bt_packet_borrow_stream_const(
			bt_message_packet_beginning_borrow_packet_const(msg));
		break;
	case BT_MESSAGE_TYPE_PACKET_END:
		stream = bt_packet_borrow_stream_const(
			bt_message_packet_end_borrow_packet_const(msg));
		break;
	case BT_MESSAGE_TYPE_DISCARDED_EVENTS:
		stream = bt_message_discarded_events_borrow_stream_const(msg);
		break;
	case BT_MESSAGE_TYPE_DISCARDED_PACKETS:
		stream = bt_message_discarded_packets_borrow_stream_const(msg);
		break;
	case BT_MESSAGE_TYPE_STREAM_ACTIVITY_BEGINNING:
		stream = bt_message_stream_activity_beginning_borrow_stream_const(msg);
		break;
	case BT_MESSAGE_TYPE_STREAM_ACTIVITY_END:
		stream = bt_message_stream_activity_end_borrow_stream_const(msg);
		break;
	case BT_MESSAGE_TYPE_STREAM_BEGINNING:
		stream = bt_message_stream_beginning_borrow_stream_const(msg);
		break;
	case BT_MESSAGE_TYPE_STREAM_END:
		stream = bt_message_stream_end_borrow_stream_const(msg);
		break;
	default:
		break;
	}

	if (G_LIKELY(stream)) {
		/* Find stream state */
		sstate = g_hash_table_lookup(trimmer_it->stream_states,
			stream);
		if (G_UNLIKELY(!sstate)) {
			/* No stream state yet: create one now */
			const bt_stream_class *sc;

			/*
			 * Validate right now that the stream's class
			 * has a registered default clock class so that
			 * an existing stream state guarantees existing
			 * default clock snapshots for its associated
			 * messages.
			 *
			 * Also check that clock snapshots are always
			 * known.
			 */
			sc = bt_stream_borrow_class_const(stream);
			if (!bt_stream_class_borrow_default_clock_class_const(sc)) {
				BT_LOGE("Unsupported stream: stream class does "
					"not have a default clock class: "
					"stream-addr=%p, "
					"stream-id=%" PRIu64 ", "
					"stream-name=\"%s\"",
					stream, bt_stream_get_id(stream),
					bt_stream_get_name(stream));
				status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
				goto end;
			}

			/*
			 * Temporary: make sure packet beginning, packet
			 * end, discarded events, and discarded packets
			 * messages have default clock snapshots until
			 * the support for not having them is
			 * implemented.
			 */
			if (!bt_stream_class_packets_have_beginning_default_clock_snapshot(
					sc)) {
				BT_LOGE("Unsupported stream: packets have "
					"no beginning clock snapshot: "
					"stream-addr=%p, "
					"stream-id=%" PRIu64 ", "
					"stream-name=\"%s\"",
					stream, bt_stream_get_id(stream),
					bt_stream_get_name(stream));
				status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
				goto end;
			}

			if (!bt_stream_class_packets_have_end_default_clock_snapshot(
					sc)) {
				BT_LOGE("Unsupported stream: packets have "
					"no end clock snapshot: "
					"stream-addr=%p, "
					"stream-id=%" PRIu64 ", "
					"stream-name=\"%s\"",
					stream, bt_stream_get_id(stream),
					bt_stream_get_name(stream));
				status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
				goto end;
			}

			if (bt_stream_class_supports_discarded_events(sc) &&
					!bt_stream_class_discarded_events_have_default_clock_snapshots(sc)) {
				BT_LOGE("Unsupported stream: discarded events "
					"have no clock snapshots: "
					"stream-addr=%p, "
					"stream-id=%" PRIu64 ", "
					"stream-name=\"%s\"",
					stream, bt_stream_get_id(stream),
					bt_stream_get_name(stream));
				status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
				goto end;
			}

			if (bt_stream_class_supports_discarded_packets(sc) &&
					!bt_stream_class_discarded_packets_have_default_clock_snapshots(sc)) {
				BT_LOGE("Unsupported stream: discarded packets "
					"have no clock snapshots: "
					"stream-addr=%p, "
					"stream-id=%" PRIu64 ", "
					"stream-name=\"%s\"",
					stream, bt_stream_get_id(stream),
					bt_stream_get_name(stream));
				status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
				goto end;
			}

			sstate = g_new0(struct trimmer_iterator_stream_state,
				1);
			if (!sstate) {
				status = BT_SELF_MESSAGE_ITERATOR_STATUS_NOMEM;
				goto end;
			}

			sstate->stream = stream;
			sstate->stream_act_end_ns_from_origin = INT64_MIN;
			g_hash_table_insert(trimmer_it->stream_states,
				(void *) stream, sstate);
		}
	}

	/* Retrieve the message's time */
	ret = get_msg_ns_from_origin(msg, &ns_from_origin, &skip);
	if (G_UNLIKELY(ret)) {
		status = BT_SELF_MESSAGE_ITERATOR_STATUS_ERROR;
		goto end;
	}

	if (G_LIKELY(sstate)) {
		/* Message associated to a stream */
		status = handle_message_with_stream_state(trimmer_it, msg,
			sstate, ns_from_origin, reached_end);

		/*
		 * handle_message_with_stream_state() unconditionally
		 * consumes `msg`.
		 */
		msg = NULL;
	} else {
		/*
		 * Message not associated to a stream (message iterator
		 * inactivity).
		 */
		if (G_UNLIKELY(ns_from_origin > trimmer_it->end.ns_from_origin)) {
			BT_MESSAGE_PUT_REF_AND_RESET(msg);
			status = end_iterator_streams(trimmer_it);
			*reached_end = true;
		} else {
			push_message(trimmer_it, msg);
			status = BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
			msg = NULL;
		}
	}

end:
	/* We release the message's reference whatever the outcome */
	bt_message_put_ref(msg);
	return status;
}

static inline
void fill_message_array_from_output_messages(
		struct trimmer_iterator *trimmer_it,
		bt_message_array_const msgs, uint64_t capacity, uint64_t *count)
{
	*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(trimmer_it->output_messages)) {
		msgs[*count] = pop_message(trimmer_it);
		capacity--;
		(*count)++;
	}

	BT_ASSERT(*count > 0);
}

static inline
bt_self_message_iterator_status state_ending(
		struct trimmer_iterator *trimmer_it,
		bt_message_array_const msgs, uint64_t capacity,
		uint64_t *count)
{
	bt_self_message_iterator_status status =
		BT_SELF_MESSAGE_ITERATOR_STATUS_OK;

	if (g_queue_is_empty(trimmer_it->output_messages)) {
		trimmer_it->state = TRIMMER_ITERATOR_STATE_ENDED;
		status = BT_SELF_MESSAGE_ITERATOR_STATUS_END;
		goto end;
	}

	fill_message_array_from_output_messages(trimmer_it, msgs,
		capacity, count);

end:
	return status;
}

static inline
bt_self_message_iterator_status state_trim(struct trimmer_iterator *trimmer_it,
		bt_message_array_const msgs, uint64_t capacity,
		uint64_t *count)
{
	bt_self_message_iterator_status status =
		BT_SELF_MESSAGE_ITERATOR_STATUS_OK;
	bt_message_array_const my_msgs;
	uint64_t my_count;
	uint64_t i;
	bool reached_end = false;

	while (g_queue_is_empty(trimmer_it->output_messages)) {
		status = (int) bt_self_component_port_input_message_iterator_next(
			trimmer_it->upstream_iter, &my_msgs, &my_count);
		if (G_UNLIKELY(status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK)) {
			if (status == BT_SELF_MESSAGE_ITERATOR_STATUS_END) {
				status = end_iterator_streams(trimmer_it);
				if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
					goto end;
				}

				trimmer_it->state =
					TRIMMER_ITERATOR_STATE_ENDING;
				status = state_ending(trimmer_it, msgs,
					capacity, count);
			}

			goto end;
		}

		BT_ASSERT(my_count > 0);

		for (i = 0; i < my_count; i++) {
			status = handle_message(trimmer_it, my_msgs[i],
				&reached_end);

			/*
			 * handle_message() unconditionally consumes the
			 * message reference.
			 */
			my_msgs[i] = NULL;

			if (G_UNLIKELY(status !=
					BT_SELF_MESSAGE_ITERATOR_STATUS_OK)) {
				put_messages(my_msgs, my_count);
				goto end;
			}

			if (G_UNLIKELY(reached_end)) {
				/*
				 * This message's time was passed the
				 * trimming time range's end time: we
				 * are done. Their might still be
				 * messages in the output message queue,
				 * so move to the "ending" state and
				 * apply it immediately since
				 * state_trim() is called within the
				 * "next" method.
				 */
				put_messages(my_msgs, my_count);
				trimmer_it->state =
					TRIMMER_ITERATOR_STATE_ENDING;
				status = state_ending(trimmer_it, msgs,
					capacity, count);
				goto end;
			}
		}
	}

	/*
	 * There's at least one message in the output message queue:
	 * move the messages to the output message array.
	 */
	BT_ASSERT(!g_queue_is_empty(trimmer_it->output_messages));
	fill_message_array_from_output_messages(trimmer_it, msgs,
		capacity, count);

end:
	return status;
}

BT_HIDDEN
bt_self_message_iterator_status trimmer_msg_iter_next(
		bt_self_message_iterator *self_msg_iter,
		bt_message_array_const msgs, uint64_t capacity,
		uint64_t *count)
{
	struct trimmer_iterator *trimmer_it =
		bt_self_message_iterator_get_data(self_msg_iter);
	bt_self_message_iterator_status status =
		BT_SELF_MESSAGE_ITERATOR_STATUS_OK;

	BT_ASSERT(trimmer_it);

	if (G_LIKELY(trimmer_it->state == TRIMMER_ITERATOR_STATE_TRIM)) {
		status = state_trim(trimmer_it, msgs, capacity, count);
		if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
			goto end;
		}
	} else {
		switch (trimmer_it->state) {
		case TRIMMER_ITERATOR_STATE_SET_BOUNDS_NS_FROM_ORIGIN:
			status = state_set_trimmer_iterator_bounds(trimmer_it);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}

			status = state_seek_initially(trimmer_it);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}

			status = state_trim(trimmer_it, msgs, capacity, count);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}

			break;
		case TRIMMER_ITERATOR_STATE_SEEK_INITIALLY:
			status = state_seek_initially(trimmer_it);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}

			status = state_trim(trimmer_it, msgs, capacity, count);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}

			break;
		case TRIMMER_ITERATOR_STATE_ENDING:
			status = state_ending(trimmer_it, msgs, capacity,
				count);
			if (status != BT_SELF_MESSAGE_ITERATOR_STATUS_OK) {
				goto end;
			}

			break;
		case TRIMMER_ITERATOR_STATE_ENDED:
			status = BT_SELF_MESSAGE_ITERATOR_STATUS_END;
			break;
		default:
			abort();
		}
	}

end:
	return status;
}

BT_HIDDEN
void trimmer_msg_iter_finalize(bt_self_message_iterator *self_msg_iter)
{
	struct trimmer_iterator *trimmer_it =
		bt_self_message_iterator_get_data(self_msg_iter);

	BT_ASSERT(trimmer_it);
	destroy_trimmer_iterator(trimmer_it);
}