Branch data     Line data    Source code

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