src.ctf.lttng-live: use the new metadata stream parser and message iterator

[deliverable/babeltrace.git] / src / plugins / ctf / lttng-live / viewer-connection.cpp

diff --git a/src/plugins/ctf/lttng-live/viewer-connection.cpp b/src/plugins/ctf/lttng-live/viewer-connection.cpp
index e504304ca604c1f4bc1d9fcff58dc6cf38d7a2aa..053910cf6dec92bdf09980c403b3f4c3e3779855 100644 (file)
--- a/src/plugins/ctf/lttng-live/viewer-connection.cpp
+++ b/src/plugins/ctf/lttng-live/viewer-connection.cpp
@@ -22,6 +22,7 @@
 #include "compat/endian.h"
 #include "compat/compiler.h"
 #include "common/common.h"
+#include "cpp-common/exc.hpp"
 #include <babeltrace2/babeltrace.h>
 #include "cpp-common/make-unique.hpp"
 
@@ -158,16 +159,16 @@ viewer_status_to_live_iterator_status(enum lttng_live_viewer_status viewer_statu
     bt_common_abort();
 }
 
-static inline enum ctf_msg_iter_medium_status
-viewer_status_to_ctf_msg_iter_medium_status(enum lttng_live_viewer_status viewer_status)
+static inline enum lttng_live_get_stream_bytes_status
+viewer_status_to_lttng_live_get_stream_bytes_status(enum lttng_live_viewer_status viewer_status)
 {
     switch (viewer_status) {
     case LTTNG_LIVE_VIEWER_STATUS_OK:
-        return CTF_MSG_ITER_MEDIUM_STATUS_OK;
+        return LTTNG_LIVE_GET_STREAM_BYTES_STATUS_OK;
     case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED:
-        return CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
+        return LTTNG_LIVE_GET_STREAM_BYTES_STATUS_AGAIN;
     case LTTNG_LIVE_VIEWER_STATUS_ERROR:
-        return CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
+        return LTTNG_LIVE_GET_STREAM_BYTES_STATUS_ERROR;
     }
 
     bt_common_abort();
@@ -1031,14 +1032,13 @@ enum lttng_live_viewer_status lttng_live_session_detach(struct lttng_live_sessio
 
 BT_HIDDEN
 enum lttng_live_get_one_metadata_status
-lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace, std::vector<char>& buf)
+lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace, std::vector<uint8_t>& buf)
 {
     uint64_t len = 0;
     enum lttng_live_viewer_status viewer_status;
     struct lttng_viewer_cmd cmd;
     struct lttng_viewer_get_metadata rq;
     struct lttng_viewer_metadata_packet rp;
-    std::vector<char> data;
     struct lttng_live_session *session = trace->session;
     struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
     struct lttng_live_metadata *metadata = trace->metadata.get();
@@ -1113,18 +1113,15 @@ lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace, std::vector<c
         return LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
     }
 
-    data.resize(len);
+    std::vector<char> localBuf(len);
 
-    viewer_status = lttng_live_recv(viewer_connection, data.data(), len);
+    viewer_status = lttng_live_recv(viewer_connection, localBuf.data(), len);
     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
         viewer_handle_recv_status(logCfg, viewer_status, "get metadata packet");
         return (lttng_live_get_one_metadata_status) viewer_status;
     }
 
-    /*
-     * Write the metadata to the file handle.
-     */
-    buf.insert(buf.end(), data.begin(), data.end());
+    buf.insert(buf.end(), localBuf.begin(), localBuf.end());
 
     return LTTNG_LIVE_GET_ONE_METADATA_STATUS_OK;
 }
@@ -1280,7 +1277,7 @@ lttng_live_get_next_index(struct lttng_live_msg_iter *lttng_live_msg_iter,
 }
 
 BT_HIDDEN
-enum ctf_msg_iter_medium_status
+lttng_live_get_stream_bytes_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)
@@ -1320,13 +1317,13 @@ lttng_live_get_stream_bytes(struct lttng_live_msg_iter *lttng_live_msg_iter,
     viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
         viewer_handle_send_status(logCfg, viewer_status, "get data packet command");
-        return viewer_status_to_ctf_msg_iter_medium_status(viewer_status);
+        return viewer_status_to_lttng_live_get_stream_bytes_status(viewer_status);
     }
 
     viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
         viewer_handle_recv_status(logCfg, viewer_status, "get data packet reply");
-        return viewer_status_to_ctf_msg_iter_medium_status(viewer_status);
+        return viewer_status_to_lttng_live_get_stream_bytes_status(viewer_status);
     }
 
     flags = be32toh(rp.flags);
@@ -1343,7 +1340,7 @@ lttng_live_get_stream_bytes(struct lttng_live_msg_iter *lttng_live_msg_iter,
         break;
     case LTTNG_VIEWER_GET_PACKET_RETRY:
         /* Unimplemented by relay daemon */
-        return CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
+        return LTTNG_LIVE_GET_STREAM_BYTES_STATUS_AGAIN;
     case LTTNG_VIEWER_GET_PACKET_ERR:
         if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
             BT_CLOGD("Marking trace as needing new metadata: "
@@ -1360,29 +1357,29 @@ lttng_live_get_stream_bytes(struct lttng_live_msg_iter *lttng_live_msg_iter,
         if (flags & (LTTNG_VIEWER_FLAG_NEW_METADATA | LTTNG_VIEWER_FLAG_NEW_STREAM)) {
             BT_CLOGD("Reply with any one flags set means we should retry: response=%s",
                      lttng_viewer_get_packet_return_code_string(rp_status));
-            return CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
+            return LTTNG_LIVE_GET_STREAM_BYTES_STATUS_AGAIN;
         }
         BT_CLOGE_APPEND_CAUSE("Received get_data_packet response: error");
-        return CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
+        return LTTNG_LIVE_GET_STREAM_BYTES_STATUS_ERROR;
     case LTTNG_VIEWER_GET_PACKET_EOF:
-        return CTF_MSG_ITER_MEDIUM_STATUS_EOF;
+        return LTTNG_LIVE_GET_STREAM_BYTES_STATUS_EOF;
     default:
         BT_CLOGE_APPEND_CAUSE("Received get_data_packet response: unknown (%d)", rp_status);
-        return CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
+        return LTTNG_LIVE_GET_STREAM_BYTES_STATUS_ERROR;
     }
 
     if (req_len == 0) {
-        return CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
+        return LTTNG_LIVE_GET_STREAM_BYTES_STATUS_ERROR;
     }
 
     viewer_status = lttng_live_recv(viewer_connection, buf, req_len);
     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
         viewer_handle_recv_status(logCfg, viewer_status, "get data packet");
-        return viewer_status_to_ctf_msg_iter_medium_status(viewer_status);
+        return viewer_status_to_lttng_live_get_stream_bytes_status(viewer_status);
     }
     *recv_len = req_len;
 
-    return CTF_MSG_ITER_MEDIUM_STATUS_OK;
+    return LTTNG_LIVE_GET_STREAM_BYTES_STATUS_OK;
 }
 
 /*