plugins/ctf/lttng-live/viewer-connection.c

   1 /*
   2  * Copyright 2016 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   3  *
   4  * Permission is hereby granted, free of charge, to any person obtaining a copy
   5  * of this software and associated documentation files (the "Software"), to deal
   6  * in the Software without restriction, including without limitation the rights
   7  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   8  * copies of the Software, and to permit persons to whom the Software is
   9  * furnished to do so, subject to the following conditions:
  10  *
  11  * The above copyright notice and this permission notice shall be included in
  12  * all copies or substantial portions of the Software.
  13  *
  14  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  15  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  16  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  17  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  18  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  19  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  20  * SOFTWARE.
  21  */
  22
  23 #define BT_LOG_TAG "PLUGIN-CTF-LTTNG-LIVE-SRC-VIEWER"
  24 #include "logging.h"
  25
  26 #include <stdio.h>
  27 #include <stdint.h>
  28 #include <stdlib.h>
  29 #include <stdbool.h>
  30 #include <unistd.h>
  31 #include <glib.h>
  32 #include <inttypes.h>
  33 #include <sys/types.h>
  34 #include <fcntl.h>
  35
  36 #include <babeltrace/compat/socket-internal.h>
  37 #include <babeltrace/endian-internal.h>
  38 #include <babeltrace/compiler-internal.h>
  39 #include <babeltrace/common-internal.h>
  40 #include <babeltrace/babeltrace.h>
  41
  42 #include "lttng-live-internal.h"
  43 #include "viewer-connection.h"
  44 #include "lttng-viewer-abi.h"
  45 #include "data-stream.h"
  46 #include "metadata.h"
  47
  48 static ssize_t lttng_live_recv(struct bt_live_viewer_connection *viewer_connection,
  49                 void *buf, size_t len)
  50 {
  51         ssize_t ret;
  52         size_t copied = 0, to_copy = len;
  53         struct lttng_live_component *lttng_live =
  54                         viewer_connection->lttng_live;
  55         BT_SOCKET sock = viewer_connection->control_sock;
  56
  57         do {
  58                 ret = bt_socket_recv(sock, buf + copied, to_copy, 0);
  59                 if (ret > 0) {
  60                         BT_ASSERT(ret <= to_copy);
  61                         copied += ret;
  62                         to_copy -= ret;
  63                 }
  64                 if (ret == BT_SOCKET_ERROR && bt_socket_interrupted()) {
  65                         if (lttng_live_is_canceled(lttng_live)) {
  66                                 break;
  67                         } else {
  68                                 continue;
  69                         }
  70                 }
  71         } while (ret > 0 && to_copy > 0);
  72         if (ret > 0)
  73                 ret = copied;
  74         /* ret = 0 means orderly shutdown, ret == BT_SOCKET_ERROR is error. */
  75         return ret;
  76 }
  77
  78 static ssize_t lttng_live_send(struct bt_live_viewer_connection *viewer_connection,
  79                 const void *buf, size_t len)
  80 {
  81         struct lttng_live_component *lttng_live =
  82                         viewer_connection->lttng_live;
  83         BT_SOCKET sock = viewer_connection->control_sock;
  84         ssize_t ret;
  85
  86         for (;;) {
  87                 ret = bt_socket_send_nosigpipe(sock, buf, len);
  88                 if (ret == BT_SOCKET_ERROR && bt_socket_interrupted()) {
  89                         if (lttng_live_is_canceled(lttng_live)) {
  90                                 break;
  91                         } else {
  92                                 continue;
  93                         }
  94                 } else {
  95                         break;
  96                 }
  97         }
  98         return ret;
  99 }
 100
 101 static int parse_url(struct bt_live_viewer_connection *viewer_connection)
 102 {
 103         char error_buf[256] = { 0 };
 104         struct bt_common_lttng_live_url_parts lttng_live_url_parts = { 0 };
 105         int ret = -1;
 106         const char *path = viewer_connection->url->str;
 107
 108         if (!path) {
 109                 goto end;
 110         }
 111
 112         lttng_live_url_parts = bt_common_parse_lttng_live_url(path,
 113                         error_buf, sizeof(error_buf));
 114         if (!lttng_live_url_parts.proto) {
 115                 BT_LOGW("Invalid LTTng live URL format: %s", error_buf);
 116                 goto end;
 117         }
 118
 119         viewer_connection->relay_hostname =
 120                         lttng_live_url_parts.hostname;
 121         lttng_live_url_parts.hostname = NULL;
 122
 123         if (lttng_live_url_parts.port >= 0) {
 124                 viewer_connection->port = lttng_live_url_parts.port;
 125         } else {
 126                 viewer_connection->port = LTTNG_DEFAULT_NETWORK_VIEWER_PORT;
 127         }
 128
 129         viewer_connection->target_hostname =
 130                         lttng_live_url_parts.target_hostname;
 131         lttng_live_url_parts.target_hostname = NULL;
 132
 133         if (lttng_live_url_parts.session_name) {
 134                 viewer_connection->session_name =
 135                                 lttng_live_url_parts.session_name;
 136                 lttng_live_url_parts.session_name = NULL;
 137         }
 138
 139         BT_LOGD("Connecting to hostname : %s, port : %d, "
 140                         "target hostname : %s, session name : %s, "
 141                         "proto : %s",
 142                         viewer_connection->relay_hostname->str,
 143                         viewer_connection->port,
 144                         viewer_connection->target_hostname == NULL ?
 145                                 "<none>" : viewer_connection->target_hostname->str,
 146                         viewer_connection->session_name == NULL ?
 147                                 "<none>" : viewer_connection->session_name->str,
 148                         lttng_live_url_parts.proto->str);
 149         ret = 0;
 150
 151 end:
 152         bt_common_destroy_lttng_live_url_parts(&lttng_live_url_parts);
 153         return ret;
 154 }
 155
 156 static int lttng_live_handshake(struct bt_live_viewer_connection *viewer_connection)
 157 {
 158         struct lttng_viewer_cmd cmd;
 159         struct lttng_viewer_connect connect;
 160         int ret;
 161         ssize_t ret_len;
 162
 163         cmd.cmd = htobe32(LTTNG_VIEWER_CONNECT);
 164         cmd.data_size = htobe64((uint64_t) sizeof(connect));
 165         cmd.cmd_version = htobe32(0);
 166
 167         connect.viewer_session_id = -1ULL;      /* will be set on recv */
 168         connect.major = htobe32(LTTNG_LIVE_MAJOR);
 169         connect.minor = htobe32(LTTNG_LIVE_MINOR);
 170         connect.type = htobe32(LTTNG_VIEWER_CLIENT_COMMAND);
 171
 172         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 173         if (ret_len == BT_SOCKET_ERROR) {
 174                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 175                 goto error;
 176         }
 177         BT_ASSERT(ret_len == sizeof(cmd));
 178
 179         ret_len = lttng_live_send(viewer_connection, &connect, sizeof(connect));
 180         if (ret_len == BT_SOCKET_ERROR) {
 181                 BT_LOGE("Error sending version: %s", bt_socket_errormsg());
 182                 goto error;
 183         }
 184         BT_ASSERT(ret_len == sizeof(connect));
 185
 186         ret_len = lttng_live_recv(viewer_connection, &connect, sizeof(connect));
 187         if (ret_len == 0) {
 188                 BT_LOGI("Remote side has closed connection");
 189                 goto error;
 190         }
 191         if (ret_len == BT_SOCKET_ERROR) {
 192                 BT_LOGE("Error receiving version: %s", bt_socket_errormsg());
 193                 goto error;
 194         }
 195         BT_ASSERT(ret_len == sizeof(connect));
 196
 197         BT_LOGD("Received viewer session ID : %" PRIu64,
 198                         (uint64_t) be64toh(connect.viewer_session_id));
 199         BT_LOGD("Relayd version : %u.%u", be32toh(connect.major),
 200                         be32toh(connect.minor));
 201
 202         if (LTTNG_LIVE_MAJOR != be32toh(connect.major)) {
 203                 BT_LOGE("Incompatible lttng-relayd protocol");
 204                 goto error;
 205         }
 206         /* Use the smallest protocol version implemented. */
 207         if (LTTNG_LIVE_MINOR > be32toh(connect.minor)) {
 208                 viewer_connection->minor =  be32toh(connect.minor);
 209         } else {
 210                 viewer_connection->minor =  LTTNG_LIVE_MINOR;
 211         }
 212         viewer_connection->major = LTTNG_LIVE_MAJOR;
 213         ret = 0;
 214         return ret;
 215
 216 error:
 217         BT_LOGE("Unable to establish connection");
 218         return -1;
 219 }
 220
 221 static int lttng_live_connect_viewer(struct bt_live_viewer_connection *viewer_connection)
 222 {
 223         struct hostent *host;
 224         struct sockaddr_in server_addr;
 225         int ret;
 226
 227         if (parse_url(viewer_connection)) {
 228                 goto error;
 229         }
 230
 231         host = gethostbyname(viewer_connection->relay_hostname->str);
 232         if (!host) {
 233                 BT_LOGE("Cannot lookup hostname %s",
 234                         viewer_connection->relay_hostname->str);
 235                 goto error;
 236         }
 237
 238         if ((viewer_connection->control_sock = socket(AF_INET, SOCK_STREAM, 0)) == BT_INVALID_SOCKET) {
 239                 BT_LOGE("Socket creation failed: %s", bt_socket_errormsg());
 240                 goto error;
 241         }
 242
 243         server_addr.sin_family = AF_INET;
 244         server_addr.sin_port = htons(viewer_connection->port);
 245         server_addr.sin_addr = *((struct in_addr *) host->h_addr);
 246         memset(&(server_addr.sin_zero), 0, 8);
 247
 248         if (connect(viewer_connection->control_sock, (struct sockaddr *) &server_addr,
 249                                 sizeof(struct sockaddr)) == BT_SOCKET_ERROR) {
 250                 BT_LOGE("Connection failed: %s", bt_socket_errormsg());
 251                 goto error;
 252         }
 253         if (lttng_live_handshake(viewer_connection)) {
 254                 goto error;
 255         }
 256
 257         ret = 0;
 258
 259         return ret;
 260
 261 error:
 262         if (viewer_connection->control_sock != BT_INVALID_SOCKET) {
 263                 if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
 264                         BT_LOGE("Close: %s", bt_socket_errormsg());
 265                 }
 266         }
 267         viewer_connection->control_sock = BT_INVALID_SOCKET;
 268         return -1;
 269 }
 270
 271 static void lttng_live_disconnect_viewer(struct bt_live_viewer_connection *viewer_connection)
 272 {
 273         if (viewer_connection->control_sock == BT_INVALID_SOCKET) {
 274                 return;
 275         }
 276         if (bt_socket_close(viewer_connection->control_sock) == BT_SOCKET_ERROR) {
 277                 BT_LOGE("Close: %s", bt_socket_errormsg());
 278                 viewer_connection->control_sock = BT_INVALID_SOCKET;
 279         }
 280 }
 281
 282 static void connection_release(bt_object *obj)
 283 {
 284         struct bt_live_viewer_connection *conn =
 285                 container_of(obj, struct bt_live_viewer_connection, obj);
 286
 287         bt_live_viewer_connection_destroy(conn);
 288 }
 289
 290 static
 291 enum bt_value_status list_update_session(bt_value *results,
 292                 const struct lttng_viewer_session *session,
 293                 bool *_found)
 294 {
 295         enum bt_value_status ret = BT_VALUE_STATUS_OK;
 296         bt_value *map = NULL;
 297         bt_value *hostname = NULL;
 298         bt_value *session_name = NULL;
 299         bt_value *btval = NULL;
 300         int i, len;
 301         bool found = false;
 302
 303         len = bt_value_array_get_size(results);
 304         if (len < 0) {
 305                 ret = BT_VALUE_STATUS_ERROR;
 306                 goto end;
 307         }
 308         for (i = 0; i < len; i++) {
 309                 const char *hostname_str = NULL;
 310                 const char *session_name_str = NULL;
 311
 312                 map = bt_value_array_get(results, (size_t) i);
 313                 if (!map) {
 314                         ret = BT_VALUE_STATUS_ERROR;
 315                         goto end;
 316                 }
 317                 hostname = bt_value_map_get(map, "target-hostname");
 318                 if (!hostname) {
 319                         ret = BT_VALUE_STATUS_ERROR;
 320                         goto end;
 321                 }
 322                 session_name = bt_value_map_get(map, "session-name");
 323                 if (!session_name) {
 324                         ret = BT_VALUE_STATUS_ERROR;
 325                         goto end;
 326                 }
 327                 hostname_str = bt_value_string_get(hostname);
 328                 session_name_str = bt_value_string_get(session_name);
 329
 330                 if (!strcmp(session->hostname, hostname_str)
 331                                 && !strcmp(session->session_name,
 332                                         session_name_str)) {
 333                         int64_t val;
 334                         uint32_t streams = be32toh(session->streams);
 335                         uint32_t clients = be32toh(session->clients);
 336
 337                         found = true;
 338
 339                         btval = bt_value_map_get(map, "stream-count");
 340                         if (!btval) {
 341                                 ret = BT_VALUE_STATUS_ERROR;
 342                                 goto end;
 343                         }
 344                         val = bt_value_integer_get(btval);
 345                         /* sum */
 346                         val += streams;
 347                         ret = bt_private_integer_bool_set(btval, val);
 348                         if (ret != BT_VALUE_STATUS_OK) {
 349                                 goto end;
 350                         }
 351                         BT_VALUE_PUT_REF_AND_RESET(btval);
 352
 353                         btval = bt_value_map_get(map, "client-count");
 354                         if (!btval) {
 355                                 ret = BT_VALUE_STATUS_ERROR;
 356                                 goto end;
 357                         }
 358                         val = bt_value_integer_get(btval);
 359                         /* max */
 360                         val = max_t(int64_t, clients, val);
 361                         ret = bt_private_integer_bool_set(btval, val);
 362                         if (ret != BT_VALUE_STATUS_OK) {
 363                                 goto end;
 364                         }
 365                         BT_VALUE_PUT_REF_AND_RESET(btval);
 366                 }
 367
 368                 BT_VALUE_PUT_REF_AND_RESET(hostname);
 369                 BT_VALUE_PUT_REF_AND_RESET(session_name);
 370                 BT_VALUE_PUT_REF_AND_RESET(map);
 371
 372                 if (found) {
 373                         break;
 374                 }
 375         }
 376 end:
 377         BT_VALUE_PUT_REF_AND_RESET(btval);
 378         BT_VALUE_PUT_REF_AND_RESET(hostname);
 379         BT_VALUE_PUT_REF_AND_RESET(session_name);
 380         BT_VALUE_PUT_REF_AND_RESET(map);
 381         *_found = found;
 382         return ret;
 383 }
 384
 385 static
 386 enum bt_value_status list_append_session(bt_value *results,
 387                 GString *base_url,
 388                 const struct lttng_viewer_session *session)
 389 {
 390         enum bt_value_status ret = BT_VALUE_STATUS_OK;
 391         bt_value *map = NULL;
 392         GString *url = NULL;
 393         bool found = false;
 394
 395         /*
 396          * If the session already exists, add the stream count to it,
 397          * and do max of client counts.
 398          */
 399         ret = list_update_session(results, session, &found);
 400         if (ret != BT_VALUE_STATUS_OK || found) {
 401                 goto end;
 402         }
 403
 404         map = bt_private_value_map_create();
 405         if (!map) {
 406                 ret = BT_VALUE_STATUS_ERROR;
 407                 goto end;
 408         }
 409
 410         if (base_url->len < 1) {
 411                 ret = BT_VALUE_STATUS_ERROR;
 412                 goto end;
 413         }
 414         /*
 415          * key = "url",
 416          * value = <string>,
 417          */
 418         url = g_string_new(base_url->str);
 419         g_string_append(url, "/host/");
 420         g_string_append(url, session->hostname);
 421         g_string_append_c(url, '/');
 422         g_string_append(url, session->session_name);
 423
 424         ret = bt_private_value_map_insert_string_entry(map, "url", url->str);
 425         if (ret != BT_VALUE_STATUS_OK) {
 426                 goto end;
 427         }
 428
 429         /*
 430          * key = "target-hostname",
 431          * value = <string>,
 432          */
 433         ret = bt_private_value_map_insert_string_entry(map, "target-hostname",
 434                 session->hostname);
 435         if (ret != BT_VALUE_STATUS_OK) {
 436                 goto end;
 437         }
 438
 439         /*
 440          * key = "session-name",
 441          * value = <string>,
 442          */
 443         ret = bt_private_value_map_insert_string_entry(map, "session-name",
 444                 session->session_name);
 445         if (ret != BT_VALUE_STATUS_OK) {
 446                 goto end;
 447         }
 448
 449         /*
 450          * key = "timer-us",
 451          * value = <integer>,
 452          */
 453         {
 454                 uint32_t live_timer = be32toh(session->live_timer);
 455
 456                 ret = bt_private_value_map_insert_integer_entry(map, "timer-us",
 457                         live_timer);
 458                 if (ret != BT_VALUE_STATUS_OK) {
 459                         goto end;
 460                 }
 461         }
 462
 463         /*
 464          * key = "stream-count",
 465          * value = <integer>,
 466          */
 467         {
 468                 uint32_t streams = be32toh(session->streams);
 469
 470                 ret = bt_private_value_map_insert_integer_entry(map, "stream-count",
 471                         streams);
 472                 if (ret != BT_VALUE_STATUS_OK) {
 473                         goto end;
 474                 }
 475         }
 476
 477
 478         /*
 479          * key = "client-count",
 480          * value = <integer>,
 481          */
 482         {
 483                 uint32_t clients = be32toh(session->clients);
 484
 485                 ret = bt_private_value_map_insert_integer_entry(map, "client-count",
 486                         clients);
 487                 if (ret != BT_VALUE_STATUS_OK) {
 488                         goto end;
 489                 }
 490         }
 491
 492         ret = bt_private_value_array_append_element(results, map);
 493 end:
 494         if (url) {
 495                 g_string_free(url, TRUE);
 496         }
 497         BT_VALUE_PUT_REF_AND_RESET(map);
 498         return ret;
 499 }
 500
 501 /*
 502  * Data structure returned:
 503  *
 504  * {
 505  *   <array> = {
 506  *     [n] = {
 507  *       <map> = {
 508  *         {
 509  *           key = "url",
 510  *           value = <string>,
 511  *         },
 512  *         {
 513  *           key = "target-hostname",
 514  *           value = <string>,
 515  *         },
 516  *         {
 517  *           key = "session-name",
 518  *           value = <string>,
 519  *         },
 520  *         {
 521  *           key = "timer-us",
 522  *           value = <integer>,
 523  *         },
 524  *         {
 525  *           key = "stream-count",
 526  *           value = <integer>,
 527  *         },
 528  *         {
 529  *           key = "client-count",
 530  *           value = <integer>,
 531  *         },
 532  *       },
 533  *     }
 534  *   }
 535  */
 536
 537 BT_HIDDEN
 538 bt_value *bt_live_viewer_connection_list_sessions(struct bt_live_viewer_connection *viewer_connection)
 539 {
 540         bt_value *results = NULL;
 541         struct lttng_viewer_cmd cmd;
 542         struct lttng_viewer_list_sessions list;
 543         uint32_t i, sessions_count;
 544         ssize_t ret_len;
 545
 546         if (lttng_live_handshake(viewer_connection)) {
 547                 goto error;
 548         }
 549
 550         results = bt_private_value_array_create();
 551         if (!results) {
 552                 BT_LOGE("Error creating array");
 553                 goto error;
 554         }
 555
 556         cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 557         cmd.data_size = htobe64((uint64_t) 0);
 558         cmd.cmd_version = htobe32(0);
 559
 560         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 561         if (ret_len == BT_SOCKET_ERROR) {
 562                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 563                 goto error;
 564         }
 565         BT_ASSERT(ret_len == sizeof(cmd));
 566
 567         ret_len = lttng_live_recv(viewer_connection, &list, sizeof(list));
 568         if (ret_len == 0) {
 569                 BT_LOGI("Remote side has closed connection");
 570                 goto error;
 571         }
 572         if (ret_len == BT_SOCKET_ERROR) {
 573                 BT_LOGE("Error receiving session list: %s", bt_socket_errormsg());
 574                 goto error;
 575         }
 576         BT_ASSERT(ret_len == sizeof(list));
 577
 578         sessions_count = be32toh(list.sessions_count);
 579         for (i = 0; i < sessions_count; i++) {
 580                 struct lttng_viewer_session lsession;
 581
 582                 ret_len = lttng_live_recv(viewer_connection,
 583                                 &lsession, sizeof(lsession));
 584                 if (ret_len == 0) {
 585                         BT_LOGI("Remote side has closed connection");
 586                         goto error;
 587                 }
 588                 if (ret_len == BT_SOCKET_ERROR) {
 589                         BT_LOGE("Error receiving session: %s", bt_socket_errormsg());
 590                         goto error;
 591                 }
 592                 BT_ASSERT(ret_len == sizeof(lsession));
 593                 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
 594                 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 595                 if (list_append_session(results,
 596                                 viewer_connection->url, &lsession)
 597                                 != BT_VALUE_STATUS_OK) {
 598                         goto error;
 599                 }
 600         }
 601         goto end;
 602 error:
 603         BT_VALUE_PUT_REF_AND_RESET(results);
 604 end:
 605         return results;
 606 }
 607
 608 static
 609 int lttng_live_query_session_ids(struct lttng_live_component *lttng_live)
 610 {
 611         struct lttng_viewer_cmd cmd;
 612         struct lttng_viewer_list_sessions list;
 613         struct lttng_viewer_session lsession;
 614         uint32_t i, sessions_count;
 615         ssize_t ret_len;
 616         uint64_t session_id;
 617         struct bt_live_viewer_connection *viewer_connection =
 618                         lttng_live->viewer_connection;
 619
 620         cmd.cmd = htobe32(LTTNG_VIEWER_LIST_SESSIONS);
 621         cmd.data_size = htobe64((uint64_t) 0);
 622         cmd.cmd_version = htobe32(0);
 623
 624         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 625         if (ret_len == BT_SOCKET_ERROR) {
 626                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 627                 goto error;
 628         }
 629         BT_ASSERT(ret_len == sizeof(cmd));
 630
 631         ret_len = lttng_live_recv(viewer_connection, &list, sizeof(list));
 632         if (ret_len == 0) {
 633                 BT_LOGI("Remote side has closed connection");
 634                 goto error;
 635         }
 636         if (ret_len == BT_SOCKET_ERROR) {
 637                 BT_LOGE("Error receiving session list: %s", bt_socket_errormsg());
 638                 goto error;
 639         }
 640         BT_ASSERT(ret_len == sizeof(list));
 641
 642         sessions_count = be32toh(list.sessions_count);
 643         for (i = 0; i < sessions_count; i++) {
 644                 ret_len = lttng_live_recv(viewer_connection,
 645                                 &lsession, sizeof(lsession));
 646                 if (ret_len == 0) {
 647                         BT_LOGI("Remote side has closed connection");
 648                         goto error;
 649                 }
 650                 if (ret_len == BT_SOCKET_ERROR) {
 651                         BT_LOGE("Error receiving session: %s", bt_socket_errormsg());
 652                         goto error;
 653                 }
 654                 BT_ASSERT(ret_len == sizeof(lsession));
 655                 lsession.hostname[LTTNG_VIEWER_HOST_NAME_MAX - 1] = '\0';
 656                 lsession.session_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 657                 session_id = be64toh(lsession.id);
 658
 659                 BT_LOGD("Adding session %" PRIu64 " hostname: %s session_name: %s",
 660                         session_id, lsession.hostname, lsession.session_name);
 661
 662                 if ((strncmp(lsession.session_name,
 663                         viewer_connection->session_name->str,
 664                         LTTNG_VIEWER_NAME_MAX) == 0) && (strncmp(lsession.hostname,
 665                                 viewer_connection->target_hostname->str,
 666                                 LTTNG_VIEWER_HOST_NAME_MAX) == 0)) {
 667                         if (lttng_live_add_session(lttng_live, session_id,
 668                                         lsession.hostname,
 669                                         lsession.session_name)) {
 670                                 goto error;
 671                         }
 672                 }
 673         }
 674
 675         return 0;
 676
 677 error:
 678         BT_LOGE("Unable to query session ids");
 679         return -1;
 680 }
 681
 682 BT_HIDDEN
 683 int lttng_live_create_viewer_session(struct lttng_live_component *lttng_live)
 684 {
 685         struct lttng_viewer_cmd cmd;
 686         struct lttng_viewer_create_session_response resp;
 687         ssize_t ret_len;
 688         struct bt_live_viewer_connection *viewer_connection =
 689                         lttng_live->viewer_connection;
 690
 691         cmd.cmd = htobe32(LTTNG_VIEWER_CREATE_SESSION);
 692         cmd.data_size = htobe64((uint64_t) 0);
 693         cmd.cmd_version = htobe32(0);
 694
 695         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 696         if (ret_len == BT_SOCKET_ERROR) {
 697                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 698                 goto error;
 699         }
 700         BT_ASSERT(ret_len == sizeof(cmd));
 701
 702         ret_len = lttng_live_recv(viewer_connection, &resp, sizeof(resp));
 703         if (ret_len == 0) {
 704                 BT_LOGI("Remote side has closed connection");
 705                 goto error;
 706         }
 707         if (ret_len == BT_SOCKET_ERROR) {
 708                 BT_LOGE("Error receiving create session reply: %s", bt_socket_errormsg());
 709                 goto error;
 710         }
 711         BT_ASSERT(ret_len == sizeof(resp));
 712
 713         if (be32toh(resp.status) != LTTNG_VIEWER_CREATE_SESSION_OK) {
 714                 BT_LOGE("Error creating viewer session");
 715                 goto error;
 716         }
 717         if (lttng_live_query_session_ids(lttng_live)) {
 718                 goto error;
 719         }
 720
 721         return 0;
 722
 723 error:
 724         return -1;
 725 }
 726
 727 static
 728 int receive_streams(struct lttng_live_session *session,
 729                 uint32_t stream_count)
 730 {
 731         ssize_t ret_len;
 732         uint32_t i;
 733         struct lttng_live_component *lttng_live = session->lttng_live;
 734         struct bt_live_viewer_connection *viewer_connection =
 735                         lttng_live->viewer_connection;
 736
 737         BT_LOGD("Getting %" PRIu32 " new streams:", stream_count);
 738         for (i = 0; i < stream_count; i++) {
 739                 struct lttng_viewer_stream stream;
 740                 struct lttng_live_stream_iterator *live_stream;
 741                 uint64_t stream_id;
 742                 uint64_t ctf_trace_id;
 743
 744                 ret_len = lttng_live_recv(viewer_connection, &stream, sizeof(stream));
 745                 if (ret_len == 0) {
 746                         BT_LOGI("Remote side has closed connection");
 747                         goto error;
 748                 }
 749                 if (ret_len == BT_SOCKET_ERROR) {
 750                         BT_LOGE("Error receiving stream");
 751                         goto error;
 752                 }
 753                 BT_ASSERT(ret_len == sizeof(stream));
 754                 stream.path_name[LTTNG_VIEWER_PATH_MAX - 1] = '\0';
 755                 stream.channel_name[LTTNG_VIEWER_NAME_MAX - 1] = '\0';
 756                 stream_id = be64toh(stream.id);
 757                 ctf_trace_id = be64toh(stream.ctf_trace_id);
 758
 759                 if (stream.metadata_flag) {
 760                         BT_LOGD("    metadata stream %" PRIu64 " : %s/%s",
 761                                         stream_id, stream.path_name,
 762                                         stream.channel_name);
 763                         if (lttng_live_metadata_create_stream(session,
 764                                         ctf_trace_id, stream_id,
 765                                         stream.path_name)) {
 766                                 BT_LOGE("Error creating metadata stream");
 767
 768                                 goto error;
 769                         }
 770                         session->lazy_stream_msg_init = true;
 771                 } else {
 772                         BT_LOGD("    stream %" PRIu64 " : %s/%s",
 773                                         stream_id, stream.path_name,
 774                                         stream.channel_name);
 775                         live_stream = lttng_live_stream_iterator_create(session,
 776                                 ctf_trace_id, stream_id);
 777                         if (!live_stream) {
 778                                 BT_LOGE("Error creating streamn");
 779                                 goto error;
 780                         }
 781                 }
 782         }
 783         return 0;
 784
 785 error:
 786         return -1;
 787 }
 788
 789 BT_HIDDEN
 790 int lttng_live_attach_session(struct lttng_live_session *session)
 791 {
 792         struct lttng_viewer_cmd cmd;
 793         struct lttng_viewer_attach_session_request rq;
 794         struct lttng_viewer_attach_session_response rp;
 795         ssize_t ret_len;
 796         struct lttng_live_component *lttng_live = session->lttng_live;
 797         struct bt_live_viewer_connection *viewer_connection =
 798                         lttng_live->viewer_connection;
 799         uint64_t session_id = session->id;
 800         uint32_t streams_count;
 801
 802         if (session->attached) {
 803                 return 0;
 804         }
 805
 806         cmd.cmd = htobe32(LTTNG_VIEWER_ATTACH_SESSION);
 807         cmd.data_size = htobe64((uint64_t) sizeof(rq));
 808         cmd.cmd_version = htobe32(0);
 809
 810         memset(&rq, 0, sizeof(rq));
 811         rq.session_id = htobe64(session_id);
 812         // TODO: add cmd line parameter to select seek beginning
 813         // rq.seek = htobe32(LTTNG_VIEWER_SEEK_BEGINNING);
 814         rq.seek = htobe32(LTTNG_VIEWER_SEEK_LAST);
 815
 816         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 817         if (ret_len == BT_SOCKET_ERROR) {
 818                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 819                 goto error;
 820         }
 821         BT_ASSERT(ret_len == sizeof(cmd));
 822
 823         ret_len = lttng_live_send(viewer_connection, &rq, sizeof(rq));
 824         if (ret_len == BT_SOCKET_ERROR) {
 825                 BT_LOGE("Error sending attach request: %s", bt_socket_errormsg());
 826                 goto error;
 827         }
 828         BT_ASSERT(ret_len == sizeof(rq));
 829
 830         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
 831         if (ret_len == 0) {
 832                 BT_LOGI("Remote side has closed connection");
 833                 goto error;
 834         }
 835         if (ret_len == BT_SOCKET_ERROR) {
 836                 BT_LOGE("Error receiving attach response: %s", bt_socket_errormsg());
 837                 goto error;
 838         }
 839         BT_ASSERT(ret_len == sizeof(rp));
 840
 841         streams_count = be32toh(rp.streams_count);
 842         switch(be32toh(rp.status)) {
 843         case LTTNG_VIEWER_ATTACH_OK:
 844                 break;
 845         case LTTNG_VIEWER_ATTACH_UNK:
 846                 BT_LOGW("Session id %" PRIu64 " is unknown", session_id);
 847                 goto error;
 848         case LTTNG_VIEWER_ATTACH_ALREADY:
 849                 BT_LOGW("There is already a viewer attached to this session");
 850                 goto error;
 851         case LTTNG_VIEWER_ATTACH_NOT_LIVE:
 852                 BT_LOGW("Not a live session");
 853                 goto error;
 854         case LTTNG_VIEWER_ATTACH_SEEK_ERR:
 855                 BT_LOGE("Wrong seek parameter");
 856                 goto error;
 857         default:
 858                 BT_LOGE("Unknown attach return code %u", be32toh(rp.status));
 859                 goto error;
 860         }
 861
 862         /* We receive the initial list of streams. */
 863         if (receive_streams(session, streams_count)) {
 864                 goto error;
 865         }
 866
 867         session->attached = true;
 868         session->new_streams_needed = false;
 869
 870         return 0;
 871
 872 error:
 873         return -1;
 874 }
 875
 876 BT_HIDDEN
 877 int lttng_live_detach_session(struct lttng_live_session *session)
 878 {
 879         struct lttng_viewer_cmd cmd;
 880         struct lttng_viewer_detach_session_request rq;
 881         struct lttng_viewer_detach_session_response rp;
 882         ssize_t ret_len;
 883         struct lttng_live_component *lttng_live = session->lttng_live;
 884         struct bt_live_viewer_connection *viewer_connection =
 885                         lttng_live->viewer_connection;
 886         uint64_t session_id = session->id;
 887
 888         if (!session->attached) {
 889                 return 0;
 890         }
 891
 892         cmd.cmd = htobe32(LTTNG_VIEWER_DETACH_SESSION);
 893         cmd.data_size = htobe64((uint64_t) sizeof(rq));
 894         cmd.cmd_version = htobe32(0);
 895
 896         memset(&rq, 0, sizeof(rq));
 897         rq.session_id = htobe64(session_id);
 898
 899         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 900         if (ret_len == BT_SOCKET_ERROR) {
 901                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 902                 goto error;
 903         }
 904         BT_ASSERT(ret_len == sizeof(cmd));
 905
 906         ret_len = lttng_live_send(viewer_connection, &rq, sizeof(rq));
 907         if (ret_len == BT_SOCKET_ERROR) {
 908                 BT_LOGE("Error sending detach request: %s", bt_socket_errormsg());
 909                 goto error;
 910         }
 911         BT_ASSERT(ret_len == sizeof(rq));
 912
 913         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
 914         if (ret_len == 0) {
 915                 BT_LOGI("Remote side has closed connection");
 916                 goto error;
 917         }
 918         if (ret_len == BT_SOCKET_ERROR) {
 919                 BT_LOGE("Error receiving detach response: %s", bt_socket_errormsg());
 920                 goto error;
 921         }
 922         BT_ASSERT(ret_len == sizeof(rp));
 923
 924         switch(be32toh(rp.status)) {
 925         case LTTNG_VIEWER_DETACH_SESSION_OK:
 926                 break;
 927         case LTTNG_VIEWER_DETACH_SESSION_UNK:
 928                 BT_LOGW("Session id %" PRIu64 " is unknown", session_id);
 929                 goto error;
 930         case LTTNG_VIEWER_DETACH_SESSION_ERR:
 931                 BT_LOGW("Error detaching session id %" PRIu64 "", session_id);
 932                 goto error;
 933         default:
 934                 BT_LOGE("Unknown detach return code %u", be32toh(rp.status));
 935                 goto error;
 936         }
 937
 938         session->attached = false;
 939
 940         return 0;
 941
 942 error:
 943         return -1;
 944 }
 945
 946 BT_HIDDEN
 947 ssize_t lttng_live_get_one_metadata_packet(struct lttng_live_trace *trace,
 948                 FILE *fp)
 949 {
 950         uint64_t len = 0;
 951         int ret;
 952         struct lttng_viewer_cmd cmd;
 953         struct lttng_viewer_get_metadata rq;
 954         struct lttng_viewer_metadata_packet rp;
 955         char *data = NULL;
 956         ssize_t ret_len;
 957         struct lttng_live_session *session = trace->session;
 958         struct lttng_live_component *lttng_live = session->lttng_live;
 959         struct lttng_live_metadata *metadata = trace->metadata;
 960         struct bt_live_viewer_connection *viewer_connection =
 961                         lttng_live->viewer_connection;
 962
 963         rq.stream_id = htobe64(metadata->stream_id);
 964         cmd.cmd = htobe32(LTTNG_VIEWER_GET_METADATA);
 965         cmd.data_size = htobe64((uint64_t) sizeof(rq));
 966         cmd.cmd_version = htobe32(0);
 967
 968         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
 969         if (ret_len == BT_SOCKET_ERROR) {
 970                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
 971                 goto error;
 972         }
 973         BT_ASSERT(ret_len == sizeof(cmd));
 974
 975         ret_len = lttng_live_send(viewer_connection, &rq, sizeof(rq));
 976         if (ret_len == BT_SOCKET_ERROR) {
 977                 BT_LOGE("Error sending get_metadata request: %s", bt_socket_errormsg());
 978                 goto error;
 979         }
 980         BT_ASSERT(ret_len == sizeof(rq));
 981
 982         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
 983         if (ret_len == 0) {
 984                 BT_LOGI("Remote side has closed connection");
 985                 goto error;
 986         }
 987         if (ret_len == BT_SOCKET_ERROR) {
 988                 BT_LOGE("Error receiving get_metadata response: %s", bt_socket_errormsg());
 989                 goto error;
 990         }
 991         BT_ASSERT(ret_len == sizeof(rp));
 992
 993         switch (be32toh(rp.status)) {
 994                 case LTTNG_VIEWER_METADATA_OK:
 995                         BT_LOGD("get_metadata : OK");
 996                         break;
 997                 case LTTNG_VIEWER_NO_NEW_METADATA:
 998                         BT_LOGD("get_metadata : NO NEW");
 999                         ret = 0;
1000                         goto end;
1001                 case LTTNG_VIEWER_METADATA_ERR:
1002                         BT_LOGD("get_metadata : ERR");
1003                         goto error;
1004                 default:
1005                         BT_LOGD("get_metadata : UNKNOWN");
1006                         goto error;
1007         }
1008
1009         len = be64toh(rp.len);
1010         BT_LOGD("Writing %" PRIu64" bytes to metadata", len);
1011         if (len <= 0) {
1012                 goto error;
1013         }
1014
1015         data = zmalloc(len);
1016         if (!data) {
1017                 BT_LOGE("relay data zmalloc: %s", strerror(errno));
1018                 goto error;
1019         }
1020         ret_len = lttng_live_recv(viewer_connection, data, len);
1021         if (ret_len == 0) {
1022                 BT_LOGI("Remote side has closed connection");
1023                 goto error_free_data;
1024         }
1025         if (ret_len == BT_SOCKET_ERROR) {
1026                 BT_LOGE("Error receiving trace packet: %s", bt_socket_errormsg());
1027                 goto error_free_data;
1028         }
1029         BT_ASSERT(ret_len == len);
1030
1031         do {
1032                 ret_len = fwrite(data, 1, len, fp);
1033         } while (ret_len < 0 && errno == EINTR);
1034         if (ret_len < 0) {
1035                 BT_LOGE("Writing in the metadata fp");
1036                 goto error_free_data;
1037         }
1038         BT_ASSERT(ret_len == len);
1039         free(data);
1040         ret = len;
1041 end:
1042         return ret;
1043
1044 error_free_data:
1045         free(data);
1046 error:
1047         return -1;
1048 }
1049
1050 /*
1051  * Assign the fields from a lttng_viewer_index to a packet_index.
1052  */
1053 static
1054 void lttng_index_to_packet_index(struct lttng_viewer_index *lindex,
1055                 struct packet_index *pindex)
1056 {
1057         BT_ASSERT(lindex);
1058         BT_ASSERT(pindex);
1059
1060         pindex->offset = be64toh(lindex->offset);
1061         pindex->packet_size = be64toh(lindex->packet_size);
1062         pindex->content_size = be64toh(lindex->content_size);
1063         pindex->ts_cycles.timestamp_begin = be64toh(lindex->timestamp_begin);
1064         pindex->ts_cycles.timestamp_end = be64toh(lindex->timestamp_end);
1065         pindex->events_discarded = be64toh(lindex->events_discarded);
1066 }
1067
1068 BT_HIDDEN
1069 enum bt_lttng_live_iterator_status lttng_live_get_next_index(struct lttng_live_component *lttng_live,
1070                 struct lttng_live_stream_iterator *stream,
1071                 struct packet_index *index)
1072 {
1073         struct lttng_viewer_cmd cmd;
1074         struct lttng_viewer_get_next_index rq;
1075         ssize_t ret_len;
1076         struct lttng_viewer_index rp;
1077         uint32_t flags, status;
1078         enum bt_lttng_live_iterator_status retstatus =
1079                         BT_LTTNG_LIVE_ITERATOR_STATUS_OK;
1080         struct bt_live_viewer_connection *viewer_connection =
1081                         lttng_live->viewer_connection;
1082         struct lttng_live_trace *trace = stream->trace;
1083
1084         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEXT_INDEX);
1085         cmd.data_size = htobe64((uint64_t) sizeof(rq));
1086         cmd.cmd_version = htobe32(0);
1087
1088         memset(&rq, 0, sizeof(rq));
1089         rq.stream_id = htobe64(stream->viewer_stream_id);
1090
1091         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
1092         if (ret_len == BT_SOCKET_ERROR) {
1093                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
1094                 goto error;
1095         }
1096         BT_ASSERT(ret_len == sizeof(cmd));
1097
1098         ret_len = lttng_live_send(viewer_connection, &rq, sizeof(rq));
1099         if (ret_len == BT_SOCKET_ERROR) {
1100                 BT_LOGE("Error sending get_next_index request: %s", bt_socket_errormsg());
1101                 goto error;
1102         }
1103         BT_ASSERT(ret_len == sizeof(rq));
1104
1105         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1106         if (ret_len == 0) {
1107                 BT_LOGI("Remote side has closed connection");
1108                 goto error;
1109         }
1110         if (ret_len == BT_SOCKET_ERROR) {
1111                 BT_LOGE("Error receiving get_next_index response: %s", bt_socket_errormsg());
1112                 goto error;
1113         }
1114         BT_ASSERT(ret_len == sizeof(rp));
1115
1116         flags = be32toh(rp.flags);
1117         status = be32toh(rp.status);
1118
1119         switch (status) {
1120         case LTTNG_VIEWER_INDEX_INACTIVE:
1121         {
1122                 uint64_t ctf_stream_class_id;
1123
1124                 BT_LOGD("get_next_index: inactive");
1125                 memset(index, 0, sizeof(struct packet_index));
1126                 index->ts_cycles.timestamp_end = be64toh(rp.timestamp_end);
1127                 stream->current_inactivity_timestamp = index->ts_cycles.timestamp_end;
1128                 ctf_stream_class_id = be64toh(rp.stream_id);
1129                 if (stream->ctf_stream_class_id != -1ULL) {
1130                         BT_ASSERT(stream->ctf_stream_class_id ==
1131                                 ctf_stream_class_id);
1132                 } else {
1133                         stream->ctf_stream_class_id = ctf_stream_class_id;
1134                 }
1135                 stream->state = LTTNG_LIVE_STREAM_QUIESCENT;
1136                 break;
1137         }
1138         case LTTNG_VIEWER_INDEX_OK:
1139         {
1140                 uint64_t ctf_stream_class_id;
1141
1142                 BT_LOGD("get_next_index: OK");
1143                 lttng_index_to_packet_index(&rp, index);
1144                 ctf_stream_class_id = be64toh(rp.stream_id);
1145                 if (stream->ctf_stream_class_id != -1ULL) {
1146                         BT_ASSERT(stream->ctf_stream_class_id ==
1147                                 ctf_stream_class_id);
1148                 } else {
1149                         stream->ctf_stream_class_id = ctf_stream_class_id;
1150                 }
1151
1152                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_DATA;
1153                 stream->current_packet_end_timestamp =
1154                         index->ts_cycles.timestamp_end;
1155
1156                 if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1157                         BT_LOGD("get_next_index: new metadata needed");
1158                         trace->new_metadata_needed = true;
1159                 }
1160                 if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1161                         BT_LOGD("get_next_index: new streams needed");
1162                         lttng_live_need_new_streams(lttng_live);
1163                 }
1164                 break;
1165         }
1166         case LTTNG_VIEWER_INDEX_RETRY:
1167                 BT_LOGD("get_next_index: retry");
1168                 memset(index, 0, sizeof(struct packet_index));
1169                 retstatus = BT_LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1170                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1171                 goto end;
1172         case LTTNG_VIEWER_INDEX_HUP:
1173                 BT_LOGD("get_next_index: stream hung up");
1174                 memset(index, 0, sizeof(struct packet_index));
1175                 index->offset = EOF;
1176                 retstatus = BT_LTTNG_LIVE_ITERATOR_STATUS_END;
1177                 stream->state = LTTNG_LIVE_STREAM_EOF;
1178                 break;
1179         case LTTNG_VIEWER_INDEX_ERR:
1180                 BT_LOGE("get_next_index: error");
1181                 memset(index, 0, sizeof(struct packet_index));
1182                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1183                 goto error;
1184         default:
1185                 BT_LOGE("get_next_index: unknown value");
1186                 memset(index, 0, sizeof(struct packet_index));
1187                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
1188                 goto error;
1189         }
1190 end:
1191         return retstatus;
1192
1193 error:
1194         if (lttng_live_is_canceled(lttng_live)) {
1195                 retstatus = BT_LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1196         } else {
1197                 retstatus = BT_LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1198         }
1199         return retstatus;
1200 }
1201
1202 BT_HIDDEN
1203 enum bt_msg_iter_medium_status lttng_live_get_stream_bytes(struct lttng_live_component *lttng_live,
1204                 struct lttng_live_stream_iterator *stream, uint8_t *buf, uint64_t offset,
1205                 uint64_t req_len, uint64_t *recv_len)
1206 {
1207         enum bt_msg_iter_medium_status retstatus = BT_MSG_ITER_MEDIUM_STATUS_OK;
1208         struct lttng_viewer_cmd cmd;
1209         struct lttng_viewer_get_packet rq;
1210         struct lttng_viewer_trace_packet rp;
1211         ssize_t ret_len;
1212         uint32_t flags, status;
1213         struct bt_live_viewer_connection *viewer_connection =
1214                         lttng_live->viewer_connection;
1215         struct lttng_live_trace *trace = stream->trace;
1216
1217         BT_LOGD("lttng_live_get_stream_bytes: offset=%" PRIu64 ", req_len=%" PRIu64,
1218                         offset, req_len);
1219         cmd.cmd = htobe32(LTTNG_VIEWER_GET_PACKET);
1220         cmd.data_size = htobe64((uint64_t) sizeof(rq));
1221         cmd.cmd_version = htobe32(0);
1222
1223         memset(&rq, 0, sizeof(rq));
1224         rq.stream_id = htobe64(stream->viewer_stream_id);
1225         rq.offset = htobe64(offset);
1226         rq.len = htobe32(req_len);
1227
1228         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
1229         if (ret_len == BT_SOCKET_ERROR) {
1230                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
1231                 goto error;
1232         }
1233         BT_ASSERT(ret_len == sizeof(cmd));
1234
1235         ret_len = lttng_live_send(viewer_connection, &rq, sizeof(rq));
1236         if (ret_len == BT_SOCKET_ERROR) {
1237                 BT_LOGE("Error sending get_data request: %s", bt_socket_errormsg());
1238                 goto error;
1239         }
1240         BT_ASSERT(ret_len == sizeof(rq));
1241
1242         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1243         if (ret_len == 0) {
1244                 BT_LOGI("Remote side has closed connection");
1245                 goto error;
1246         }
1247         if (ret_len == BT_SOCKET_ERROR) {
1248                 BT_LOGE("Error receiving get_data response: %s", bt_socket_errormsg());
1249                 goto error;
1250         }
1251         if (ret_len != sizeof(rp)) {
1252                 BT_LOGE("get_data_packet: expected %zu"
1253                                 ", received %zd", sizeof(rp),
1254                                 ret_len);
1255                 goto error;
1256         }
1257
1258         flags = be32toh(rp.flags);
1259         status = be32toh(rp.status);
1260
1261         switch (status) {
1262         case LTTNG_VIEWER_GET_PACKET_OK:
1263                 req_len = be32toh(rp.len);
1264                 BT_LOGD("get_data_packet: Ok, packet size : %" PRIu64 "", req_len);
1265                 break;
1266         case LTTNG_VIEWER_GET_PACKET_RETRY:
1267                 /* Unimplemented by relay daemon */
1268                 BT_LOGD("get_data_packet: retry");
1269                 retstatus = BT_MSG_ITER_MEDIUM_STATUS_AGAIN;
1270                 goto end;
1271         case LTTNG_VIEWER_GET_PACKET_ERR:
1272                 if (flags & LTTNG_VIEWER_FLAG_NEW_METADATA) {
1273                         BT_LOGD("get_data_packet: new metadata needed, try again later");
1274                         trace->new_metadata_needed = true;
1275                 }
1276                 if (flags & LTTNG_VIEWER_FLAG_NEW_STREAM) {
1277                         BT_LOGD("get_data_packet: new streams needed, try again later");
1278                         lttng_live_need_new_streams(lttng_live);
1279                 }
1280                 if (flags & (LTTNG_VIEWER_FLAG_NEW_METADATA
1281                                 | LTTNG_VIEWER_FLAG_NEW_STREAM)) {
1282                         retstatus = BT_MSG_ITER_MEDIUM_STATUS_AGAIN;
1283                         goto end;
1284                 }
1285                 BT_LOGE("get_data_packet: error");
1286                 goto error;
1287         case LTTNG_VIEWER_GET_PACKET_EOF:
1288                 retstatus = BT_MSG_ITER_MEDIUM_STATUS_EOF;
1289                 goto end;
1290         default:
1291                 BT_LOGE("get_data_packet: unknown");
1292                 goto error;
1293         }
1294
1295         if (req_len == 0) {
1296                 goto error;
1297         }
1298
1299         ret_len = lttng_live_recv(viewer_connection, buf, req_len);
1300         if (ret_len == 0) {
1301                 BT_LOGI("Remote side has closed connection");
1302                 goto error;
1303         }
1304         if (ret_len == BT_SOCKET_ERROR) {
1305                 BT_LOGE("Error receiving trace packet: %s", bt_socket_errormsg());
1306                 goto error;
1307         }
1308         BT_ASSERT(ret_len == req_len);
1309         *recv_len = ret_len;
1310 end:
1311         return retstatus;
1312
1313 error:
1314         if (lttng_live_is_canceled(lttng_live)) {
1315                 retstatus = BT_MSG_ITER_MEDIUM_STATUS_AGAIN;
1316         } else {
1317                 retstatus = BT_MSG_ITER_MEDIUM_STATUS_ERROR;
1318         }
1319         return retstatus;
1320 }
1321
1322 /*
1323  * Request new streams for a session.
1324  */
1325 BT_HIDDEN
1326 enum bt_lttng_live_iterator_status lttng_live_get_new_streams(
1327                 struct lttng_live_session *session)
1328 {
1329         enum bt_lttng_live_iterator_status status =
1330                         BT_LTTNG_LIVE_ITERATOR_STATUS_OK;
1331         struct lttng_viewer_cmd cmd;
1332         struct lttng_viewer_new_streams_request rq;
1333         struct lttng_viewer_new_streams_response rp;
1334         ssize_t ret_len;
1335         struct lttng_live_component *lttng_live = session->lttng_live;
1336         struct bt_live_viewer_connection *viewer_connection =
1337                         lttng_live->viewer_connection;
1338         uint32_t streams_count;
1339
1340         if (!session->new_streams_needed) {
1341                 return BT_LTTNG_LIVE_ITERATOR_STATUS_OK;
1342         }
1343
1344         cmd.cmd = htobe32(LTTNG_VIEWER_GET_NEW_STREAMS);
1345         cmd.data_size = htobe64((uint64_t) sizeof(rq));
1346         cmd.cmd_version = htobe32(0);
1347
1348         memset(&rq, 0, sizeof(rq));
1349         rq.session_id = htobe64(session->id);
1350
1351         ret_len = lttng_live_send(viewer_connection, &cmd, sizeof(cmd));
1352         if (ret_len == BT_SOCKET_ERROR) {
1353                 BT_LOGE("Error sending cmd: %s", bt_socket_errormsg());
1354                 goto error;
1355         }
1356         BT_ASSERT(ret_len == sizeof(cmd));
1357
1358         ret_len = lttng_live_send(viewer_connection, &rq, sizeof(rq));
1359         if (ret_len == BT_SOCKET_ERROR) {
1360                 BT_LOGE("Error sending get_new_streams request: %s", bt_socket_errormsg());
1361                 goto error;
1362         }
1363         BT_ASSERT(ret_len == sizeof(rq));
1364
1365         ret_len = lttng_live_recv(viewer_connection, &rp, sizeof(rp));
1366         if (ret_len == 0) {
1367                 BT_LOGI("Remote side has closed connection");
1368                 goto error;
1369         }
1370         if (ret_len == BT_SOCKET_ERROR) {
1371                 BT_LOGE("Error receiving get_new_streams response");
1372                 goto error;
1373         }
1374         BT_ASSERT(ret_len == sizeof(rp));
1375
1376         streams_count = be32toh(rp.streams_count);
1377
1378         switch(be32toh(rp.status)) {
1379         case LTTNG_VIEWER_NEW_STREAMS_OK:
1380                 session->new_streams_needed = false;
1381                 break;
1382         case LTTNG_VIEWER_NEW_STREAMS_NO_NEW:
1383                 session->new_streams_needed = false;
1384                 goto end;
1385         case LTTNG_VIEWER_NEW_STREAMS_HUP:
1386                 session->new_streams_needed = false;
1387                 session->closed = true;
1388                 status = BT_LTTNG_LIVE_ITERATOR_STATUS_END;
1389                 goto end;
1390         case LTTNG_VIEWER_NEW_STREAMS_ERR:
1391                 BT_LOGE("get_new_streams error");
1392                 goto error;
1393         default:
1394                 BT_LOGE("Unknown return code %u", be32toh(rp.status));
1395                 goto error;
1396         }
1397
1398         if (receive_streams(session, streams_count)) {
1399                 goto error;
1400         }
1401 end:
1402         return status;
1403
1404 error:
1405         if (lttng_live_is_canceled(lttng_live)) {
1406                 status = BT_LTTNG_LIVE_ITERATOR_STATUS_AGAIN;
1407         } else {
1408                 status = BT_LTTNG_LIVE_ITERATOR_STATUS_ERROR;
1409         }
1410         return status;
1411 }
1412
1413 BT_HIDDEN
1414 struct bt_live_viewer_connection *
1415         bt_live_viewer_connection_create(const char *url,
1416                         struct lttng_live_component *lttng_live)
1417 {
1418         struct bt_live_viewer_connection *viewer_connection;
1419
1420         viewer_connection = g_new0(struct bt_live_viewer_connection, 1);
1421
1422         if (bt_socket_init() != 0) {
1423                 goto error;
1424         }
1425
1426         bt_object_init(&viewer_connection->obj, connection_release);
1427         viewer_connection->control_sock = BT_INVALID_SOCKET;
1428         viewer_connection->port = -1;
1429         viewer_connection->lttng_live = lttng_live;
1430         viewer_connection->url = g_string_new(url);
1431         if (!viewer_connection->url) {
1432                 goto error;
1433         }
1434
1435         BT_LOGD("Establishing connection to url \"%s\"...", url);
1436         if (lttng_live_connect_viewer(viewer_connection)) {
1437                 goto error_report;
1438         }
1439         BT_LOGD("Connection to url \"%s\" is established", url);
1440         return viewer_connection;
1441
1442 error_report:
1443         BT_LOGW("Failure to establish connection to url \"%s\"", url);
1444 error:
1445         g_free(viewer_connection);
1446         return NULL;
1447 }
1448
1449 BT_HIDDEN
1450 void bt_live_viewer_connection_destroy(struct bt_live_viewer_connection *viewer_connection)
1451 {
1452         BT_LOGD("Closing connection to url \"%s\"", viewer_connection->url->str);
1453         lttng_live_disconnect_viewer(viewer_connection);
1454         g_string_free(viewer_connection->url, TRUE);
1455         if (viewer_connection->relay_hostname) {
1456                 g_string_free(viewer_connection->relay_hostname, TRUE);
1457         }
1458         if (viewer_connection->target_hostname) {
1459                 g_string_free(viewer_connection->target_hostname, TRUE);
1460         }
1461         if (viewer_connection->session_name) {
1462                 g_string_free(viewer_connection->session_name, TRUE);
1463         }
1464         g_free(viewer_connection);
1465
1466         bt_socket_fini();
1467 }