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