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

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