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