src/logger/reader_common.c

   1 #include "reader_common.h"
   2
   3 #include "logger_internal.h"
   4 #include "subreader_file.h"
   5 #include "subreader_metrics.h"
   6
   7 #include <unistd.h>
   8
   9 static void close_reader_fd_entity(struct fd_entity *fde, struct epoll_metadata *emd)
  10 {
  11         if (fde->fd < 0)
  12                 return;
  13
  14         remove_fd_entity(emd, fde);
  15         close(fde->fd);
  16 }
  17
  18 static void reader_deinit_common(struct reader_common *reader)
  19 {
  20         list_clear_custom(&reader->subs, NULL, subreader_free);
  21
  22         close_reader_fd_entity(&reader->fd_entity_sink,   &reader->server->epoll_common);
  23         close_reader_fd_entity(&reader->fd_entity_source, &reader->server->epoll_common);
  24 }
  25
  26 int reader_apply_log_to_subs(struct reader_common *reader, const struct dlogutil_entry *de)
  27 {
  28         const int ret = list_foreach_ret(reader->subs, (void *) de, subreader_apply_log);
  29         return ret;
  30 }
  31
  32 void reader_common_init(struct reader_common *reader, struct logger *server)
  33 {
  34         reader->server = server;
  35         init_fd_entity(&reader->fd_entity_sink  , NULL, NULL);
  36         init_fd_entity(&reader->fd_entity_source, NULL, NULL);
  37 }
  38
  39 void subreader_free(void *_sub, void *userdata)
  40 {
  41         struct subreader_common *const sub = (struct subreader_common *) _sub;
  42         assert(sub);
  43         assert(userdata == NULL);
  44
  45         sub->sub_destroy(sub->sub_userdata);
  46         if (sub->filter)
  47                 log_filter_free(sub->filter);
  48         free(sub->sub_userdata);
  49 }
  50
  51 int subreader_apply_log(void *_sub, void *userdata)
  52 {
  53         struct subreader_common *const sub = (struct subreader_common *) _sub;
  54         assert(sub);
  55
  56         const struct dlogutil_entry *const due = (const struct dlogutil_entry *) userdata;
  57         assert(due);
  58
  59         if (!log_should_print_line(sub->filter, due))
  60                 return 0;
  61
  62         return sub->sub_apply_log(sub, due);
  63 }
  64
  65 int subreader_flush(void *_sub, void *userdata)
  66 {
  67         struct subreader_common *const sub = (struct subreader_common *) _sub;
  68         assert(sub);
  69         assert(sub->sub_flush);
  70
  71         struct subreader_flush_args *const args = (struct subreader_flush_args *) userdata;
  72
  73         return sub->sub_flush(sub->sub_userdata, args->ts, args->flush_time);
  74 }
  75
  76 void reader_add_sub(struct reader_common *reader, void *sub)
  77 {
  78         list_add(&reader->subs, sub);
  79 }
  80
  81 void dispatch_event_reader(struct logger *server, struct epoll_event *event, void *userdata)
  82 {
  83         struct reader_common *const reader = (struct reader_common *) userdata;
  84         assert(reader);
  85
  86         if (event->events & (EPOLLHUP | EPOLLERR))
  87                 goto reader_free;
  88
  89         struct now_t now;
  90         int r = get_now(&now);
  91         if (r < 0)
  92                 goto reader_free;
  93
  94         r = reader->service_reader(reader, now);
  95         if (r != 0) {
  96                 /* This can happen for example if a pipe gets clogged,
  97                  * and is not necessarily a reason to remove the reader. */
  98         }
  99
 100         return;
 101
 102 reader_free:
 103         reader_free(reader);
 104 }
 105
 106 int reader_flush(struct reader_common *reader, struct timespec now_mono, int flush)
 107 {
 108         return list_foreach_ret(reader->subs, &(struct subreader_flush_args) {
 109                 .ts = now_mono,
 110                 .flush_time = flush,
 111         }, subreader_flush);
 112 }
 113
 114 void reader_free(struct reader_common *reader)
 115 {
 116         if (!reader)
 117                 return;
 118
 119         reader->free_reader(reader);
 120         reader_deinit_common(reader);
 121
 122         free(reader);
 123 }
 124
 125 void reader_free_ptr(void *reader)
 126 {
 127         assert(reader); // to be used with attribute cleanup
 128         reader_free(* ((struct reader_common **) reader));
 129 }