lib: introduce bt_message_iterator_class
[babeltrace.git] / src / plugins / ctf / lttng-live / lttng-live.c
index efaec623fab9bf4eb51feec70cbf3d1165bf0a01..6d77035b3ecde36fedcfac32db8fb097bef84980 100644 (file)
@@ -133,7 +133,7 @@ end:
 }
 
 static
-struct lttng_live_trace *lttng_live_find_trace(struct lttng_live_session *session,
+struct lttng_live_trace *lttng_live_session_borrow_trace_by_id(struct lttng_live_session *session,
                uint64_t trace_id)
 {
        uint64_t trace_idx;
@@ -196,7 +196,7 @@ struct lttng_live_trace *lttng_live_create_trace(struct lttng_live_session *sess
        trace->stream_iterators = g_ptr_array_new_with_free_func(
                (GDestroyNotify) lttng_live_stream_iterator_destroy);
        BT_ASSERT(trace->stream_iterators);
-       trace->new_metadata_needed = true;
+       trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
        g_ptr_array_add(session->traces, trace);
 
        goto end;
@@ -208,12 +208,12 @@ end:
 }
 
 BT_HIDDEN
-struct lttng_live_trace *lttng_live_borrow_trace(
+struct lttng_live_trace *lttng_live_session_borrow_or_create_trace_by_id(
                struct lttng_live_session *session, uint64_t trace_id)
 {
        struct lttng_live_trace *trace;
 
-       trace = lttng_live_find_trace(session, trace_id);
+       trace = lttng_live_session_borrow_trace_by_id(session, trace_id);
        if (trace) {
                goto end;
        }
@@ -285,7 +285,7 @@ void lttng_live_destroy_session(struct lttng_live_session *session)
                "session-id=%"PRIu64", session-name=\"%s\"",
                session->id, session->session_name->str);
        if (session->id != -1ULL) {
-               if (lttng_live_detach_session(session)) {
+               if (lttng_live_session_detach(session)) {
                        if (!lttng_live_graph_is_canceled(
                                        session->lttng_live_msg_iter)) {
                                /* Old relayd cannot detach sessions. */
@@ -395,7 +395,8 @@ enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_no_data_stre
        enum lttng_live_stream_state orig_state = lttng_live_stream->state;
        struct packet_index index;
 
-       if (lttng_live_stream->trace->new_metadata_needed) {
+       if (lttng_live_stream->trace->metadata_stream_state ==
+                       LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
                ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
                goto end;
        }
@@ -459,7 +460,7 @@ enum lttng_live_iterator_status lttng_live_get_session(
 
        if (!session->attached) {
                enum lttng_live_viewer_status attach_status =
-                       lttng_live_attach_session(session,
+                       lttng_live_session_attach(session,
                                lttng_live_msg_iter->self_msg_iter);
                if (attach_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
                        if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
@@ -480,7 +481,7 @@ enum lttng_live_iterator_status lttng_live_get_session(
                }
        }
 
-       status = lttng_live_get_new_streams(session,
+       status = lttng_live_session_get_new_streams(session,
                lttng_live_msg_iter->self_msg_iter);
        if (status != LTTNG_LIVE_ITERATOR_STATUS_OK &&
                        status != LTTNG_LIVE_ITERATOR_STATUS_END) {
@@ -493,22 +494,13 @@ enum lttng_live_iterator_status lttng_live_get_session(
 
                status = lttng_live_metadata_update(trace);
                switch (status) {
+               case LTTNG_LIVE_ITERATOR_STATUS_END:
                case LTTNG_LIVE_ITERATOR_STATUS_OK:
                        trace_idx++;
                        break;
-               case LTTNG_LIVE_ITERATOR_STATUS_END:
-                       /*
-                        * The trace has ended. Remove it of the array an
-                        * continue the iteration.
-                        * We can remove the trace safely when using the
-                        * g_ptr_array_remove_index_fast because it replaces
-                        * the element at trace_idx with the array's last
-                        * element. trace_idx is not incremented because of
-                        * that.
-                        */
-                       (void) g_ptr_array_remove_index_fast(session->traces,
-                               trace_idx);
-                       break;
+               case LTTNG_LIVE_ITERATOR_STATUS_CONTINUE:
+               case LTTNG_LIVE_ITERATOR_STATUS_AGAIN:
+                       goto end;
                default:
                        BT_COMP_LOGE_APPEND_CAUSE(self_comp,
                                "Error updating trace metadata: "
@@ -525,19 +517,6 @@ end:
        return status;
 }
 
-BT_HIDDEN
-void lttng_live_need_new_streams(struct lttng_live_msg_iter *lttng_live_msg_iter)
-{
-       uint64_t session_idx;
-
-       for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len;
-                       session_idx++) {
-               struct lttng_live_session *session =
-                       g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
-               session->new_streams_needed = true;
-       }
-}
-
 static
 void lttng_live_force_new_streams_and_metadata(struct lttng_live_msg_iter *lttng_live_msg_iter)
 {
@@ -552,7 +531,11 @@ void lttng_live_force_new_streams_and_metadata(struct lttng_live_msg_iter *lttng
                                trace_idx++) {
                        struct lttng_live_trace *trace =
                                g_ptr_array_index(session->traces, trace_idx);
-                       trace->new_metadata_needed = true;
+
+                       BT_ASSERT(trace->metadata_stream_state !=
+                               LTTNG_LIVE_METADATA_STREAM_STATE_CLOSED);
+
+                       trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
                }
        }
 }
@@ -640,7 +623,7 @@ static
 enum lttng_live_iterator_status emit_inactivity_message(
                struct lttng_live_msg_iter *lttng_live_msg_iter,
                struct lttng_live_stream_iterator *stream_iter,
-               bt_message **message, uint64_t timestamp)
+               const bt_message **message, uint64_t timestamp)
 {
        enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
        bt_logging_level log_level = lttng_live_msg_iter->log_level;
@@ -672,7 +655,7 @@ static
 enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_quiescent_stream(
                struct lttng_live_msg_iter *lttng_live_msg_iter,
                struct lttng_live_stream_iterator *lttng_live_stream,
-               bt_message **message)
+               const bt_message **message)
 {
        enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
 
@@ -798,7 +781,7 @@ static
 enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_stream(
                struct lttng_live_msg_iter *lttng_live_msg_iter,
                struct lttng_live_stream_iterator *lttng_live_stream,
-               bt_message **message)
+               const bt_message **message)
 {
        enum lttng_live_iterator_status ret = LTTNG_LIVE_ITERATOR_STATUS_OK;
        bt_logging_level log_level = lttng_live_msg_iter->log_level;
@@ -819,7 +802,7 @@ enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_
                                trace_idx++) {
                        struct lttng_live_trace *trace =
                                g_ptr_array_index(session->traces, trace_idx);
-                       if (trace->new_metadata_needed) {
+                       if (trace->metadata_stream_state == LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED) {
                                ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
                                goto end;
                        }
@@ -852,8 +835,6 @@ enum lttng_live_iterator_status lttng_live_iterator_next_handle_one_active_data_
                 */
                ret = LTTNG_LIVE_ITERATOR_STATUS_CONTINUE;
                break;
-       case CTF_MSG_ITER_STATUS_INVAL:
-               /* No argument provided by the user, so don't return INVAL. */
        case CTF_MSG_ITER_STATUS_ERROR:
        default:
                ret = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
@@ -873,7 +854,7 @@ static
 enum lttng_live_iterator_status lttng_live_iterator_close_stream(
                struct lttng_live_msg_iter *lttng_live_msg_iter,
                struct lttng_live_stream_iterator *stream_iter,
-               bt_message **curr_msg)
+               const bt_message **curr_msg)
 {
        enum lttng_live_iterator_status live_status =
                LTTNG_LIVE_ITERATOR_STATUS_OK;
@@ -952,7 +933,7 @@ static
 enum lttng_live_iterator_status lttng_live_iterator_next_msg_on_stream(
                struct lttng_live_msg_iter *lttng_live_msg_iter,
                struct lttng_live_stream_iterator *stream_iter,
-               bt_message **curr_msg)
+               const bt_message **curr_msg)
 {
        bt_logging_level log_level = lttng_live_msg_iter->log_level;
        bt_self_component *self_comp = lttng_live_msg_iter->self_comp;
@@ -1052,7 +1033,7 @@ enum lttng_live_iterator_status next_stream_iterator_for_trace(
                 * iterator get it.
                 */
                while (!stream_iter->current_msg) {
-                       bt_message *msg = NULL;
+                       const bt_message *msg = NULL;
                        int64_t curr_msg_ts_ns = INT64_MAX;
                        stream_iter_status = lttng_live_iterator_next_msg_on_stream(
                                lttng_live_msg_iter, stream_iter, &msg);
@@ -1298,12 +1279,12 @@ void put_messages(bt_message_array_const msgs, uint64_t count)
 }
 
 BT_HIDDEN
-bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next(
+bt_message_iterator_class_next_method_status lttng_live_msg_iter_next(
                bt_self_message_iterator *self_msg_it,
                bt_message_array_const msgs, uint64_t capacity,
                uint64_t *count)
 {
-       bt_component_class_message_iterator_next_method_status status;
+       bt_message_iterator_class_next_method_status status;
        enum lttng_live_viewer_status viewer_status;
        struct lttng_live_msg_iter *lttng_live_msg_iter =
                bt_self_message_iterator_get_data(self_msg_it);
@@ -1327,7 +1308,7 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next(
                 * is to prevent other graph users from using this live
                 * iterator in an messed up internal state.
                 */
-               status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
+               status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
                BT_COMP_LOGE_APPEND_CAUSE(self_comp,
                        "Message iterator was interrupted during a previous call to the `next()` and currently does not support continuing after such event.");
                goto end;
@@ -1347,7 +1328,7 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next(
        if (lttng_live_msg_iter->sessions->len == 0) {
                if (lttng_live->params.sess_not_found_act !=
                                SESSION_NOT_FOUND_ACTION_CONTINUE) {
-                       status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
+                       status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
                        goto end;
                } else {
                        /*
@@ -1358,11 +1339,11 @@ bt_component_class_message_iterator_next_method_status lttng_live_msg_iter_next(
                        viewer_status = lttng_live_create_viewer_session(lttng_live_msg_iter);
                        if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
                                if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
-                                       status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
+                                       status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
                                        BT_COMP_LOGE_APPEND_CAUSE(self_comp,
                                                "Error creating LTTng live viewer session");
                                } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
-                                       status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN;
+                                       status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
                                } else {
                                        bt_common_abort();
                                }
@@ -1513,20 +1494,20 @@ return_status:
                 * doesn't support restarting after an interruption.
                 */
                if (*count > 0) {
-                       status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_OK;
+                       status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_OK;
                } else {
-                       status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_AGAIN;
+                       status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_AGAIN;
                }
                break;
        case LTTNG_LIVE_ITERATOR_STATUS_END:
-               status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_END;
+               status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_END;
                break;
        case LTTNG_LIVE_ITERATOR_STATUS_NOMEM:
                BT_COMP_LOGE_APPEND_CAUSE(self_comp,
                        "Memory error preparing the next batch of messages: "
                        "live-iter-status=%s",
                        lttng_live_iterator_status_string(stream_iter_status));
-               status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_MEMORY_ERROR;
+               status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_MEMORY_ERROR;
                break;
        case LTTNG_LIVE_ITERATOR_STATUS_ERROR:
        case LTTNG_LIVE_ITERATOR_STATUS_INVAL:
@@ -1536,7 +1517,7 @@ return_status:
                        "live-iter-status=%s",
                        lttng_live_iterator_status_string(stream_iter_status));
 
-               status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_NEXT_METHOD_STATUS_ERROR;
+               status = BT_MESSAGE_ITERATOR_CLASS_NEXT_METHOD_STATUS_ERROR;
                /* Put all existing messages on error. */
                put_messages(msgs, *count);
                break;
@@ -1583,15 +1564,13 @@ end:
 }
 
 BT_HIDDEN
-bt_component_class_message_iterator_initialize_method_status lttng_live_msg_iter_init(
+bt_message_iterator_class_initialize_method_status lttng_live_msg_iter_init(
                bt_self_message_iterator *self_msg_it,
                bt_self_message_iterator_configuration *config,
-               bt_self_component_source *self_comp_src,
+               bt_self_component *self_comp,
                bt_self_component_port_output *self_port)
 {
-       bt_component_class_message_iterator_initialize_method_status status;
-       bt_self_component *self_comp =
-               bt_self_component_source_as_self_component(self_comp_src);
+       bt_message_iterator_class_initialize_method_status status;
        struct lttng_live_component *lttng_live;
        struct lttng_live_msg_iter *lttng_live_msg_iter;
        enum lttng_live_viewer_status viewer_status;
@@ -1611,7 +1590,7 @@ bt_component_class_message_iterator_initialize_method_status lttng_live_msg_iter
        lttng_live_msg_iter = lttng_live_msg_iter_create(lttng_live,
                self_msg_it);
        if (!lttng_live_msg_iter) {
-               status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
+               status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_MEMORY_ERROR;
                BT_COMP_LOGE_APPEND_CAUSE(self_comp,
                        "Failed to create lttng_live_msg_iter");
                goto error;
@@ -1681,11 +1660,11 @@ bt_component_class_message_iterator_initialize_method_status lttng_live_msg_iter
        }
 
        bt_self_message_iterator_set_data(self_msg_it, lttng_live_msg_iter);
-       status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_OK;
+       status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_OK;
        goto end;
 
 error:
-       status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INITIALIZE_METHOD_STATUS_ERROR;
+       status = BT_MESSAGE_ITERATOR_CLASS_INITIALIZE_METHOD_STATUS_ERROR;
        lttng_live_msg_iter_destroy(lttng_live_msg_iter);
 end:
        return status;
This page took 0.028736 seconds and 4 git commands to generate.