plugins/ctf/fs-sink/write.c

   1 /*
   2  * writer.c
   3  *
   4  * Babeltrace CTF Writer Output Plugin Event Handling
   5  *
   6  * Copyright 2016 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   7  *
   8  * Author: Jérémie Galarneau <jeremie.galarneau@efficios.com>
   9  *
  10  * Permission is hereby granted, free of charge, to any person obtaining a copy
  11  * of this software and associated documentation files (the "Software"), to deal
  12  * in the Software without restriction, including without limitation the rights
  13  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  14  * copies of the Software, and to permit persons to whom the Software is
  15  * furnished to do so, subject to the following conditions:
  16  *
  17  * The above copyright notice and this permission notice shall be included in
  18  * all copies or substantial portions of the Software.
  19  *
  20  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  21  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  22  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  23  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  24  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  25  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  26  * SOFTWARE.
  27  */
  28
  29 #define BT_LOG_TAG "PLUGIN-CTF-FS-SINK-WRITE"
  30 #include "logging.h"
  31
  32 #include <babeltrace/ctf-ir/event.h>
  33 #include <babeltrace/ctf-ir/packet.h>
  34 #include <babeltrace/ctf-ir/event-class.h>
  35 #include <babeltrace/ctf-ir/stream.h>
  36 #include <babeltrace/ctf-ir/stream-class.h>
  37 #include <babeltrace/ctf-ir/clock-class.h>
  38 #include <babeltrace/ctf-ir/fields.h>
  39 #include <babeltrace/ctf-writer/stream-class.h>
  40 #include <babeltrace/ctf-writer/stream.h>
  41 #include <assert.h>
  42 #include <glib.h>
  43
  44 #include <ctfcopytrace.h>
  45
  46 #include "writer.h"
  47
  48 static
  49 void unref_stream_class(struct bt_ctf_stream_class *writer_stream_class)
  50 {
  51         bt_put(writer_stream_class);
  52 }
  53
  54 static
  55 void unref_stream(struct bt_ctf_stream_class *writer_stream)
  56 {
  57         bt_put(writer_stream);
  58 }
  59
  60 static
  61 gboolean empty_ht(gpointer key, gpointer value, gpointer user_data)
  62 {
  63         return TRUE;
  64 }
  65
  66 static
  67 gboolean empty_streams_ht(gpointer key, gpointer value, gpointer user_data)
  68 {
  69         struct bt_ctf_stream *writer_stream = value;
  70
  71         bt_ctf_stream_flush(writer_stream);
  72
  73         return TRUE;
  74 }
  75
  76 static
  77 void destroy_stream_state_key(gpointer key)
  78 {
  79         g_free((enum fs_writer_stream_state *) key);
  80 }
  81
  82 static
  83 void check_completed_trace(gpointer key, gpointer value, gpointer user_data)
  84 {
  85         enum fs_writer_stream_state *state = value;
  86         int *trace_completed = user_data;
  87
  88         if (*state != FS_WRITER_COMPLETED_STREAM) {
  89                 *trace_completed = 0;
  90         }
  91 }
  92
  93 static
  94 void trace_is_static_listener(struct bt_ctf_trace *trace, void *data)
  95 {
  96         struct fs_writer *fs_writer = data;
  97         int trace_completed = 1;
  98
  99         fs_writer->trace_static = 1;
 100
 101         g_hash_table_foreach(fs_writer->stream_states,
 102                         check_completed_trace, &trace_completed);
 103         if (trace_completed) {
 104                 writer_close(fs_writer->writer_component, fs_writer);
 105                 g_hash_table_remove(fs_writer->writer_component->trace_map,
 106                                 fs_writer->trace);
 107         }
 108 }
 109
 110 static
 111 struct bt_ctf_stream_class *insert_new_stream_class(
 112                 struct writer_component *writer_component,
 113                 struct fs_writer *fs_writer,
 114                 struct bt_ctf_stream_class *stream_class)
 115 {
 116         struct bt_ctf_stream_class *writer_stream_class = NULL;
 117         struct bt_ctf_trace *trace = NULL, *writer_trace = NULL;
 118         struct bt_ctf_writer *ctf_writer = fs_writer->writer;
 119         enum bt_component_status ret;
 120
 121         trace = bt_ctf_stream_class_get_trace(stream_class);
 122         assert(trace);
 123
 124         writer_trace = bt_ctf_writer_get_trace(ctf_writer);
 125         assert(writer_trace);
 126
 127         ret = ctf_copy_clock_classes(writer_component->err, writer_trace,
 128                         writer_stream_class, trace);
 129         if (ret != BT_COMPONENT_STATUS_OK) {
 130                 BT_LOGE_STR("Failed to copy clock classes.");
 131                 goto error;
 132         }
 133
 134         writer_stream_class = ctf_copy_stream_class(writer_component->err,
 135                         stream_class, writer_trace, true);
 136         if (!writer_stream_class) {
 137                 BT_LOGE_STR("Failed to copy stream class.");
 138                 goto error;
 139         }
 140
 141         ret = bt_ctf_trace_add_stream_class(writer_trace, writer_stream_class);
 142         if (ret) {
 143                 BT_LOGE_STR("Failed to add stream_class.");
 144                 goto error;
 145         }
 146
 147         g_hash_table_insert(fs_writer->stream_class_map,
 148                         (gpointer) stream_class, writer_stream_class);
 149
 150         goto end;
 151
 152 error:
 153         BT_PUT(writer_stream_class);
 154 end:
 155         bt_put(writer_trace);
 156         bt_put(trace);
 157         return writer_stream_class;
 158 }
 159
 160 static
 161 enum fs_writer_stream_state *insert_new_stream_state(
 162                 struct writer_component *writer_component,
 163                 struct fs_writer *fs_writer, struct bt_ctf_stream *stream)
 164 {
 165         enum fs_writer_stream_state *v = NULL;
 166
 167         v = g_new0(enum fs_writer_stream_state, 1);
 168         if (!v) {
 169                 BT_LOGE_STR("Failed to allocate fs_writer_stream_state.");
 170                 goto end;
 171         }
 172         *v = FS_WRITER_UNKNOWN_STREAM;
 173
 174         g_hash_table_insert(fs_writer->stream_states, stream, v);
 175 end:
 176         return v;
 177 }
 178
 179 /*
 180  * Make sure the output path is valid for a single trace: either it does
 181  * not exists or it is empty.
 182  *
 183  * Return 0 if the path is valid, -1 otherwise.
 184  */
 185 static
 186 bool valid_single_trace_path(const char *path)
 187 {
 188         GError *error = NULL;
 189         GDir *dir = NULL;
 190         int ret = 0;
 191
 192         dir = g_dir_open(path, 0, &error);
 193
 194         /* Non-existent directory. */
 195         if (!dir) {
 196                 /* For any other error, return an error */
 197                 if (error->code != G_FILE_ERROR_NOENT) {
 198                         ret = -1;
 199                 }
 200                 goto end;
 201         }
 202
 203         /* g_dir_read_name skips "." and "..", error out on first result */
 204         while (g_dir_read_name(dir) != NULL) {
 205                 ret = -1;
 206                 break;
 207         }
 208
 209 end:
 210         if (dir) {
 211                 g_dir_close(dir);
 212         }
 213         if (error) {
 214                 g_error_free(error);
 215         }
 216
 217         return ret;
 218 }
 219
 220 static
 221 int make_trace_path(struct writer_component *writer_component,
 222                 struct bt_ctf_trace *trace, char *trace_path)
 223 {
 224         int ret;
 225         const char *trace_name;
 226
 227         if (writer_component->single_trace) {
 228                 trace_name = "\0";
 229         } else {
 230                 trace_name = bt_ctf_trace_get_name(trace);
 231                 if (!trace_name) {
 232                         trace_name = writer_component->trace_name_base->str;
 233                 }
 234         }
 235
 236         /* Sanitize the trace name. */
 237         if (strlen(trace_name) == 2 && !strcmp(trace_name, "..")) {
 238                 BT_LOGE_STR("Trace name cannot be \"..\".");
 239                 goto error;
 240         }
 241
 242         if (strstr(trace_name, "../")) {
 243                 BT_LOGE_STR("Trace name cannot contain \"../\".");
 244                 goto error;
 245
 246         }
 247
 248         snprintf(trace_path, PATH_MAX, "%s/%s",
 249                         writer_component->base_path->str,
 250                         trace_name);
 251         /*
 252          * Append a suffix if the trace_path exists and we are not in
 253          * single-trace mode.
 254          */
 255         if (writer_component->single_trace) {
 256                 if (valid_single_trace_path(trace_path) != 0) {
 257                         BT_LOGE_STR("Invalid output directory.");
 258                         goto error;
 259                 }
 260         } else {
 261                 if (g_file_test(trace_path, G_FILE_TEST_EXISTS)) {
 262                         int i = 0;
 263
 264                         do {
 265                                 snprintf(trace_path, PATH_MAX, "%s/%s-%d",
 266                                                 writer_component->base_path->str,
 267                                                 trace_name, ++i);
 268                         } while (g_file_test(trace_path, G_FILE_TEST_EXISTS) && i < INT_MAX);
 269                         if (i == INT_MAX) {
 270                                 BT_LOGE_STR("Unable to find a unique trace path.");
 271                                 goto error;
 272                         }
 273                 }
 274         }
 275
 276         ret = 0;
 277         goto end;
 278
 279 error:
 280         ret = -1;
 281 end:
 282         return ret;
 283 }
 284
 285 static
 286 struct fs_writer *insert_new_writer(
 287                 struct writer_component *writer_component,
 288                 struct bt_ctf_trace *trace)
 289 {
 290         struct bt_ctf_writer *ctf_writer = NULL;
 291         struct bt_ctf_trace *writer_trace = NULL;
 292         char trace_path[PATH_MAX];
 293         enum bt_component_status ret;
 294         struct bt_ctf_stream *stream = NULL;
 295         struct fs_writer *fs_writer = NULL;
 296         int nr_stream, i;
 297
 298         if (writer_component->single_trace && writer_component->nr_traces > 0) {
 299                 BT_LOGE_STR("Trying to process more than one trace but single trace mode enabled.");
 300                 goto error;
 301         }
 302
 303         ret = make_trace_path(writer_component, trace, trace_path);
 304         if (ret) {
 305                 BT_LOGE_STR("Failed to make trace path.");
 306                 goto error;
 307         }
 308
 309         printf("ctf.fs sink creating trace in %s\n", trace_path);
 310
 311         ctf_writer = bt_ctf_writer_create(trace_path);
 312         if (!ctf_writer) {
 313                 BT_LOGE_STR("Failed to create CTF writer.");
 314                 goto error;
 315         }
 316
 317         writer_trace = bt_ctf_writer_get_trace(ctf_writer);
 318         assert(writer_trace);
 319
 320         ret = ctf_copy_trace(writer_component->err, trace, writer_trace);
 321         if (ret != BT_COMPONENT_STATUS_OK) {
 322                 BT_LOGE_STR("Failed to copy trace.");
 323                 BT_PUT(ctf_writer);
 324                 goto error;
 325         }
 326
 327         fs_writer = g_new0(struct fs_writer, 1);
 328         if (!fs_writer) {
 329                 BT_LOGE_STR("Failed to allocate fs_writer.");
 330                 goto error;
 331         }
 332         fs_writer->writer = ctf_writer;
 333         fs_writer->trace = trace;
 334         fs_writer->writer_trace = writer_trace;
 335         fs_writer->writer_component = writer_component;
 336         BT_PUT(writer_trace);
 337         fs_writer->stream_class_map = g_hash_table_new_full(g_direct_hash,
 338                         g_direct_equal, NULL, (GDestroyNotify) unref_stream_class);
 339         fs_writer->stream_map = g_hash_table_new_full(g_direct_hash,
 340                         g_direct_equal, NULL, (GDestroyNotify) unref_stream);
 341         fs_writer->stream_states = g_hash_table_new_full(g_direct_hash,
 342                         g_direct_equal, NULL, destroy_stream_state_key);
 343
 344         /* Set all the existing streams in the unknown state. */
 345         nr_stream = bt_ctf_trace_get_stream_count(trace);
 346         for (i = 0; i < nr_stream; i++) {
 347                 stream = bt_ctf_trace_get_stream_by_index(trace, i);
 348                 assert(stream);
 349
 350                 insert_new_stream_state(writer_component, fs_writer, stream);
 351                 BT_PUT(stream);
 352         }
 353
 354         /* Check if the trace is already static or register a listener. */
 355         if (bt_ctf_trace_is_static(trace)) {
 356                 fs_writer->trace_static = 1;
 357                 fs_writer->static_listener_id = -1;
 358         } else {
 359                 ret = bt_ctf_trace_add_is_static_listener(trace,
 360                                 trace_is_static_listener, fs_writer);
 361                 assert(ret >= 0);
 362                 fs_writer->static_listener_id = ret;
 363         }
 364
 365         writer_component->nr_traces++;
 366         g_hash_table_insert(writer_component->trace_map, (gpointer) trace,
 367                         fs_writer);
 368
 369         goto end;
 370
 371 error:
 372         g_free(fs_writer);
 373         fs_writer = NULL;
 374         bt_put(writer_trace);
 375         bt_put(stream);
 376         BT_PUT(ctf_writer);
 377 end:
 378         return fs_writer;
 379 }
 380
 381 static
 382 struct fs_writer *get_fs_writer(struct writer_component *writer_component,
 383                 struct bt_ctf_stream_class *stream_class)
 384 {
 385         struct bt_ctf_trace *trace = NULL;
 386         struct fs_writer *fs_writer;
 387
 388         trace = bt_ctf_stream_class_get_trace(stream_class);
 389         assert(trace);
 390
 391         fs_writer = g_hash_table_lookup(writer_component->trace_map,
 392                         (gpointer) trace);
 393         if (!fs_writer) {
 394                 fs_writer = insert_new_writer(writer_component, trace);
 395         }
 396         BT_PUT(trace);
 397
 398         return fs_writer;
 399 }
 400
 401 static
 402 struct fs_writer *get_fs_writer_from_stream(
 403                 struct writer_component *writer_component,
 404                 struct bt_ctf_stream *stream)
 405 {
 406         struct bt_ctf_stream_class *stream_class = NULL;
 407         struct fs_writer *fs_writer;
 408
 409         stream_class = bt_ctf_stream_get_class(stream);
 410         assert(stream_class);
 411
 412         fs_writer = get_fs_writer(writer_component, stream_class);
 413
 414         bt_put(stream_class);
 415         return fs_writer;
 416 }
 417
 418 static
 419 struct bt_ctf_stream_class *lookup_stream_class(
 420                 struct writer_component *writer_component,
 421                 struct bt_ctf_stream_class *stream_class)
 422 {
 423         struct fs_writer *fs_writer = get_fs_writer(
 424                         writer_component, stream_class);
 425         assert(fs_writer);
 426         return (struct bt_ctf_stream_class *) g_hash_table_lookup(
 427                         fs_writer->stream_class_map, (gpointer) stream_class);
 428 }
 429
 430 static
 431 struct bt_ctf_stream *lookup_stream(struct writer_component *writer_component,
 432                 struct bt_ctf_stream *stream)
 433 {
 434         struct fs_writer *fs_writer = get_fs_writer_from_stream(
 435                         writer_component, stream);
 436         assert(fs_writer);
 437         return (struct bt_ctf_stream *) g_hash_table_lookup(
 438                         fs_writer->stream_map, (gpointer) stream);
 439 }
 440
 441 static
 442 struct bt_ctf_stream *insert_new_stream(
 443                 struct writer_component *writer_component,
 444                 struct fs_writer *fs_writer,
 445                 struct bt_ctf_stream_class *stream_class,
 446                 struct bt_ctf_stream *stream)
 447 {
 448         struct bt_ctf_stream *writer_stream = NULL;
 449         struct bt_ctf_stream_class *writer_stream_class = NULL;
 450         struct bt_ctf_writer *ctf_writer = bt_get(fs_writer->writer);
 451
 452         writer_stream_class = lookup_stream_class(writer_component,
 453                         stream_class);
 454         if (!writer_stream_class) {
 455                 writer_stream_class = insert_new_stream_class(
 456                                 writer_component, fs_writer, stream_class);
 457                 if (!writer_stream_class) {
 458                         BT_LOGE_STR("Failed to insert a new stream_class.");
 459                         goto error;
 460                 }
 461         }
 462         bt_get(writer_stream_class);
 463
 464         writer_stream = bt_ctf_stream_create(writer_stream_class,
 465                 bt_ctf_stream_get_name(stream));
 466         assert(writer_stream);
 467
 468         g_hash_table_insert(fs_writer->stream_map, (gpointer) stream,
 469                         writer_stream);
 470
 471         goto end;
 472
 473 error:
 474         BT_PUT(writer_stream);
 475 end:
 476         bt_put(ctf_writer);
 477         bt_put(writer_stream_class);
 478         return writer_stream;
 479 }
 480
 481 static
 482 struct bt_ctf_event_class *get_event_class(struct writer_component *writer_component,
 483                 struct bt_ctf_stream_class *writer_stream_class,
 484                 struct bt_ctf_event_class *event_class)
 485 {
 486         return bt_ctf_stream_class_get_event_class_by_id(writer_stream_class,
 487                         bt_ctf_event_class_get_id(event_class));
 488 }
 489
 490 static
 491 struct bt_ctf_stream *get_writer_stream(
 492                 struct writer_component *writer_component,
 493                 struct bt_ctf_packet *packet, struct bt_ctf_stream *stream)
 494 {
 495         struct bt_ctf_stream *writer_stream = NULL;
 496
 497         writer_stream = lookup_stream(writer_component, stream);
 498         if (!writer_stream) {
 499                 BT_LOGE_STR("Failed to find existing stream.");
 500                 goto error;
 501         }
 502         bt_get(writer_stream);
 503
 504         goto end;
 505
 506 error:
 507         BT_PUT(writer_stream);
 508 end:
 509         return writer_stream;
 510 }
 511
 512 BT_HIDDEN
 513 void writer_close(struct writer_component *writer_component,
 514                 struct fs_writer *fs_writer)
 515 {
 516         if (fs_writer->static_listener_id >= 0) {
 517                 bt_ctf_trace_remove_is_static_listener(fs_writer->trace,
 518                                 fs_writer->static_listener_id);
 519         }
 520
 521         /* Empty the stream class HT. */
 522         g_hash_table_foreach_remove(fs_writer->stream_class_map,
 523                         empty_ht, NULL);
 524         g_hash_table_destroy(fs_writer->stream_class_map);
 525
 526         /* Empty the stream HT. */
 527         g_hash_table_foreach_remove(fs_writer->stream_map,
 528                         empty_streams_ht, NULL);
 529         g_hash_table_destroy(fs_writer->stream_map);
 530
 531         /* Empty the stream state HT. */
 532         g_hash_table_foreach_remove(fs_writer->stream_states,
 533                         empty_ht, NULL);
 534         g_hash_table_destroy(fs_writer->stream_states);
 535 }
 536
 537 BT_HIDDEN
 538 enum bt_component_status writer_stream_begin(
 539                 struct writer_component *writer_component,
 540                 struct bt_ctf_stream *stream)
 541 {
 542         struct bt_ctf_stream_class *stream_class = NULL;
 543         struct fs_writer *fs_writer;
 544         struct bt_ctf_stream *writer_stream = NULL;
 545         enum bt_component_status ret = BT_COMPONENT_STATUS_OK;
 546         enum fs_writer_stream_state *state;
 547
 548         stream_class = bt_ctf_stream_get_class(stream);
 549         assert(stream_class);
 550
 551         fs_writer = get_fs_writer(writer_component, stream_class);
 552         if (!fs_writer) {
 553                 BT_LOGE_STR("Failed to get fs_writer.");
 554                 goto error;
 555         }
 556
 557         /* Set the stream as active */
 558         state = g_hash_table_lookup(fs_writer->stream_states, stream);
 559         if (!state) {
 560                 if (fs_writer->trace_static) {
 561                         BT_LOGE_STR("Cannot add new stream on a static trace.");
 562                         goto error;
 563                 }
 564                 state = insert_new_stream_state(writer_component, fs_writer,
 565                                 stream);
 566         }
 567         if (*state != FS_WRITER_UNKNOWN_STREAM) {
 568                 BT_LOGE("Unexpected stream state: state=%d", *state);
 569                 goto error;
 570         }
 571         *state = FS_WRITER_ACTIVE_STREAM;
 572
 573         writer_stream = insert_new_stream(writer_component, fs_writer,
 574                         stream_class, stream);
 575         if (!writer_stream) {
 576                 BT_LOGE_STR("Failed to insert new stream.");
 577                 goto error;
 578         }
 579
 580         goto end;
 581
 582 error:
 583         ret = BT_COMPONENT_STATUS_ERROR;
 584 end:
 585         bt_put(stream_class);
 586         return ret;
 587 }
 588
 589 BT_HIDDEN
 590 enum bt_component_status writer_stream_end(
 591                 struct writer_component *writer_component,
 592                 struct bt_ctf_stream *stream)
 593 {
 594         struct bt_ctf_stream_class *stream_class = NULL;
 595         struct fs_writer *fs_writer;
 596         struct bt_ctf_trace *trace = NULL;
 597         enum bt_component_status ret = BT_COMPONENT_STATUS_OK;
 598         enum fs_writer_stream_state *state;
 599
 600         stream_class = bt_ctf_stream_get_class(stream);
 601         assert(stream_class);
 602
 603         fs_writer = get_fs_writer(writer_component, stream_class);
 604         if (!fs_writer) {
 605                 BT_LOGE_STR("Failed to get fs_writer.");
 606                 goto error;
 607         }
 608
 609         state = g_hash_table_lookup(fs_writer->stream_states, stream);
 610         if (*state != FS_WRITER_ACTIVE_STREAM) {
 611                 BT_LOGE("Unexpected stream state: state=%d", *state);
 612                 goto error;
 613         }
 614         *state = FS_WRITER_COMPLETED_STREAM;
 615
 616         g_hash_table_remove(fs_writer->stream_map, stream);
 617
 618         if (fs_writer->trace_static) {
 619                 int trace_completed = 1;
 620
 621                 g_hash_table_foreach(fs_writer->stream_states,
 622                                 check_completed_trace, &trace_completed);
 623                 if (trace_completed) {
 624                         writer_close(writer_component, fs_writer);
 625                         g_hash_table_remove(writer_component->trace_map,
 626                                         fs_writer->trace);
 627                 }
 628         }
 629
 630         goto end;
 631
 632 error:
 633         ret = BT_COMPONENT_STATUS_ERROR;
 634 end:
 635         BT_PUT(trace);
 636         BT_PUT(stream_class);
 637         return ret;
 638 }
 639
 640 BT_HIDDEN
 641 enum bt_component_status writer_new_packet(
 642                 struct writer_component *writer_component,
 643                 struct bt_ctf_packet *packet)
 644 {
 645         struct bt_ctf_stream *stream = NULL, *writer_stream = NULL;
 646         enum bt_component_status ret = BT_COMPONENT_STATUS_OK;
 647         int int_ret;
 648
 649         stream = bt_ctf_packet_get_stream(packet);
 650         assert(stream);
 651
 652         writer_stream = get_writer_stream(writer_component, packet, stream);
 653         if (!writer_stream) {
 654                 BT_LOGE_STR("Failed to get writer_stream.");
 655                 goto error;
 656         }
 657         BT_PUT(stream);
 658
 659         int_ret = ctf_stream_copy_packet_context(
 660                         writer_component->err, packet, writer_stream);
 661         if (int_ret < 0) {
 662                 BT_LOGE_STR("Failed to copy packet_context.");
 663                 goto error;
 664         }
 665
 666         ret = ctf_stream_copy_packet_header(writer_component->err,
 667                         packet, writer_stream);
 668         if (ret != 0) {
 669                 BT_LOGE_STR("Failed to copy packet_header.");
 670                 goto error;
 671         }
 672
 673         goto end;
 674
 675 error:
 676         ret = BT_COMPONENT_STATUS_ERROR;
 677 end:
 678         bt_put(writer_stream);
 679         bt_put(stream);
 680         return ret;
 681 }
 682
 683 BT_HIDDEN
 684 enum bt_component_status writer_close_packet(
 685                 struct writer_component *writer_component,
 686                 struct bt_ctf_packet *packet)
 687 {
 688         struct bt_ctf_stream *stream = NULL, *writer_stream = NULL;
 689         enum bt_component_status ret;
 690
 691         stream = bt_ctf_packet_get_stream(packet);
 692         assert(stream);
 693
 694         writer_stream = lookup_stream(writer_component, stream);
 695         if (!writer_stream) {
 696                 BT_LOGE_STR("Failed to find existing stream.");
 697                 goto error;
 698         }
 699         BT_PUT(stream);
 700
 701         bt_get(writer_stream);
 702
 703         ret = bt_ctf_stream_flush(writer_stream);
 704         if (ret) {
 705                 BT_LOGE_STR("Failed to flush stream.");
 706                 goto error;
 707         }
 708
 709         BT_PUT(writer_stream);
 710
 711         ret = BT_COMPONENT_STATUS_OK;
 712         goto end;
 713
 714 error:
 715         ret = BT_COMPONENT_STATUS_ERROR;
 716 end:
 717         bt_put(writer_stream);
 718         bt_put(stream);
 719         return ret;
 720 }
 721
 722 BT_HIDDEN
 723 enum bt_component_status writer_output_event(
 724                 struct writer_component *writer_component,
 725                 struct bt_ctf_event *event)
 726 {
 727         enum bt_component_status ret;
 728         struct bt_ctf_event_class *event_class = NULL, *writer_event_class = NULL;
 729         struct bt_ctf_stream *stream = NULL, *writer_stream = NULL;
 730         struct bt_ctf_stream_class *stream_class = NULL, *writer_stream_class = NULL;
 731         struct bt_ctf_event *writer_event = NULL;
 732         int int_ret;
 733
 734         event_class = bt_ctf_event_get_class(event);
 735         assert(event_class);
 736
 737         stream = bt_ctf_event_get_stream(event);
 738         assert(stream);
 739
 740         writer_stream = lookup_stream(writer_component, stream);
 741         if (!writer_stream || !bt_get(writer_stream)) {
 742                 BT_LOGE_STR("Failed for find existing stream.");
 743                 goto error;
 744         }
 745
 746         stream_class = bt_ctf_event_class_get_stream_class(event_class);
 747         assert(stream_class);
 748
 749         writer_stream_class = lookup_stream_class(writer_component, stream_class);
 750         if (!writer_stream_class || !bt_get(writer_stream_class)) {
 751                 BT_LOGE_STR("Failed to find existing stream_class.");
 752                 goto error;
 753         }
 754
 755         writer_event_class = get_event_class(writer_component,
 756                         writer_stream_class, event_class);
 757         if (!writer_event_class) {
 758                 writer_event_class = ctf_copy_event_class(writer_component->err,
 759                                 event_class);
 760                 if (!writer_event_class) {
 761                         BT_LOGE_STR("Failed to copy event_class.");
 762                         goto error;
 763                 }
 764                 int_ret = bt_ctf_stream_class_add_event_class(
 765                                 writer_stream_class, writer_event_class);
 766                 if (int_ret) {
 767                         BT_LOGE("Failed to add event_class: event_name=\"%s\"",
 768                                         bt_ctf_event_class_get_name(event_class));
 769                         goto error;
 770                 }
 771         }
 772
 773         writer_event = ctf_copy_event(writer_component->err, event,
 774                         writer_event_class, true);
 775         if (!writer_event) {
 776                 BT_LOGE("Failed to copy event: event_class=\"%s\"",
 777                                 bt_ctf_event_class_get_name(writer_event_class));
 778                 goto error;
 779         }
 780
 781         int_ret = bt_ctf_stream_append_event(writer_stream, writer_event);
 782         if (int_ret < 0) {
 783                 BT_LOGE("Failed to append event: event_class=\"%s\"",
 784                                 bt_ctf_event_class_get_name(writer_event_class));
 785                 goto error;
 786         }
 787
 788         ret = BT_COMPONENT_STATUS_OK;
 789         goto end;
 790
 791 error:
 792         ret = BT_COMPONENT_STATUS_ERROR;
 793 end:
 794         bt_put(writer_event);
 795         bt_put(writer_event_class);
 796         bt_put(writer_stream_class);
 797         bt_put(stream_class);
 798         bt_put(writer_stream);
 799         bt_put(stream);
 800         bt_put(event_class);
 801         return ret;
 802 }