* BabelTrace - LTTng-live client Component
*/
-#ifndef BABELTRACE_PLUGIN_CTF_LTTNG_LIVE_H
-#define BABELTRACE_PLUGIN_CTF_LTTNG_LIVE_H
-
-#include <stdbool.h>
-#include <stdint.h>
+#ifndef BABELTRACE_PLUGINS_CTF_LTTNG_LIVE_LTTNG_LIVE_HPP
+#define BABELTRACE_PLUGINS_CTF_LTTNG_LIVE_LTTNG_LIVE_HPP
#include <glib.h>
+#include <stdint.h>
#include <babeltrace2/babeltrace.h>
-#include "common/macros.h"
-#include "../common/metadata/decoder.hpp"
-#include "../common/msg-iter/msg-iter.hpp"
+#include "cpp-common/bt2/message.hpp"
+#include "cpp-common/vendor/fmt/format.h" /* IWYU pragma: keep */
+
+#include "../common/src/metadata/metadata-stream-parser-utils.hpp"
+#include "../common/src/msg-iter.hpp"
#include "viewer-connection.hpp"
-struct lttng_live_component;
-struct lttng_live_session;
-struct lttng_live_msg_iter;
+/*
+ * bt_common_lttng_live_url_parts is defined in common code, and is also used
+ * by C code, so it can't be C++-ified yet. Use this separate deleter object
+ * in the mean time.
+ */
+struct bt_common_lttng_live_url_parts_deleter
+{
+ explicit bt_common_lttng_live_url_parts_deleter(bt_common_lttng_live_url_parts& obj) noexcept :
+ _mObj {&obj}
+ {
+ }
+
+ bt_common_lttng_live_url_parts_deleter(const bt_common_lttng_live_url_parts_deleter&) = delete;
+ bt_common_lttng_live_url_parts&
+ operator=(const bt_common_lttng_live_url_parts_deleter&) = delete;
+
+ ~bt_common_lttng_live_url_parts_deleter()
+ {
+ bt_common_destroy_lttng_live_url_parts(_mObj);
+ }
+
+private:
+ bt_common_lttng_live_url_parts *_mObj;
+};
enum lttng_live_stream_state
{
LTTNG_LIVE_STREAM_EOF,
};
+inline const char *format_as(const lttng_live_stream_state state) noexcept
+{
+ switch (state) {
+ case LTTNG_LIVE_STREAM_ACTIVE_NO_DATA:
+ return "ACTIVE_NO_DATA";
+
+ case LTTNG_LIVE_STREAM_QUIESCENT_NO_DATA:
+ return "QUIESCENT_NO_DATA";
+
+ case LTTNG_LIVE_STREAM_QUIESCENT:
+ return "QUIESCENT";
+
+ case LTTNG_LIVE_STREAM_ACTIVE_DATA:
+ return "ACTIVE_DATA";
+
+ case LTTNG_LIVE_STREAM_EOF:
+ return "EOF";
+ }
+
+ bt_common_abort();
+}
+
/* Iterator over a live stream. */
struct lttng_live_stream_iterator
{
- bt_logging_level log_level;
- bt_self_component *self_comp;
+ using UP = std::unique_ptr<lttng_live_stream_iterator>;
+
+ explicit lttng_live_stream_iterator(const bt2c::Logger& parentLogger) :
+ logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/STREAM-ITER"}
+ {
+ }
+
+ ~lttng_live_stream_iterator();
+
+ bt2c::Logger logger;
- /* Owned by this. */
- bt_stream *stream;
+ bt2::Stream::Shared stream;
/* Weak reference. */
- struct lttng_live_trace *trace;
+ struct lttng_live_trace *trace = nullptr;
/*
* Since only a single iterator per viewer connection, we have
* only a single message iterator per stream.
*/
- struct ctf_msg_iter *msg_iter;
+ bt2s::optional<ctf::src::MsgIter> msg_iter;
- uint64_t viewer_stream_id;
+ uint64_t viewer_stream_id = 0;
struct
{
- bool is_set;
- uint64_t value;
+ bool is_set = false;
+ uint64_t value = 0;
} ctf_stream_class_id;
- /* base offset in current index. */
- uint64_t base_offset;
- /* len to read in current index. */
- uint64_t len;
- /* offset in current index. */
- uint64_t offset;
-
/*
* Clock Snapshot value of the last message iterator inactivity message
* sent downstream.
*/
struct
{
- bool is_set;
- uint64_t value;
+ bool is_set = false;
+ uint64_t value = 0;
} last_inactivity_ts;
/*
* Clock Snapshot value of the current message iterator inactivity
* message we might want to send downstream.
*/
- uint64_t current_inactivity_ts;
+ uint64_t current_inactivity_ts = 0;
- enum lttng_live_stream_state state;
+ enum lttng_live_stream_state state = LTTNG_LIVE_STREAM_QUIESCENT;
- /*
- * The current message produced by this live stream iterator. Owned by
- * this.
- */
- const bt_message *current_msg;
+ /* The current message produced by this live stream iterator. */
+ bt2::ConstMessage::Shared current_msg;
/* Timestamp in nanoseconds of the current message (current_msg). */
- int64_t current_msg_ts_ns;
+ int64_t current_msg_ts_ns = 0;
- /* Owned by this. */
- uint8_t *buf;
- size_t buflen;
+ std::string name;
- /* Owned by this. */
- GString *name;
+ bool has_stream_hung_up = false;
+
+ struct CurPktInfo
+ {
+ bt2c::DataLen offsetInRelay;
+ bt2c::DataLen len;
+ };
- bool has_stream_hung_up;
+ bt2s::optional<CurPktInfo> curPktInfo;
};
struct lttng_live_metadata
{
- bt_logging_level log_level;
- bt_self_component *self_comp;
+ using UP = std::unique_ptr<lttng_live_metadata>;
- uint64_t stream_id;
- /* Weak reference. */
- struct ctf_metadata_decoder *decoder;
+ explicit lttng_live_metadata(const bt2::SelfComponent selfComp,
+ const bt2c::Logger& parentLogger) :
+ logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/METADATA"},
+ _mSelfComp {selfComp}
+
+ {
+ }
+
+ const ctf::src::TraceCls *traceCls() const
+ {
+ return _mMetadataStreamParser->traceCls();
+ }
+
+ const bt2s::optional<bt2c::Uuid>& metadataStreamUuid() const noexcept
+ {
+ return _mMetadataStreamParser->metadataStreamUuid();
+ }
+
+ void parseSection(const bt2c::ConstBytes data)
+ {
+ if (!_mMetadataStreamParser) {
+ _mMetadataStreamParser =
+ ctf::src::createMetadataStreamParser(data, _mSelfComp, {}, logger);
+ }
+
+ _mMetadataStreamParser->parseSection(data);
+ }
+
+ bt2::SelfComponent selfComp() const noexcept
+ {
+ return _mSelfComp;
+ }
+
+ bt2c::Logger logger;
+
+ uint64_t stream_id = 0;
+
+private:
+ bt2::SelfComponent _mSelfComp;
+ ctf::src::MetadataStreamParser::UP _mMetadataStreamParser;
};
enum lttng_live_metadata_stream_state
struct lttng_live_trace
{
- bt_logging_level log_level;
- bt_self_component *self_comp;
+ using UP = std::unique_ptr<lttng_live_trace>;
+
+ explicit lttng_live_trace(const bt2c::Logger& parentLogger) :
+ logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/TRACE"}
+ {
+ }
+
+ bt2c::Logger logger;
/* Back reference to session. */
- struct lttng_live_session *session;
+ struct lttng_live_session *session = nullptr;
/* ctf trace ID within the session. */
- uint64_t id;
+ uint64_t id = 0;
- /* Owned by this. */
- bt_trace *trace;
+ bt2::Trace::Shared trace;
- /* Weak reference. */
- bt_trace_class *trace_class;
+ lttng_live_metadata::UP metadata;
- struct lttng_live_metadata *metadata;
+ bt2::OptionalBorrowedObject<bt2::ConstClockClass> clock_class;
- const bt_clock_class *clock_class;
+ std::vector<lttng_live_stream_iterator::UP> stream_iterators;
- /* Array of pointers to struct lttng_live_stream_iterator. */
- /* Owned by this. */
- GPtrArray *stream_iterators;
-
- enum lttng_live_metadata_stream_state metadata_stream_state;
+ enum lttng_live_metadata_stream_state metadata_stream_state =
+ LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
};
struct lttng_live_session
{
- bt_logging_level log_level;
- bt_self_component *self_comp;
+ using UP = std::unique_ptr<lttng_live_session>;
+
+ explicit lttng_live_session(const bt2c::Logger& parentLogger,
+ const bt2::SelfComponent selfCompParam) :
+ logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/SESSION"},
+ selfComp {selfCompParam}
+ {
+ }
+
+ ~lttng_live_session();
+
+ bt2c::Logger logger;
+
+ bt2::SelfComponent selfComp;
/* Weak reference. */
- struct lttng_live_msg_iter *lttng_live_msg_iter;
+ struct lttng_live_msg_iter *lttng_live_msg_iter = nullptr;
- /* Owned by this. */
- GString *hostname;
+ std::string hostname;
- /* Owned by this. */
- GString *session_name;
+ std::string session_name;
- uint64_t id;
+ uint64_t id = 0;
- /* Array of pointers to struct lttng_live_trace. */
- GPtrArray *traces;
+ std::vector<lttng_live_trace::UP> traces;
- bool attached;
- bool new_streams_needed;
- bool lazy_stream_msg_init;
- bool closed;
+ bool attached = false;
+ bool new_streams_needed = false;
+ bool lazy_stream_msg_init = false;
+ bool closed = false;
};
enum session_not_found_action
*/
struct lttng_live_component
{
- bt_logging_level log_level;
+ using UP = std::unique_ptr<lttng_live_component>;
- /* Weak reference. */
- bt_self_component *self_comp;
+ explicit lttng_live_component(bt2c::Logger loggerParam,
+ const bt2::SelfComponent selfCompParam) noexcept :
+ logger {std::move(loggerParam)},
+ selfComp {selfCompParam}
+ {
+ }
+
+ bt2c::Logger logger;
+
+ bt2::SelfComponent selfComp;
struct
{
- GString *url;
- enum session_not_found_action sess_not_found_act;
+ std::string url;
+ enum session_not_found_action sess_not_found_act = SESSION_NOT_FOUND_ACTION_CONTINUE;
} params;
- size_t max_query_size;
+ size_t max_query_size = 0;
/*
* Keeps track of whether the downstream component already has a
* message iterator on this component.
*/
- bool has_msg_iter;
+ bool has_msg_iter = false;
};
struct lttng_live_msg_iter
{
- bt_logging_level log_level;
- bt_self_component *self_comp;
+ using UP = std::unique_ptr<lttng_live_msg_iter>;
- /* Weak reference. */
- struct lttng_live_component *lttng_live_comp;
+ explicit lttng_live_msg_iter(const bt2c::Logger& parentLogger,
+ const bt2::SelfComponent selfCompParam,
+ const bt2::SelfMessageIterator selfMsgIter) :
+ logger {parentLogger, "PLUGIN/SRC.CTF.LTTNG-LIVE/MSG-ITER"},
+ selfComp {selfCompParam}, selfMsgIter {selfMsgIter}
+ {
+ }
+
+ ~lttng_live_msg_iter();
+
+ bt2c::Logger logger;
+
+ bt2::SelfComponent selfComp;
/* Weak reference. */
- bt_self_message_iterator *self_msg_iter;
+ struct lttng_live_component *lttng_live_comp = nullptr;
+
+ bt2::SelfMessageIterator selfMsgIter;
- /* Owned by this. */
- struct live_viewer_connection *viewer_connection;
+ live_viewer_connection::UP viewer_connection;
- /* Array of pointers to struct lttng_live_session. */
- GPtrArray *sessions;
+ std::vector<lttng_live_session::UP> sessions;
/* Number of live stream iterator this message iterator has.*/
- uint64_t active_stream_iter;
+ uint64_t active_stream_iter = 0;
/* Timestamp in nanosecond of the last message sent downstream. */
- int64_t last_msg_ts_ns;
+ int64_t last_msg_ts_ns = 0;
/* True if the iterator was interrupted. */
- bool was_interrupted;
+ bool was_interrupted = false;
};
enum lttng_live_iterator_status
LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED = -4,
};
+inline const char *format_as(const lttng_live_iterator_status status) noexcept
+{
+ switch (status) {
+ case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
+ return "LTTNG_LIVE_ITERATOR_STATUS_CONTINUE";
+
+ case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
+ return "LTTNG_LIVE_ITERATOR_STATUS_AGAIN";
+
+ case LTTNG_LIVE_ITERATOR_STATUS_END:
+ return "LTTNG_LIVE_ITERATOR_STATUS_END";
+
+ case LTTNG_LIVE_ITERATOR_STATUS_OK:
+ return "LTTNG_LIVE_ITERATOR_STATUS_OK";
+
+ case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
+ return "LTTNG_LIVE_ITERATOR_STATUS_INVAL";
+
+ case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
+ return "LTTNG_LIVE_ITERATOR_STATUS_ERROR";
+
+ case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
+ return "LTTNG_LIVE_ITERATOR_STATUS_NOMEM";
+
+ case LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED:
+ return "LTTNG_LIVE_ITERATOR_STATUS_UNSUPPORTED";
+ }
+
+ bt_common_abort();
+}
+
bt_component_class_initialize_method_status
lttng_live_component_init(bt_self_component_source *self_comp,
bt_self_component_source_configuration *config, const bt_value *params,
void lttng_live_component_finalize(bt_self_component_source *component);
bt_message_iterator_class_next_method_status
-lttng_live_msg_iter_next(bt_self_message_iterator *iterator, bt_message_array_const msgs,
+lttng_live_msg_iter_next(bt_self_message_iterator *self_msg_it, bt_message_array_const msgs,
uint64_t capacity, uint64_t *count);
bt_message_iterator_class_initialize_method_status
void lttng_live_msg_iter_finalize(bt_self_message_iterator *it);
-enum lttng_live_viewer_status lttng_live_session_attach(struct lttng_live_session *session,
- bt_self_message_iterator *self_msg_iter);
+enum lttng_live_viewer_status lttng_live_session_attach(struct lttng_live_session *session);
enum lttng_live_viewer_status lttng_live_session_detach(struct lttng_live_session *session);
enum lttng_live_iterator_status
-lttng_live_session_get_new_streams(struct lttng_live_session *session,
- bt_self_message_iterator *self_msg_iter);
+lttng_live_session_get_new_streams(struct lttng_live_session *session);
struct lttng_live_trace *
lttng_live_session_borrow_or_create_trace_by_id(struct lttng_live_session *session,
* written to the file.
*/
enum lttng_live_get_one_metadata_status
-lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace, FILE *fp, size_t *reply_len);
+lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace, std::vector<uint8_t>& buf);
enum lttng_live_iterator_status
lttng_live_get_next_index(struct lttng_live_msg_iter *lttng_live_msg_iter,
struct lttng_live_stream_iterator *stream, struct packet_index *index);
-enum ctf_msg_iter_medium_status
-lttng_live_get_stream_bytes(struct lttng_live_msg_iter *lttng_live_msg_iter,
- struct lttng_live_stream_iterator *stream, uint8_t *buf,
- uint64_t offset, uint64_t req_len, uint64_t *recv_len);
-
bool lttng_live_graph_is_canceled(struct lttng_live_msg_iter *msg_iter);
-BT_HIDDEN
void lttng_live_stream_iterator_set_state(struct lttng_live_stream_iterator *stream_iter,
enum lttng_live_stream_state new_state);
-#endif /* BABELTRACE_PLUGIN_CTF_LTTNG_LIVE_H */
+void lttng_live_stream_iterator_set_stream_class(lttng_live_stream_iterator *streamIter,
+ uint64_t ctfStreamClsId);
+
+#endif /* BABELTRACE_PLUGINS_CTF_LTTNG_LIVE_LTTNG_LIVE_HPP */