src/journal-remote/journal-remote-parse.c

   1 /* SPDX-License-Identifier: LGPL-2.1+ */
   2 /***
   3   This file is part of systemd.
   4
   5   Copyright 2014 Zbigniew Jędrzejewski-Szmek
   6 ***/
   7
   8 #include "alloc-util.h"
   9 #include "fd-util.h"
  10 #include "journal-remote-parse.h"
  11 #include "journald-native.h"
  12 #include "parse-util.h"
  13 #include "string-util.h"
  14
  15 void source_free(RemoteSource *source) {
  16         if (!source)
  17                 return;
  18
  19         journal_importer_cleanup(&source->importer);
  20
  21         log_debug("Writer ref count %i", source->writer->n_ref);
  22         writer_unref(source->writer);
  23
  24         sd_event_source_unref(source->event);
  25         sd_event_source_unref(source->buffer_event);
  26
  27         free(source);
  28 }
  29
  30 /**
  31  * Initialize zero-filled source with given values. On success, takes
  32  * ownership of fd, name, and writer, otherwise does not touch them.
  33  */
  34 RemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *writer) {
  35
  36         RemoteSource *source;
  37
  38         log_debug("Creating source for %sfd:%d (%s)",
  39                   passive_fd ? "passive " : "", fd, name);
  40
  41         assert(fd >= 0);
  42
  43         source = new0(RemoteSource, 1);
  44         if (!source)
  45                 return NULL;
  46
  47         source->importer.fd = fd;
  48         source->importer.passive_fd = passive_fd;
  49         source->importer.name = name;
  50
  51         source->writer = writer;
  52
  53         return source;
  54 }
  55
  56 int process_source(RemoteSource *source, bool compress, bool seal) {
  57         int r;
  58
  59         assert(source);
  60         assert(source->writer);
  61
  62         r = journal_importer_process_data(&source->importer);
  63         if (r <= 0)
  64                 return r;
  65
  66         /* We have a full event */
  67         log_trace("Received full event from source@%p fd:%d (%s)",
  68                   source, source->importer.fd, source->importer.name);
  69
  70         if (source->importer.iovw.count == 0) {
  71                 log_warning("Entry with no payload, skipping");
  72                 goto freeing;
  73         }
  74
  75         assert(source->importer.iovw.iovec);
  76
  77         r = writer_write(source->writer, &source->importer.iovw, &source->importer.ts, compress, seal);
  78         if (r < 0)
  79                 log_error_errno(r, "Failed to write entry of %zu bytes: %m",
  80                                 iovw_size(&source->importer.iovw));
  81         else
  82                 r = 1;
  83
  84  freeing:
  85         journal_importer_drop_iovw(&source->importer);
  86         return r;
  87 }