src/plugins/ctf/lttng-live/viewer-connection.cpp

   1 /*
   2  * SPDX-License-Identifier: MIT
   3  *
   4  * Copyright 2019 Francis Deslauriers <francis.deslauriers@efficios.com>
   5  * Copyright 2016 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   6  */
   7
   8 #include <glib.h>
   9 #include <stdint.h>
  10 #include <stdio.h>
  11
  12 #include <babeltrace2/babeltrace.h>
  13
  14 #include "common/common.h"
  15 #include "compat/endian.h" /* IWYU pragma: keep  */
  16
  17 #include "data-stream.hpp"
  18 #include "lttng-live.hpp"
  19 #include "lttng-viewer-abi.hpp"
  20 #include "metadata.hpp"
  21 #include "viewer-connection.hpp"
  22
  23 #define viewer_handle_send_recv_status(_status, _action, _msg_str)                                 \
  24     do {                                                                                           \
  25         switch (_status) {                                                                         \
  26         case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED:                                                 \
  27             break;                                                                                 \
  28         case LTTNG_LIVE_VIEWER_STATUS_ERROR:                                                       \
  29             BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,                                \
  30                                          "Error " _action " " _msg_str);                           \
  31             break;                                                                                 \
  32         default:                                                                                   \
  33             bt_common_abort();                                                                     \
  34         }                                                                                          \
  35     } while (0)
  36
  37 #define viewer_handle_send_status(_status, _msg_str)                                               \
  38     viewer_handle_send_recv_status(_status, "sending", _msg_str)
  39
  40 #define viewer_handle_recv_status(_status, _msg_str)                                               \
  41     viewer_handle_send_recv_status(_status, "receiving", _msg_str)
  42
  43 #define LTTNG_LIVE_CPPLOGE_APPEND_CAUSE_ERRNO(_msg, _fmt, ...)                                     \
  44     do {                                                                                           \
  45         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, _msg ": {}" _fmt,                  \
  46                                      bt_socket_errormsg(), ##__VA_ARGS__);                         \
  47     } while (0)
  48
  49 static inline enum lttng_live_iterator_status
  50 viewer_status_to_live_iterator_status(enum lttng_live_viewer_status viewer_status)
  51 {
  52     switch (viewer_status) {
  53     case LTTNG_LIVE_VIEWER_STATUS_OK:
  54         return LTTNG_LIVE_ITERATOR_STATUS_OK;
  55     case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED:
  56         return LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
  57     case LTTNG_LIVE_VIEWER_STATUS_ERROR:
  58         return LTTNG_LIVE_ITERATOR_STATUS_ERROR;
  59     }
  60
  61     bt_common_abort();
  62 }
  63
  64 static inline enum ctf_msg_iter_medium_status
  65 viewer_status_to_ctf_msg_iter_medium_status(enum lttng_live_viewer_status viewer_status)
  66 {
  67     switch (viewer_status) {
  68     case LTTNG_LIVE_VIEWER_STATUS_OK:
  69         return CTF_MSG_ITER_MEDIUM_STATUS_OK;
  70     case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED:
  71         return CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
  72     case LTTNG_LIVE_VIEWER_STATUS_ERROR:
  73         return CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
  74     }
  75
  76     bt_common_abort();
  77 }
  78
  79 static inline void viewer_connection_close_socket(struct live_viewer_connection *viewer_connection)
  80 {
  81     int ret = bt_socket_close(viewer_connection->control_sock);
  82     if (ret == -1) {
  83         BT_CPPLOGW_ERRNO_SPEC(viewer_connection->logger,
  84                               "Error closing viewer connection socket: ", ".");
  85     }
  86
  87     viewer_connection->control_sock = BT_INVALID_SOCKET;
  88 }
  89
  90 /*
  91  * This function receives a message from the Relay daemon.
  92  * If it received the entire message, it returns _OK,
  93  * If it's interrupted, it returns _INTERRUPTED,
  94  * otherwise, it returns _ERROR.
  95  */
  96 static enum lttng_live_viewer_status
  97 lttng_live_recv(struct live_viewer_connection *viewer_connection, void *buf, size_t len)
  98 {
  99     ssize_t received;
 100     size_t total_received = 0, to_receive = len;
 101     struct lttng_live_msg_iter *lttng_live_msg_iter = viewer_connection->lttng_live_msg_iter;
 102     enum lttng_live_viewer_status status;
 103     BT_SOCKET sock = viewer_connection->control_sock;
 104
 105     /*
 106      * Receive a message from the Relay.
 107      */
 108     do {
 109         received = bt_socket_recv(sock, (char *) buf + total_received, to_receive, 0);
 110         if (received == BT_SOCKET_ERROR) {
 111             if (bt_socket_interrupted()) {
 112                 if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
 113                     /*
 114                      * This interruption was due to a
 115                      * SIGINT and the graph is being torn
 116                      * down.
 117                      */
 118                     status = LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED;
 119                     lttng_live_msg_iter->was_interrupted = true;
 120                     goto end;
 121                 } else {
 122                     /*
 123                      * A signal was received, but the graph
 124                      * is not being torn down. Carry on.
 125                      */
 126                     continue;
 127                 }
 128             } else {
 129                 /*
 130                  * For any other types of socket error, close
 131                  * the socket and return an error.
 132                  */
 133                 LTTNG_LIVE_CPPLOGE_APPEND_CAUSE_ERRNO("Error receiving from Relay", ".");
 134
 135                 viewer_connection_close_socket(viewer_connection);
 136                 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 137                 goto end;
 138             }
 139         } else if (received == 0) {
 140             /*
 141              * The recv() call returned 0. This means the
 142              * connection was orderly shutdown from the other peer.
 143              * If that happens when we are trying to receive
 144              * a message from it, it means something when wrong.
 145              * Close the socket and return an error.
 146              */
 147             BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 148                                          "Remote side has closed connection");
 149             viewer_connection_close_socket(viewer_connection);
 150             status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 151             goto end;
 152         }
 153
 154         BT_ASSERT(received <= to_receive);
 155         total_received += received;
 156         to_receive -= received;
 157
 158     } while (to_receive > 0);
 159
 160     BT_ASSERT(total_received == len);
 161     status = LTTNG_LIVE_VIEWER_STATUS_OK;
 162
 163 end:
 164     return status;
 165 }
 166
 167 /*
 168  * This function sends a message to the Relay daemon.
 169  * If it send the message, it returns _OK,
 170  * If it's interrupted, it returns _INTERRUPTED,
 171  * otherwise, it returns _ERROR.
 172  */
 173 static enum lttng_live_viewer_status
 174 lttng_live_send(struct live_viewer_connection *viewer_connection, const void *buf, size_t len)
 175 {
 176     enum lttng_live_viewer_status status;
 177     struct lttng_live_msg_iter *lttng_live_msg_iter = viewer_connection->lttng_live_msg_iter;
 178     BT_SOCKET sock = viewer_connection->control_sock;
 179     size_t to_send = len;
 180     ssize_t total_sent = 0;
 181
 182     do {
 183         ssize_t sent = bt_socket_send_nosigpipe(sock, (char *) buf + total_sent, to_send);
 184         if (sent == BT_SOCKET_ERROR) {
 185             if (bt_socket_interrupted()) {
 186                 if (lttng_live_graph_is_canceled(lttng_live_msg_iter)) {
 187                     /*
 188                      * This interruption was a SIGINT and
 189                      * the graph is being teared down.
 190                      */
 191                     status = LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED;
 192                     lttng_live_msg_iter->was_interrupted = true;
 193                     goto end;
 194                 } else {
 195                     /*
 196                      * A signal was received, but the graph
 197                      * is not being teared down. Carry on.
 198                      */
 199                     continue;
 200                 }
 201             } else {
 202                 /*
 203                  * For any other types of socket error, close
 204                  * the socket and return an error.
 205                  */
 206                 LTTNG_LIVE_CPPLOGE_APPEND_CAUSE_ERRNO("Error sending to Relay", ".");
 207
 208                 viewer_connection_close_socket(viewer_connection);
 209                 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 210                 goto end;
 211             }
 212         }
 213
 214         BT_ASSERT(sent <= to_send);
 215         total_sent += sent;
 216         to_send -= sent;
 217
 218     } while (to_send > 0);
 219
 220     BT_ASSERT(total_sent == len);
 221     status = LTTNG_LIVE_VIEWER_STATUS_OK;
 222
 223 end:
 224     return status;
 225 }
 226
 227 static int parse_url(struct live_viewer_connection *viewer_connection)
 228 {
 229     char error_buf[256] = {0};
 230     struct bt_common_lttng_live_url_parts lttng_live_url_parts = {};
 231     int ret = -1;
 232     const char *path = viewer_connection->url->str;
 233
 234     if (!path) {
 235         goto end;
 236     }
 237
 238     lttng_live_url_parts = bt_common_parse_lttng_live_url(path, error_buf, sizeof(error_buf));
 239     if (!lttng_live_url_parts.proto) {
 240         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Invalid LTTng live URL format: {}",
 241                                      error_buf);
 242         goto end;
 243     }
 244     viewer_connection->proto = lttng_live_url_parts.proto;
 245     lttng_live_url_parts.proto = NULL;
 246
 247     viewer_connection->relay_hostname = lttng_live_url_parts.hostname;
 248     lttng_live_url_parts.hostname = NULL;
 249
 250     if (lttng_live_url_parts.port >= 0) {
 251         viewer_connection->port = lttng_live_url_parts.port;
 252     } else {
 253         viewer_connection->port = LTTNG_DEFAULT_NETWORK_VIEWER_PORT;
 254     }
 255
 256     viewer_connection->target_hostname = lttng_live_url_parts.target_hostname;
 257     lttng_live_url_parts.target_hostname = NULL;
 258
 259     if (lttng_live_url_parts.session_name) {
 260         viewer_connection->session_name = lttng_live_url_parts.session_name;
 261         lttng_live_url_parts.session_name = NULL;
 262     }
 263
 264     ret = 0;
 265
 266 end:
 267     bt_common_destroy_lttng_live_url_parts(&lttng_live_url_parts);
 268     return ret;
 269 }
 270
 271 static enum lttng_live_viewer_status
 272 lttng_live_handshake(struct live_viewer_connection *viewer_connection)
 273 {
 274     struct lttng_viewer_cmd cmd;
 275     struct lttng_viewer_connect connect;
 276     enum lttng_live_viewer_status status;
 277     const size_t cmd_buf_len = sizeof(cmd) + sizeof(connect);
 278     char cmd_buf[cmd_buf_len];
 279
 280     BT_CPPLOGD_SPEC(viewer_connection->logger,
 281                     "Handshaking with the relay daemon: cmd={}, major-version={}, minor-version={}",
 282                     LTTNG_VIEWER_CONNECT, LTTNG_LIVE_MAJOR, LTTNG_LIVE_MINOR);
 283
 284     cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
 285     cmd.data_size = htobe64((uint64_t) sizeof(connect));
 286     cmd.cmd_version = htobe32(0);
 287
 288     connect.viewer_session_id = -1ULL; /* will be set on recv */
 289     connect.major = htobe32(LTTNG_LIVE_MAJOR);
 290     connect.minor = htobe32(LTTNG_LIVE_MINOR);
 291     connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
 292
 293     /*
 294      * Merge the cmd and connection request to prevent a write-write
 295      * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
 296      * second write to be performed quickly in presence of Nagle's algorithm
 297      */
 298     memcpy(cmd_buf, &cmd, sizeof(cmd));
 299     memcpy(cmd_buf + sizeof(cmd), &connect, sizeof(connect));
 300
 301     status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
 302     if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 303         viewer_handle_send_status(status, "viewer connect command");
 304         goto end;
 305     }
 306
 307     status = lttng_live_recv(viewer_connection, &connect, sizeof(connect));
 308     if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 309         viewer_handle_recv_status(status, "viewer connect reply");
 310         goto end;
 311     }
 312
 313     BT_CPPLOGI_SPEC(viewer_connection->logger, "Received viewer session ID : {}",
 314                     (uint64_t) be64toh(connect.viewer_session_id));
 315     BT_CPPLOGI_SPEC(viewer_connection->logger, "Relayd version : {}.{}", be32toh(connect.major),
 316                     be32toh(connect.minor));
 317
 318     if (LTTNG_LIVE_MAJOR != be32toh(connect.major)) {
 319         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 320                                      "Incompatible lttng-relayd protocol");
 321         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 322         goto end;
 323     }
 324     /* Use the smallest protocol version implemented. */
 325     if (LTTNG_LIVE_MINOR > be32toh(connect.minor)) {
 326         viewer_connection->minor = be32toh(connect.minor);
 327     } else {
 328         viewer_connection->minor = LTTNG_LIVE_MINOR;
 329     }
 330     viewer_connection->major = LTTNG_LIVE_MAJOR;
 331
 332     status = LTTNG_LIVE_VIEWER_STATUS_OK;
 333
 334     goto end;
 335
 336 end:
 337     return status;
 338 }
 339
 340 static enum lttng_live_viewer_status
 341 lttng_live_connect_viewer(struct live_viewer_connection *viewer_connection)
 342 {
 343     struct hostent *host;
 344     struct sockaddr_in server_addr;
 345     enum lttng_live_viewer_status status;
 346
 347     if (parse_url(viewer_connection)) {
 348         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Failed to parse URL");
 349         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 350         goto error;
 351     }
 352
 353     BT_CPPLOGD_SPEC(
 354         viewer_connection->logger,
 355         "Connecting to hostname : {}, port : {}, target hostname : {}, session name : {}, proto : {}",
 356         viewer_connection->relay_hostname->str, viewer_connection->port,
 357         !viewer_connection->target_hostname ? "<none>" : viewer_connection->target_hostname->str,
 358         !viewer_connection->session_name ? "<none>" : viewer_connection->session_name->str,
 359         viewer_connection->proto->str);
 360
 361     host = gethostbyname(viewer_connection->relay_hostname->str);
 362     if (!host) {
 363         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 364                                      "Cannot lookup hostname: hostname=\"{}\"",
 365                                      viewer_connection->relay_hostname->str);
 366         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 367         goto error;
 368     }
 369
 370     if ((viewer_connection->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == BT_INVALID_SOCKET) {
 371         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Socket creation failed: {}",
 372                                      bt_socket_errormsg());
 373         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 374         goto error;
 375     }
 376
 377     server_addr.sin_family = AF_INET;
 378     server_addr.sin_port = htons(viewer_connection->port);
 379     server_addr.sin_addr = *((struct in_addr *) host->h_addr);
 380     memset(&(server_addr.sin_zero), 0, 8);
 381
 382     if (connect(viewer_connection->control_sock, (struct sockaddr *) &server_addr,
 383                 sizeof(struct sockaddr)) == BT_SOCKET_ERROR) {
 384         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Connection failed: {}",
 385                                      bt_socket_errormsg());
 386         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 387         goto error;
 388     }
 389
 390     status = lttng_live_handshake(viewer_connection);
 391
 392     /*
 393      * Only print error and append cause in case of error. not in case of
 394      * interruption.
 395      */
 396     if (status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
 397         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Viewer handshake failed");
 398         goto error;
 399     } else if (status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
 400         goto end;
 401     }
 402
 403     goto end;
 404
 405 error:
 406     if (viewer_connection->control_sock != BT_INVALID_SOCKET) {
 407         if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
 408             BT_CPPLOGW_SPEC(viewer_connection->logger, "Error closing socket: {}.",
 409                             bt_socket_errormsg());
 410         }
 411     }
 412     viewer_connection->control_sock = BT_INVALID_SOCKET;
 413 end:
 414     return status;
 415 }
 416
 417 static void lttng_live_disconnect_viewer(struct live_viewer_connection *viewer_connection)
 418 {
 419     if (viewer_connection->control_sock == BT_INVALID_SOCKET) {
 420         return;
 421     }
 422     if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
 423         BT_CPPLOGW_SPEC(viewer_connection->logger, "Error closing socket: {}",
 424                         bt_socket_errormsg());
 425         viewer_connection->control_sock = BT_INVALID_SOCKET;
 426     }
 427 }
 428
 429 static int list_update_session(bt_value *results, const struct lttng_viewer_session *session,
 430                                bool *_found, struct live_viewer_connection *viewer_connection)
 431 {
 432     int ret = 0;
 433     uint64_t i, len;
 434     bt_value *map = NULL;
 435     bt_value *hostname = NULL;
 436     bt_value *session_name = NULL;
 437     bt_value *btval = NULL;
 438     bool found = false;
 439
 440     len = bt_value_array_get_length(results);
 441     for (i = 0; i < len; i++) {
 442         const char *hostname_str = NULL;
 443         const char *session_name_str = NULL;
 444
 445         map = bt_value_array_borrow_element_by_index(results, i);
 446         hostname = bt_value_map_borrow_entry_value(map, "target-hostname");
 447         if (!hostname) {
 448             BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 449                                          "Error borrowing \"target-hostname\" entry.");
 450             ret = -1;
 451             goto end;
 452         }
 453         session_name = bt_value_map_borrow_entry_value(map, "session-name");
 454         if (!session_name) {
 455             BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 456                                          "Error borrowing \"session-name\" entry.");
 457             ret = -1;
 458             goto end;
 459         }
 460         hostname_str = bt_value_string_get(hostname);
 461         session_name_str = bt_value_string_get(session_name);
 462
 463         if (strcmp(session->hostname, hostname_str) == 0 &&
 464             strcmp(session->session_name, session_name_str) == 0) {
 465             int64_t val;
 466             uint32_t streams = be32toh(session->streams);
 467             uint32_t clients = be32toh(session->clients);
 468
 469             found = true;
 470
 471             btval = bt_value_map_borrow_entry_value(map, "stream-count");
 472             if (!btval) {
 473                 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 474                                              "Error borrowing \"stream-count\" entry.");
 475                 ret = -1;
 476                 goto end;
 477             }
 478             val = bt_value_integer_unsigned_get(btval);
 479             /* sum */
 480             val += streams;
 481             bt_value_integer_unsigned_set(btval, val);
 482
 483             btval = bt_value_map_borrow_entry_value(map, "client-count");
 484             if (!btval) {
 485                 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 486                                              "Error borrowing \"client-count\" entry.");
 487                 ret = -1;
 488                 goto end;
 489             }
 490             val = bt_value_integer_unsigned_get(btval);
 491             /* max */
 492             val = bt_max_t(int64_t, clients, val);
 493             bt_value_integer_unsigned_set(btval, val);
 494         }
 495
 496         if (found) {
 497             break;
 498         }
 499     }
 500 end:
 501     *_found = found;
 502     return ret;
 503 }
 504
 505 static int list_append_session(bt_value *results, GString *base_url,
 506                                const struct lttng_viewer_session *session,
 507                                struct live_viewer_connection *viewer_connection)
 508 {
 509     int ret = 0;
 510     bt_value_map_insert_entry_status insert_status;
 511     bt_value_array_append_element_status append_status;
 512     bt_value *map = NULL;
 513     GString *url = NULL;
 514     bool found = false;
 515
 516     /*
 517      * If the session already exists, add the stream count to it,
 518      * and do max of client counts.
 519      */
 520     ret = list_update_session(results, session, &found, viewer_connection);
 521     if (ret || found) {
 522         goto end;
 523     }
 524
 525     map = bt_value_map_create();
 526     if (!map) {
 527         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error creating map value.");
 528         ret = -1;
 529         goto end;
 530     }
 531
 532     if (base_url->len < 1) {
 533         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 534                                      "Error: base_url length smaller than 1.");
 535         ret = -1;
 536         goto end;
 537     }
 538     /*
 539      * key = "url",
 540      * value = <string>,
 541      */
 542     url = g_string_new(base_url->str);
 543     g_string_append(url, "/host/");
 544     g_string_append(url, session->hostname);
 545     g_string_append_c(url, '/');
 546     g_string_append(url, session->session_name);
 547
 548     insert_status = bt_value_map_insert_string_entry(map, "url", url->str);
 549     if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 550         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error inserting \"url\" entry.");
 551         ret = -1;
 552         goto end;
 553     }
 554
 555     /*
 556      * key = "target-hostname",
 557      * value = <string>,
 558      */
 559     insert_status = bt_value_map_insert_string_entry(map, "target-hostname", session->hostname);
 560     if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 561         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 562                                      "Error inserting \"target-hostname\" entry.");
 563         ret = -1;
 564         goto end;
 565     }
 566
 567     /*
 568      * key = "session-name",
 569      * value = <string>,
 570      */
 571     insert_status = bt_value_map_insert_string_entry(map, "session-name", session->session_name);
 572     if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 573         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 574                                      "Error inserting \"session-name\" entry.");
 575         ret = -1;
 576         goto end;
 577     }
 578
 579     /*
 580      * key = "timer-us",
 581      * value = <integer>,
 582      */
 583     {
 584         uint32_t live_timer = be32toh(session->live_timer);
 585
 586         insert_status = bt_value_map_insert_unsigned_integer_entry(map, "timer-us", live_timer);
 587         if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 588             BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 589                                          "Error inserting \"timer-us\" entry.");
 590             ret = -1;
 591             goto end;
 592         }
 593     }
 594
 595     /*
 596      * key = "stream-count",
 597      * value = <integer>,
 598      */
 599     {
 600         uint32_t streams = be32toh(session->streams);
 601
 602         insert_status = bt_value_map_insert_unsigned_integer_entry(map, "stream-count", streams);
 603         if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 604             BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 605                                          "Error inserting \"stream-count\" entry.");
 606             ret = -1;
 607             goto end;
 608         }
 609     }
 610
 611     /*
 612      * key = "client-count",
 613      * value = <integer>,
 614      */
 615     {
 616         uint32_t clients = be32toh(session->clients);
 617
 618         insert_status = bt_value_map_insert_unsigned_integer_entry(map, "client-count", clients);
 619         if (insert_status != BT_VALUE_MAP_INSERT_ENTRY_STATUS_OK) {
 620             BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 621                                          "Error inserting \"client-count\" entry.");
 622             ret = -1;
 623             goto end;
 624         }
 625     }
 626
 627     append_status = bt_value_array_append_element(results, map);
 628     if (append_status != BT_VALUE_ARRAY_APPEND_ELEMENT_STATUS_OK) {
 629         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error appending map to results.");
 630         ret = -1;
 631     }
 632
 633 end:
 634     if (url) {
 635         g_string_free(url, true);
 636     }
 637     BT_VALUE_PUT_REF_AND_RESET(map);
 638     return ret;
 639 }
 640
 641 /*
 642  * Data structure returned:
 643  *
 644  * {
 645  *   <array> = {
 646  *     [n] = {
 647  *       <map> = {
 648  *         {
 649  *           key = "url",
 650  *           value = <string>,
 651  *         },
 652  *         {
 653  *           key = "target-hostname",
 654  *           value = <string>,
 655  *         },
 656  *         {
 657  *           key = "session-name",
 658  *           value = <string>,
 659  *         },
 660  *         {
 661  *           key = "timer-us",
 662  *           value = <integer>,
 663  *         },
 664  *         {
 665  *           key = "stream-count",
 666  *           value = <integer>,
 667  *         },
 668  *         {
 669  *           key = "client-count",
 670  *           value = <integer>,
 671  *         },
 672  *       },
 673  *     }
 674  *   }
 675  */
 676
 677 bt_component_class_query_method_status
 678 live_viewer_connection_list_sessions(struct live_viewer_connection *viewer_connection,
 679                                      const bt_value **user_result)
 680 {
 681     bt_component_class_query_method_status status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_OK;
 682     bt_value *result = NULL;
 683     enum lttng_live_viewer_status viewer_status;
 684     struct lttng_viewer_cmd cmd;
 685     struct lttng_viewer_list_sessions list;
 686     uint32_t i, sessions_count;
 687
 688     result = bt_value_array_create();
 689     if (!result) {
 690         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error creating array");
 691         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_MEMORY_ERROR;
 692         goto error;
 693     }
 694
 695     BT_CPPLOGD_SPEC(viewer_connection->logger, "Requesting list of sessions: cmd={}",
 696                     LTTNG_VIEWER_LIST_SESSIONS);
 697
 698     cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 699     cmd.data_size = htobe64((uint64_t) 0);
 700     cmd.cmd_version = htobe32(0);
 701
 702     viewer_status = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 703     if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
 704         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 705                                      "Error sending list sessions command");
 706         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
 707         goto error;
 708     } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
 709         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
 710         goto error;
 711     }
 712
 713     viewer_status = lttng_live_recv(viewer_connection, &list, sizeof(list));
 714     if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
 715         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error receiving session list");
 716         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
 717         goto error;
 718     } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
 719         status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
 720         goto error;
 721     }
 722
 723     sessions_count = be32toh(list.sessions_count);
 724     for (i = 0; i < sessions_count; i++) {
 725         struct lttng_viewer_session lsession;
 726
 727         viewer_status = lttng_live_recv(viewer_connection, &lsession, sizeof(lsession));
 728         if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
 729             BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error receiving session:");
 730             status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
 731             goto error;
 732         } else if (viewer_status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
 733             status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_AGAIN;
 734             goto error;
 735         }
 736
 737         lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
 738         lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 739         if (list_append_session(result, viewer_connection->url, &lsession, viewer_connection)) {
 740             BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error appending session");
 741             status = BT_COMPONENT_CLASS_QUERY_METHOD_STATUS_ERROR;
 742             goto error;
 743         }
 744     }
 745
 746     *user_result = result;
 747     goto end;
 748 error:
 749     BT_VALUE_PUT_REF_AND_RESET(result);
 750 end:
 751     return status;
 752 }
 753
 754 static enum lttng_live_viewer_status
 755 lttng_live_query_session_ids(struct lttng_live_msg_iter *lttng_live_msg_iter)
 756 {
 757     struct lttng_viewer_cmd cmd;
 758     struct lttng_viewer_list_sessions list;
 759     struct lttng_viewer_session lsession;
 760     uint32_t i, sessions_count;
 761     uint64_t session_id;
 762     enum lttng_live_viewer_status status;
 763     struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
 764
 765     BT_CPPLOGD_SPEC(viewer_connection->logger,
 766                     "Asking the relay daemon for the list of sessions: cmd={}",
 767                     LTTNG_VIEWER_LIST_SESSIONS);
 768
 769     cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 770     cmd.data_size = htobe64((uint64_t) 0);
 771     cmd.cmd_version = htobe32(0);
 772
 773     status = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 774     if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 775         viewer_handle_send_status(status, "list sessions command");
 776         goto end;
 777     }
 778
 779     status = lttng_live_recv(viewer_connection, &list, sizeof(list));
 780     if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 781         viewer_handle_recv_status(status, "session list reply");
 782         goto end;
 783     }
 784
 785     sessions_count = be32toh(list.sessions_count);
 786     for (i = 0; i < sessions_count; i++) {
 787         status = lttng_live_recv(viewer_connection, &lsession, sizeof(lsession));
 788         if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 789             viewer_handle_recv_status(status, "session reply");
 790             goto end;
 791         }
 792         lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
 793         lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 794         session_id = be64toh(lsession.id);
 795
 796         BT_CPPLOGI_SPEC(viewer_connection->logger,
 797                         "Adding session to internal list: "
 798                         "session-id={}, hostname=\"{}\", session-name=\"{}\"",
 799                         session_id, lsession.hostname, lsession.session_name);
 800
 801         if ((strncmp(lsession.session_name, viewer_connection->session_name->str,
 802                      LTTNG_VIEWER_NAME_MAX) == 0) &&
 803             (strncmp(lsession.hostname, viewer_connection->target_hostname->str,
 804                      LTTNG_VIEWER_HOST_NAME_MAX) == 0)) {
 805             if (lttng_live_add_session(lttng_live_msg_iter, session_id, lsession.hostname,
 806                                        lsession.session_name)) {
 807                 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 808                                              "Failed to add live session");
 809                 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 810                 goto end;
 811             }
 812         }
 813     }
 814
 815     status = LTTNG_LIVE_VIEWER_STATUS_OK;
 816
 817 end:
 818     return status;
 819 }
 820
 821 enum lttng_live_viewer_status
 822 lttng_live_create_viewer_session(struct lttng_live_msg_iter *lttng_live_msg_iter)
 823 {
 824     struct lttng_viewer_cmd cmd;
 825     struct lttng_viewer_create_session_response resp;
 826     enum lttng_live_viewer_status status;
 827     struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
 828
 829     BT_CPPLOGD_SPEC(viewer_connection->logger, "Creating a viewer session: cmd={}",
 830                     LTTNG_VIEWER_CREATE_SESSION);
 831
 832     cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
 833     cmd.data_size = htobe64((uint64_t) 0);
 834     cmd.cmd_version = htobe32(0);
 835
 836     status = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 837     if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 838         viewer_handle_send_status(status, "create session command");
 839         goto end;
 840     }
 841
 842     status = lttng_live_recv(viewer_connection, &resp, sizeof(resp));
 843     if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 844         viewer_handle_recv_status(status, "create session reply");
 845         goto end;
 846     }
 847
 848     if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
 849         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error creating viewer session");
 850         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 851         goto end;
 852     }
 853
 854     status = lttng_live_query_session_ids(lttng_live_msg_iter);
 855     if (status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
 856         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 857                                      "Failed to query live viewer session ids");
 858         goto end;
 859     } else if (status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
 860         goto end;
 861     }
 862
 863 end:
 864     return status;
 865 }
 866
 867 static enum lttng_live_viewer_status receive_streams(struct lttng_live_session *session,
 868                                                      uint32_t stream_count,
 869                                                      bt_self_message_iterator *self_msg_iter)
 870 {
 871     uint32_t i;
 872     struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
 873     enum lttng_live_viewer_status status;
 874     struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
 875
 876     BT_CPPLOGI_SPEC(viewer_connection->logger, "Getting {} new streams", stream_count);
 877     for (i = 0; i < stream_count; i++) {
 878         struct lttng_viewer_stream stream;
 879         struct lttng_live_stream_iterator *live_stream;
 880         uint64_t stream_id;
 881         uint64_t ctf_trace_id;
 882
 883         status = lttng_live_recv(viewer_connection, &stream, sizeof(stream));
 884         if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 885             viewer_handle_recv_status(status, "stream reply");
 886             goto end;
 887         }
 888         stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
 889         stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 890         stream_id = be64toh(stream.id);
 891         ctf_trace_id = be64toh(stream.ctf_trace_id);
 892
 893         if (stream.metadata_flag) {
 894             BT_CPPLOGI_SPEC(viewer_connection->logger, "    metadata stream {} : {}/{}", stream_id,
 895                             stream.path_name, stream.channel_name);
 896             if (lttng_live_metadata_create_stream(session, ctf_trace_id, stream_id)) {
 897                 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 898                                              "Error creating metadata stream");
 899                 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 900                 goto end;
 901             }
 902             session->lazy_stream_msg_init = true;
 903         } else {
 904             BT_CPPLOGI_SPEC(viewer_connection->logger, "    stream {} : {}/{}", stream_id,
 905                             stream.path_name, stream.channel_name);
 906             live_stream =
 907                 lttng_live_stream_iterator_create(session, ctf_trace_id, stream_id, self_msg_iter);
 908             if (!live_stream) {
 909                 BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error creating stream");
 910                 status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 911                 goto end;
 912             }
 913         }
 914     }
 915     status = LTTNG_LIVE_VIEWER_STATUS_OK;
 916
 917 end:
 918     return status;
 919 }
 920
 921 enum lttng_live_viewer_status lttng_live_session_attach(struct lttng_live_session *session,
 922                                                         bt_self_message_iterator *self_msg_iter)
 923 {
 924     struct lttng_viewer_cmd cmd;
 925     enum lttng_live_viewer_status status;
 926     struct lttng_viewer_attach_session_request rq;
 927     struct lttng_viewer_attach_session_response rp;
 928     struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
 929     struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
 930     uint64_t session_id = session->id;
 931     uint32_t streams_count;
 932     const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
 933     char cmd_buf[cmd_buf_len];
 934
 935     BT_CPPLOGD_SPEC(viewer_connection->logger,
 936                     "Attaching to session: cmd={}, session-id={}, seek={}",
 937                     LTTNG_VIEWER_ATTACH_SESSION, session_id, LTTNG_VIEWER_SEEK_LAST);
 938
 939     cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
 940     cmd.data_size = htobe64((uint64_t) sizeof(rq));
 941     cmd.cmd_version = htobe32(0);
 942
 943     memset(&rq, 0, sizeof(rq));
 944     rq.session_id = htobe64(session_id);
 945     // TODO: add cmd line parameter to select seek beginning
 946     // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
 947     rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST);
 948
 949     /*
 950      * Merge the cmd and connection request to prevent a write-write
 951      * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
 952      * second write to be performed quickly in presence of Nagle's algorithm.
 953      */
 954     memcpy(cmd_buf, &cmd, sizeof(cmd));
 955     memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
 956     status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
 957     if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 958         viewer_handle_send_status(status, "attach session command");
 959         goto end;
 960     }
 961
 962     status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
 963     if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
 964         viewer_handle_recv_status(status, "attach session reply");
 965         goto end;
 966     }
 967
 968     streams_count = be32toh(rp.streams_count);
 969     switch (be32toh(rp.status)) {
 970     case LTTNG_VIEWER_ATTACH_OK:
 971         break;
 972     case LTTNG_VIEWER_ATTACH_UNK:
 973         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Session id {} is unknown",
 974                                      session_id);
 975         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 976         goto end;
 977     case LTTNG_VIEWER_ATTACH_ALREADY:
 978         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
 979                                      "There is already a viewer attached to this session");
 980         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 981         goto end;
 982     case LTTNG_VIEWER_ATTACH_NOT_LIVE:
 983         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Not a live session");
 984         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 985         goto end;
 986     case LTTNG_VIEWER_ATTACH_SEEK_ERR:
 987         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Wrong seek parameter");
 988         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 989         goto end;
 990     default:
 991         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Unknown attach return code {}",
 992                                      be32toh(rp.status));
 993         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
 994         goto end;
 995     }
 996
 997     /* We receive the initial list of streams. */
 998     status = receive_streams(session, streams_count, self_msg_iter);
 999     switch (status) {
1000     case LTTNG_LIVE_VIEWER_STATUS_OK:
1001         break;
1002     case LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED:
1003         goto end;
1004     case LTTNG_LIVE_VIEWER_STATUS_ERROR:
1005         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Error receiving streams");
1006         goto end;
1007     default:
1008         bt_common_abort();
1009     }
1010
1011     session->attached = true;
1012     session->new_streams_needed = false;
1013
1014 end:
1015     return status;
1016 }
1017
1018 enum lttng_live_viewer_status lttng_live_session_detach(struct lttng_live_session *session)
1019 {
1020     struct lttng_viewer_cmd cmd;
1021     enum lttng_live_viewer_status status;
1022     struct lttng_viewer_detach_session_request rq;
1023     struct lttng_viewer_detach_session_response rp;
1024     struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
1025     struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1026     uint64_t session_id = session->id;
1027     const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1028     char cmd_buf[cmd_buf_len];
1029
1030     /*
1031      * The session might already be detached and the viewer socket might
1032      * already been closed. This happens when calling this function when
1033      * tearing down the graph after an error.
1034      */
1035     if (!session->attached || viewer_connection->control_sock == BT_INVALID_SOCKET) {
1036         return LTTNG_LIVE_VIEWER_STATUS_OK;
1037     }
1038
1039     BT_CPPLOGD_SPEC(viewer_connection->logger, "Detaching from session: cmd={}, session-id={}",
1040                     LTTNG_VIEWER_DETACH_SESSION, session_id);
1041
1042     cmd.cmd = htobe32(LTTNG_VIEWER_DETACH_SESSION);
1043     cmd.data_size = htobe64((uint64_t) sizeof(rq));
1044     cmd.cmd_version = htobe32(0);
1045
1046     memset(&rq, 0, sizeof(rq));
1047     rq.session_id = htobe64(session_id);
1048
1049     /*
1050      * Merge the cmd and connection request to prevent a write-write
1051      * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1052      * second write to be performed quickly in presence of Nagle's algorithm.
1053      */
1054     memcpy(cmd_buf, &cmd, sizeof(cmd));
1055     memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1056     status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1057     if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1058         viewer_handle_send_status(status, "detach session command");
1059         goto end;
1060     }
1061
1062     status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1063     if (status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1064         viewer_handle_recv_status(status, "detach session reply");
1065         goto end;
1066     }
1067
1068     switch (be32toh(rp.status)) {
1069     case LTTNG_VIEWER_DETACH_SESSION_OK:
1070         break;
1071     case LTTNG_VIEWER_DETACH_SESSION_UNK:
1072         BT_CPPLOGW_SPEC(viewer_connection->logger, "Session id {} is unknown", session_id);
1073         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1074         goto end;
1075     case LTTNG_VIEWER_DETACH_SESSION_ERR:
1076         BT_CPPLOGW_SPEC(viewer_connection->logger, "Error detaching session id {}", session_id);
1077         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1078         goto end;
1079     default:
1080         BT_CPPLOGE_SPEC(viewer_connection->logger, "Unknown detach return code {}",
1081                         be32toh(rp.status));
1082         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1083         goto end;
1084     }
1085
1086     session->attached = false;
1087
1088     status = LTTNG_LIVE_VIEWER_STATUS_OK;
1089
1090 end:
1091     return status;
1092 }
1093
1094 enum lttng_live_get_one_metadata_status
1095 lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace, FILE *fp, size_t *reply_len)
1096 {
1097     uint64_t len = 0;
1098     enum lttng_live_get_one_metadata_status status;
1099     enum lttng_live_viewer_status viewer_status;
1100     struct lttng_viewer_cmd cmd;
1101     struct lttng_viewer_get_metadata rq;
1102     struct lttng_viewer_metadata_packet rp;
1103     gchar *data = NULL;
1104     ssize_t writelen;
1105     struct lttng_live_session *session = trace->session;
1106     struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
1107     struct lttng_live_metadata *metadata = trace->metadata;
1108     struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1109     const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1110     char cmd_buf[cmd_buf_len];
1111
1112     BT_CPPLOGD_SPEC(viewer_connection->logger,
1113                     "Requesting new metadata for trace:"
1114                     "cmd={}, trace-id={}, metadata-stream-id={}",
1115                     LTTNG_VIEWER_GET_METADATA, trace->id, metadata->stream_id);
1116
1117     rq.stream_id = htobe64(metadata->stream_id);
1118     cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
1119     cmd.data_size = htobe64((uint64_t) sizeof(rq));
1120     cmd.cmd_version = htobe32(0);
1121
1122     /*
1123      * Merge the cmd and connection request to prevent a write-write
1124      * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1125      * second write to be performed quickly in presence of Nagle's algorithm.
1126      */
1127     memcpy(cmd_buf, &cmd, sizeof(cmd));
1128     memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1129     viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1130     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1131         viewer_handle_send_status(viewer_status, "get metadata command");
1132         status = (enum lttng_live_get_one_metadata_status) viewer_status;
1133         goto end;
1134     }
1135
1136     viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1137     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1138         viewer_handle_recv_status(viewer_status, "get metadata reply");
1139         status = (enum lttng_live_get_one_metadata_status) viewer_status;
1140         goto end;
1141     }
1142
1143     switch (be32toh(rp.status)) {
1144     case LTTNG_VIEWER_METADATA_OK:
1145         BT_CPPLOGD_SPEC(viewer_connection->logger, "Received get_metadata response: ok");
1146         break;
1147     case LTTNG_VIEWER_NO_NEW_METADATA:
1148         BT_CPPLOGD_SPEC(viewer_connection->logger, "Received get_metadata response: no new");
1149         status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_END;
1150         goto end;
1151     case LTTNG_VIEWER_METADATA_ERR:
1152         /*
1153              * The Relayd cannot find this stream id. Maybe its
1154              * gone already. This can happen in short lived UST app
1155              * in a per-pid session.
1156              */
1157         BT_CPPLOGD_SPEC(viewer_connection->logger, "Received get_metadata response: error");
1158         status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_CLOSED;
1159         goto end;
1160     default:
1161         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
1162                                      "Received get_metadata response: unknown");
1163         status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
1164         goto end;
1165     }
1166
1167     len = be64toh(rp.len);
1168     if (len == 0) {
1169         /*
1170          * We received a `LTTNG_VIEWER_METADATA_OK` with a packet
1171          * length of 0. This means we must try again. This scenario
1172          * arises when a clear command is performed on an lttng session.
1173          */
1174         BT_CPPLOGD_SPEC(
1175             viewer_connection->logger,
1176             "Expecting a metadata packet of size 0. Retry to get a packet from the relay.");
1177         goto empty_metadata_packet_retry;
1178     }
1179
1180     BT_CPPLOGD_SPEC(viewer_connection->logger, "Writing {} bytes to metadata", len);
1181     if (len <= 0) {
1182         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Erroneous response length");
1183         status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
1184         goto end;
1185     }
1186
1187     data = g_new0(gchar, len);
1188     if (!data) {
1189         BT_CPPLOGE_ERRNO_APPEND_CAUSE_SPEC(viewer_connection->logger,
1190                                            "Failed to allocate data buffer", ".");
1191         status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
1192         goto end;
1193     }
1194
1195     viewer_status = lttng_live_recv(viewer_connection, data, len);
1196     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1197         viewer_handle_recv_status(viewer_status, "get metadata packet");
1198         status = (enum lttng_live_get_one_metadata_status) viewer_status;
1199         goto end;
1200     }
1201
1202     /*
1203      * Write the metadata to the file handle.
1204      */
1205     writelen = fwrite(data, sizeof(uint8_t), len, fp);
1206     if (writelen != len) {
1207         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
1208                                      "Writing in the metadata file stream");
1209         status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_ERROR;
1210         goto end;
1211     }
1212
1213 empty_metadata_packet_retry:
1214     *reply_len = len;
1215     status = LTTNG_LIVE_GET_ONE_METADATA_STATUS_OK;
1216
1217 end:
1218     g_free(data);
1219     return status;
1220 }
1221
1222 /*
1223  * Assign the fields from a lttng_viewer_index to a packet_index.
1224  */
1225 static void lttng_index_to_packet_index(struct lttng_viewer_index *lindex,
1226                                         struct packet_index *pindex)
1227 {
1228     BT_ASSERT(lindex);
1229     BT_ASSERT(pindex);
1230
1231     pindex->offset = be64toh(lindex->offset);
1232     pindex->packet_size = be64toh(lindex->packet_size);
1233     pindex->content_size = be64toh(lindex->content_size);
1234     pindex->ts_cycles.timestamp_begin = be64toh(lindex->timestamp_begin);
1235     pindex->ts_cycles.timestamp_end = be64toh(lindex->timestamp_end);
1236     pindex->events_discarded = be64toh(lindex->events_discarded);
1237 }
1238
1239 static void lttng_live_need_new_streams(struct lttng_live_msg_iter *lttng_live_msg_iter)
1240 {
1241     uint64_t session_idx;
1242
1243     for (session_idx = 0; session_idx < lttng_live_msg_iter->sessions->len; session_idx++) {
1244         struct lttng_live_session *session =
1245             (lttng_live_session *) g_ptr_array_index(lttng_live_msg_iter->sessions, session_idx);
1246         BT_CPPLOGD_SPEC(lttng_live_msg_iter->logger,
1247                         "Marking session as needing new streams: "
1248                         "session-id={}",
1249                         session->id);
1250         session->new_streams_needed = true;
1251     }
1252 }
1253
1254 enum lttng_live_iterator_status
1255 lttng_live_get_next_index(struct lttng_live_msg_iter *lttng_live_msg_iter,
1256                           struct lttng_live_stream_iterator *stream, struct packet_index *index)
1257 {
1258     struct lttng_viewer_cmd cmd;
1259     struct lttng_viewer_get_next_index rq;
1260     enum lttng_live_viewer_status viewer_status;
1261     struct lttng_viewer_index rp;
1262     enum lttng_live_iterator_status status;
1263     struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1264     struct lttng_live_trace *trace = stream->trace;
1265     const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1266     char cmd_buf[cmd_buf_len];
1267     uint32_t flags, rp_status;
1268
1269     BT_CPPLOGD_SPEC(viewer_connection->logger,
1270                     "Requesting next index for stream: cmd={}, "
1271                     "viewer-stream-id={}",
1272                     LTTNG_VIEWER_GET_NEXT_INDEX, stream->viewer_stream_id);
1273     cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
1274     cmd.data_size = htobe64((uint64_t) sizeof(rq));
1275     cmd.cmd_version = htobe32(0);
1276
1277     memset(&rq, 0, sizeof(rq));
1278     rq.stream_id = htobe64(stream->viewer_stream_id);
1279
1280     /*
1281      * Merge the cmd and connection request to prevent a write-write
1282      * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1283      * second write to be performed quickly in presence of Nagle's algorithm.
1284      */
1285     memcpy(cmd_buf, &cmd, sizeof(cmd));
1286     memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1287
1288     viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1289     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1290         viewer_handle_send_status(viewer_status, "get next index command");
1291         goto error;
1292     }
1293
1294     viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1295     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1296         viewer_handle_recv_status(viewer_status, "get next index reply");
1297         goto error;
1298     }
1299
1300     flags = be32toh(rp.flags);
1301     rp_status = be32toh(rp.status);
1302
1303     BT_CPPLOGD_SPEC(
1304         viewer_connection->logger, "Received response from relay daemon: cmd=%s, response={}",
1305         LTTNG_VIEWER_GET_NEXT_INDEX, static_cast<lttng_viewer_next_index_return_code>(rp_status));
1306
1307     if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1308         BT_CPPLOGD_SPEC(viewer_connection->logger,
1309                         "Marking all sessions as possibly needing new streams: "
1310                         "response={}, response-flag=NEW_STREAM",
1311                         static_cast<lttng_viewer_next_index_return_code>(rp_status));
1312         lttng_live_need_new_streams(lttng_live_msg_iter);
1313     }
1314
1315     switch (rp_status) {
1316     case LTTNG_VIEWER_INDEX_INACTIVE:
1317     {
1318         uint64_t ctf_stream_class_id;
1319
1320         memset(index, 0, sizeof(struct packet_index));
1321         index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
1322         stream->current_inactivity_ts = index->ts_cycles.timestamp_end;
1323         ctf_stream_class_id = be64toh(rp.stream_id);
1324         if (stream->ctf_stream_class_id.is_set) {
1325             BT_ASSERT(stream->ctf_stream_class_id.value == ctf_stream_class_id);
1326         } else {
1327             stream->ctf_stream_class_id.value = ctf_stream_class_id;
1328             stream->ctf_stream_class_id.is_set = true;
1329         }
1330         lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_QUIESCENT);
1331         status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1332         break;
1333     }
1334     case LTTNG_VIEWER_INDEX_OK:
1335     {
1336         uint64_t ctf_stream_class_id;
1337
1338         lttng_index_to_packet_index(&rp, index);
1339         ctf_stream_class_id = be64toh(rp.stream_id);
1340         if (stream->ctf_stream_class_id.is_set) {
1341             BT_ASSERT(stream->ctf_stream_class_id.value == ctf_stream_class_id);
1342         } else {
1343             stream->ctf_stream_class_id.value = ctf_stream_class_id;
1344             stream->ctf_stream_class_id.is_set = true;
1345         }
1346         lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_DATA);
1347
1348         if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1349             BT_CPPLOGD_SPEC(viewer_connection->logger,
1350                             "Marking trace as needing new metadata: "
1351                             "response={}, response-flag=NEW_METADATA, trace-id={}",
1352                             static_cast<lttng_viewer_next_index_return_code>(rp_status), trace->id);
1353             trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
1354         }
1355         status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1356         break;
1357     }
1358     case LTTNG_VIEWER_INDEX_RETRY:
1359         memset(index, 0, sizeof(struct packet_index));
1360         lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_NO_DATA);
1361         status = LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1362         goto end;
1363     case LTTNG_VIEWER_INDEX_HUP:
1364         memset(index, 0, sizeof(struct packet_index));
1365         index->offset = EOF;
1366         lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_EOF);
1367         stream->has_stream_hung_up = true;
1368         status = LTTNG_LIVE_ITERATOR_STATUS_END;
1369         break;
1370     case LTTNG_VIEWER_INDEX_ERR:
1371         memset(index, 0, sizeof(struct packet_index));
1372         lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_NO_DATA);
1373         status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1374         goto end;
1375     default:
1376         BT_CPPLOGD_SPEC(viewer_connection->logger,
1377                         "Received get_next_index response: unknown value");
1378         memset(index, 0, sizeof(struct packet_index));
1379         lttng_live_stream_iterator_set_state(stream, LTTNG_LIVE_STREAM_ACTIVE_NO_DATA);
1380         status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1381         goto end;
1382     }
1383
1384     goto end;
1385
1386 error:
1387     status = viewer_status_to_live_iterator_status(viewer_status);
1388 end:
1389     return status;
1390 }
1391
1392 enum ctf_msg_iter_medium_status
1393 lttng_live_get_stream_bytes(struct lttng_live_msg_iter *lttng_live_msg_iter,
1394                             struct lttng_live_stream_iterator *stream, uint8_t *buf,
1395                             uint64_t offset, uint64_t req_len, uint64_t *recv_len)
1396 {
1397     enum ctf_msg_iter_medium_status status;
1398     enum lttng_live_viewer_status viewer_status;
1399     struct lttng_viewer_trace_packet rp;
1400     struct lttng_viewer_cmd cmd;
1401     struct lttng_viewer_get_packet rq;
1402     struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1403     struct lttng_live_trace *trace = stream->trace;
1404     const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1405     char cmd_buf[cmd_buf_len];
1406     uint32_t flags, rp_status;
1407
1408     BT_CPPLOGD_SPEC(viewer_connection->logger,
1409                     "Requesting data from stream: cmd={}, "
1410                     "offset={}, request-len={}",
1411                     LTTNG_VIEWER_GET_PACKET, offset, req_len);
1412
1413     cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
1414     cmd.data_size = htobe64((uint64_t) sizeof(rq));
1415     cmd.cmd_version = htobe32(0);
1416
1417     memset(&rq, 0, sizeof(rq));
1418     rq.stream_id = htobe64(stream->viewer_stream_id);
1419     rq.offset = htobe64(offset);
1420     rq.len = htobe32(req_len);
1421
1422     /*
1423      * Merge the cmd and connection request to prevent a write-write
1424      * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1425      * second write to be performed quickly in presence of Nagle's algorithm.
1426      */
1427     memcpy(cmd_buf, &cmd, sizeof(cmd));
1428     memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1429
1430     viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1431     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1432         viewer_handle_send_status(viewer_status, "get data packet command");
1433         goto error_convert_status;
1434     }
1435
1436     viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1437     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1438         viewer_handle_recv_status(viewer_status, "get data packet reply");
1439         goto error_convert_status;
1440     }
1441
1442     flags = be32toh(rp.flags);
1443     rp_status = be32toh(rp.status);
1444
1445     BT_CPPLOGD_SPEC(
1446         viewer_connection->logger, "Received response from relay daemon: cmd={}, response={}",
1447         LTTNG_VIEWER_GET_PACKET, static_cast<lttng_viewer_get_packet_return_code>(rp_status));
1448     switch (rp_status) {
1449     case LTTNG_VIEWER_GET_PACKET_OK:
1450         req_len = be32toh(rp.len);
1451         BT_CPPLOGD_SPEC(viewer_connection->logger,
1452                         "Got packet from relay daemon: response={}, packet-len={}",
1453                         static_cast<lttng_viewer_get_packet_return_code>(rp_status), req_len);
1454         break;
1455     case LTTNG_VIEWER_GET_PACKET_RETRY:
1456         /* Unimplemented by relay daemon */
1457         status = CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
1458         goto end;
1459     case LTTNG_VIEWER_GET_PACKET_ERR:
1460         if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1461             BT_CPPLOGD_SPEC(viewer_connection->logger,
1462                             "Marking trace as needing new metadata: "
1463                             "response={}, response-flag=NEW_METADATA, trace-id={}",
1464                             static_cast<lttng_viewer_get_packet_return_code>(rp_status), trace->id);
1465             trace->metadata_stream_state = LTTNG_LIVE_METADATA_STREAM_STATE_NEEDED;
1466         }
1467         if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1468             BT_CPPLOGD_SPEC(viewer_connection->logger,
1469                             "Marking all sessions as possibly needing new streams: "
1470                             "response={}, response-flag=NEW_STREAM",
1471                             static_cast<lttng_viewer_get_packet_return_code>(rp_status));
1472             lttng_live_need_new_streams(lttng_live_msg_iter);
1473         }
1474         if (flags & (LTTNG_VIEWER_FLAG_NEW_METADATA | LTTNG_VIEWER_FLAG_NEW_STREAM)) {
1475             status = CTF_MSG_ITER_MEDIUM_STATUS_AGAIN;
1476             BT_CPPLOGD_SPEC(viewer_connection->logger,
1477                             "Reply with any one flags set means we should retry: response={}",
1478                             static_cast<lttng_viewer_get_packet_return_code>(rp_status));
1479             goto end;
1480         }
1481         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
1482                                      "Received get_data_packet response: error");
1483         status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
1484         goto end;
1485     case LTTNG_VIEWER_GET_PACKET_EOF:
1486         status = CTF_MSG_ITER_MEDIUM_STATUS_EOF;
1487         goto end;
1488     default:
1489         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
1490                                      "Received get_data_packet response: unknown ({})", rp_status);
1491         status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
1492         goto end;
1493     }
1494
1495     if (req_len == 0) {
1496         status = CTF_MSG_ITER_MEDIUM_STATUS_ERROR;
1497         goto end;
1498     }
1499
1500     viewer_status = lttng_live_recv(viewer_connection, buf, req_len);
1501     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1502         viewer_handle_recv_status(viewer_status, "get data packet");
1503         goto error_convert_status;
1504     }
1505     *recv_len = req_len;
1506
1507     status = CTF_MSG_ITER_MEDIUM_STATUS_OK;
1508     goto end;
1509
1510 error_convert_status:
1511     status = viewer_status_to_ctf_msg_iter_medium_status(viewer_status);
1512 end:
1513     return status;
1514 }
1515
1516 /*
1517  * Request new streams for a session.
1518  */
1519 enum lttng_live_iterator_status
1520 lttng_live_session_get_new_streams(struct lttng_live_session *session,
1521                                    bt_self_message_iterator *self_msg_iter)
1522 {
1523     enum lttng_live_iterator_status status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1524     struct lttng_viewer_cmd cmd;
1525     struct lttng_viewer_new_streams_request rq;
1526     struct lttng_viewer_new_streams_response rp;
1527     struct lttng_live_msg_iter *lttng_live_msg_iter = session->lttng_live_msg_iter;
1528     enum lttng_live_viewer_status viewer_status;
1529     struct live_viewer_connection *viewer_connection = lttng_live_msg_iter->viewer_connection;
1530     uint32_t streams_count;
1531     const size_t cmd_buf_len = sizeof(cmd) + sizeof(rq);
1532     char cmd_buf[cmd_buf_len];
1533
1534     if (!session->new_streams_needed) {
1535         status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1536         goto end;
1537     }
1538
1539     BT_CPPLOGD_SPEC(viewer_connection->logger,
1540                     "Requesting new streams for session: cmd={}, session-id={}",
1541                     LTTNG_VIEWER_GET_NEW_STREAMS, session->id);
1542
1543     cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
1544     cmd.data_size = htobe64((uint64_t) sizeof(rq));
1545     cmd.cmd_version = htobe32(0);
1546
1547     memset(&rq, 0, sizeof(rq));
1548     rq.session_id = htobe64(session->id);
1549
1550     /*
1551      * Merge the cmd and connection request to prevent a write-write
1552      * sequence on the TCP socket. Otherwise, a delayed ACK will prevent the
1553      * second write to be performed quickly in presence of Nagle's algorithm.
1554      */
1555     memcpy(cmd_buf, &cmd, sizeof(cmd));
1556     memcpy(cmd_buf + sizeof(cmd), &rq, sizeof(rq));
1557
1558     viewer_status = lttng_live_send(viewer_connection, &cmd_buf, cmd_buf_len);
1559     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1560         viewer_handle_send_status(viewer_status, "get new streams command");
1561         status = viewer_status_to_live_iterator_status(viewer_status);
1562         goto end;
1563     }
1564
1565     viewer_status = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1566     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1567         viewer_handle_recv_status(viewer_status, "get new streams reply");
1568         status = viewer_status_to_live_iterator_status(viewer_status);
1569         goto end;
1570     }
1571
1572     streams_count = be32toh(rp.streams_count);
1573
1574     switch (be32toh(rp.status)) {
1575     case LTTNG_VIEWER_NEW_STREAMS_OK:
1576         session->new_streams_needed = false;
1577         break;
1578     case LTTNG_VIEWER_NEW_STREAMS_NO_NEW:
1579         session->new_streams_needed = false;
1580         goto end;
1581     case LTTNG_VIEWER_NEW_STREAMS_HUP:
1582         session->new_streams_needed = false;
1583         session->closed = true;
1584         status = LTTNG_LIVE_ITERATOR_STATUS_END;
1585         goto end;
1586     case LTTNG_VIEWER_NEW_STREAMS_ERR:
1587         BT_CPPLOGD_SPEC(viewer_connection->logger, "Received get_new_streams response: error");
1588         status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1589         goto end;
1590     default:
1591         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
1592                                      "Received get_new_streams response: Unknown:"
1593                                      "return code {}",
1594                                      be32toh(rp.status));
1595         status = LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1596         goto end;
1597     }
1598
1599     viewer_status = receive_streams(session, streams_count, self_msg_iter);
1600     if (viewer_status != LTTNG_LIVE_VIEWER_STATUS_OK) {
1601         viewer_handle_recv_status(viewer_status, "new streams");
1602         status = viewer_status_to_live_iterator_status(viewer_status);
1603         goto end;
1604     }
1605
1606     status = LTTNG_LIVE_ITERATOR_STATUS_OK;
1607 end:
1608     return status;
1609 }
1610
1611 enum lttng_live_viewer_status live_viewer_connection_create(
1612     const char *url, bool in_query, struct lttng_live_msg_iter *lttng_live_msg_iter,
1613     const bt2c::Logger& parentLogger, struct live_viewer_connection **viewer)
1614 {
1615     enum lttng_live_viewer_status status;
1616
1617     live_viewer_connection *viewer_connection = new live_viewer_connection {parentLogger};
1618
1619     if (bt_socket_init(viewer_connection->logger) != 0) {
1620         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Failed to init socket");
1621         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1622         goto error;
1623     }
1624
1625     viewer_connection->control_sock = BT_INVALID_SOCKET;
1626     viewer_connection->port = -1;
1627     viewer_connection->in_query = in_query;
1628     viewer_connection->lttng_live_msg_iter = lttng_live_msg_iter;
1629     viewer_connection->url = g_string_new(url);
1630     if (!viewer_connection->url) {
1631         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger, "Failed to allocate URL buffer");
1632         status = LTTNG_LIVE_VIEWER_STATUS_ERROR;
1633         goto error;
1634     }
1635
1636     BT_CPPLOGD_SPEC(viewer_connection->logger, "Establishing connection to url \"{}\"...", url);
1637     status = lttng_live_connect_viewer(viewer_connection);
1638     /*
1639      * Only print error and append cause in case of error. not in case of
1640      * interruption.
1641      */
1642     if (status == LTTNG_LIVE_VIEWER_STATUS_ERROR) {
1643         BT_CPPLOGE_APPEND_CAUSE_SPEC(viewer_connection->logger,
1644                                      "Failed to establish connection: "
1645                                      "url=\"{}\"",
1646                                      url);
1647         goto error;
1648     } else if (status == LTTNG_LIVE_VIEWER_STATUS_INTERRUPTED) {
1649         goto error;
1650     }
1651     BT_CPPLOGD_SPEC(viewer_connection->logger, "Connection to url \"{}\" is established", url);
1652
1653     *viewer = viewer_connection;
1654     status = LTTNG_LIVE_VIEWER_STATUS_OK;
1655     goto end;
1656
1657 error:
1658     if (viewer_connection) {
1659         live_viewer_connection_destroy(viewer_connection);
1660     }
1661 end:
1662     return status;
1663 }
1664
1665 void live_viewer_connection_destroy(struct live_viewer_connection *viewer_connection)
1666 {
1667     if (!viewer_connection) {
1668         goto end;
1669     }
1670
1671     BT_CPPLOGD_SPEC(viewer_connection->logger, "Closing connection to relay: relay-url=\"{}\"",
1672                     viewer_connection->url->str);
1673
1674     lttng_live_disconnect_viewer(viewer_connection);
1675
1676     if (viewer_connection->url) {
1677         g_string_free(viewer_connection->url, true);
1678     }
1679
1680     if (viewer_connection->relay_hostname) {
1681         g_string_free(viewer_connection->relay_hostname, true);
1682     }
1683
1684     if (viewer_connection->target_hostname) {
1685         g_string_free(viewer_connection->target_hostname, true);
1686     }
1687
1688     if (viewer_connection->session_name) {
1689         g_string_free(viewer_connection->session_name, true);
1690     }
1691
1692     if (viewer_connection->proto) {
1693         g_string_free(viewer_connection->proto, true);
1694     }
1695
1696     delete viewer_connection;
1697
1698     bt_socket_fini();
1699
1700 end:
1701     return;
1702 }