Move export format parsing from src/journal-remote/ to src/basic/

No functional change.

11 files changed, 582 insertions, 513 deletions

diff --git a/Makefile.am b/Makefile.am
index c45755f36e..64343b6b26 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -960,7 +960,9 @@ libbasic_la_SOURCES = \
 	src/basic/format-util.h \
 	src/basic/nss-util.h \
 	src/basic/khash.h \
-	src/basic/khash.c
+	src/basic/khash.c \
+	src/basic/journal-importer.h \
+	src/basic/journal-importer.c
 
 nodist_libbasic_la_SOURCES = \
 	src/basic/errno-from-name.h \
diff --git a/src/basic/journal-importer.c b/src/basic/journal-importer.c
new file mode 100644
index 0000000000..4c13e46a49
--- /dev/null
+++ b/src/basic/journal-importer.c
@@ -0,0 +1,481 @@
+/***
+  This file is part of systemd.
+
+  Copyright 2014 Zbigniew Jędrzejewski-Szmek
+
+  systemd is free software; you can redistribute it and/or modify it
+  under the terms of the GNU Lesser General Public License as published by
+  the Free Software Foundation; either version 2.1 of the License, or
+  (at your option) any later version.
+
+  systemd is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+  Lesser General Public License for more details.
+
+  You should have received a copy of the GNU Lesser General Public License
+  along with systemd; If not, see <http://www.gnu.org/licenses/>.
+***/
+
+#include <unistd.h>
+
+#include "alloc-util.h"
+#include "journal-importer.h"
+#include "fd-util.h"
+#include "parse-util.h"
+#include "string-util.h"
+
+enum {
+        IMPORTER_STATE_LINE = 0,    /* waiting to read, or reading line */
+        IMPORTER_STATE_DATA_START,  /* reading binary data header */
+        IMPORTER_STATE_DATA,        /* reading binary data */
+        IMPORTER_STATE_DATA_FINISH, /* expecting newline */
+        IMPORTER_STATE_EOF,         /* done */
+};
+
+static int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len) {
+        if (!GREEDY_REALLOC(iovw->iovec, iovw->size_bytes, iovw->count + 1))
+                return log_oom();
+
+        iovw->iovec[iovw->count++] = (struct iovec) {data, len};
+        return 0;
+}
+
+static void iovw_free_contents(struct iovec_wrapper *iovw) {
+        iovw->iovec = mfree(iovw->iovec);
+        iovw->size_bytes = iovw->count = 0;
+}
+
+static void iovw_rebase(struct iovec_wrapper *iovw, char *old, char *new) {
+        size_t i;
+
+        for (i = 0; i < iovw->count; i++)
+                iovw->iovec[i].iov_base = (char*) iovw->iovec[i].iov_base - old + new;
+}
+
+size_t iovw_size(struct iovec_wrapper *iovw) {
+        size_t n = 0, i;
+
+        for (i = 0; i < iovw->count; i++)
+                n += iovw->iovec[i].iov_len;
+
+        return n;
+}
+
+void journal_importer_cleanup(JournalImporter *imp) {
+        if (imp->fd >= 0 && !imp->passive_fd) {
+                log_debug("Closing %s (fd=%d)", imp->name ?: "importer", imp->fd);
+                safe_close(imp->fd);
+        }
+
+        free(imp->buf);
+        iovw_free_contents(&imp->iovw);
+}
+
+static char* realloc_buffer(JournalImporter *imp, size_t size) {
+        char *b, *old = imp->buf;
+
+        b = GREEDY_REALLOC(imp->buf, imp->size, size);
+        if (!b)
+                return NULL;
+
+        iovw_rebase(&imp->iovw, old, imp->buf);
+
+        return b;
+}
+
+static int get_line(JournalImporter *imp, char **line, size_t *size) {
+        ssize_t n;
+        char *c = NULL;
+
+        assert(imp);
+        assert(imp->state == IMPORTER_STATE_LINE);
+        assert(imp->offset <= imp->filled);
+        assert(imp->filled <= imp->size);
+        assert(imp->buf == NULL || imp->size > 0);
+        assert(imp->fd >= 0);
+
+        for (;;) {
+                if (imp->buf) {
+                        size_t start = MAX(imp->scanned, imp->offset);
+
+                        c = memchr(imp->buf + start, '\n',
+                                   imp->filled - start);
+                        if (c != NULL)
+                                break;
+                }
+
+                imp->scanned = imp->filled;
+                if (imp->scanned >= DATA_SIZE_MAX) {
+                        log_error("Entry is bigger than %u bytes.", DATA_SIZE_MAX);
+                        return -E2BIG;
+                }
+
+                if (imp->passive_fd)
+                        /* we have to wait for some data to come to us */
+                        return -EAGAIN;
+
+                /* We know that imp->filled is at most DATA_SIZE_MAX, so if
+                   we reallocate it, we'll increase the size at least a bit. */
+                assert_cc(DATA_SIZE_MAX < ENTRY_SIZE_MAX);
+                if (imp->size - imp->filled < LINE_CHUNK &&
+                    !realloc_buffer(imp, MIN(imp->filled + LINE_CHUNK, ENTRY_SIZE_MAX)))
+                                return log_oom();
+
+                assert(imp->buf);
+                assert(imp->size - imp->filled >= LINE_CHUNK ||
+                       imp->size == ENTRY_SIZE_MAX);
+
+                n = read(imp->fd,
+                         imp->buf + imp->filled,
+                         imp->size - imp->filled);
+                if (n < 0) {
+                        if (errno != EAGAIN)
+                                log_error_errno(errno, "read(%d, ..., %zu): %m",
+                                                imp->fd,
+                                                imp->size - imp->filled);
+                        return -errno;
+                } else if (n == 0)
+                        return 0;
+
+                imp->filled += n;
+        }
+
+        *line = imp->buf + imp->offset;
+        *size = c + 1 - imp->buf - imp->offset;
+        imp->offset += *size;
+
+        return 1;
+}
+
+static int fill_fixed_size(JournalImporter *imp, void **data, size_t size) {
+
+        assert(imp);
+        assert(imp->state == IMPORTER_STATE_DATA_START ||
+               imp->state == IMPORTER_STATE_DATA ||
+               imp->state == IMPORTER_STATE_DATA_FINISH);
+        assert(size <= DATA_SIZE_MAX);
+        assert(imp->offset <= imp->filled);
+        assert(imp->filled <= imp->size);
+        assert(imp->buf != NULL || imp->size == 0);
+        assert(imp->buf == NULL || imp->size > 0);
+        assert(imp->fd >= 0);
+        assert(data);
+
+        while (imp->filled - imp->offset < size) {
+                int n;
+
+                if (imp->passive_fd)
+                        /* we have to wait for some data to come to us */
+                        return -EAGAIN;
+
+                if (!realloc_buffer(imp, imp->offset + size))
+                        return log_oom();
+
+                n = read(imp->fd, imp->buf + imp->filled,
+                         imp->size - imp->filled);
+                if (n < 0) {
+                        if (errno != EAGAIN)
+                                log_error_errno(errno, "read(%d, ..., %zu): %m", imp->fd,
+                                                imp->size - imp->filled);
+                        return -errno;
+                } else if (n == 0)
+                        return 0;
+
+                imp->filled += n;
+        }
+
+        *data = imp->buf + imp->offset;
+        imp->offset += size;
+
+        return 1;
+}
+
+static int get_data_size(JournalImporter *imp) {
+        int r;
+        void *data;
+
+        assert(imp);
+        assert(imp->state == IMPORTER_STATE_DATA_START);
+        assert(imp->data_size == 0);
+
+        r = fill_fixed_size(imp, &data, sizeof(uint64_t));
+        if (r <= 0)
+                return r;
+
+        imp->data_size = le64toh( *(uint64_t *) data );
+        if (imp->data_size > DATA_SIZE_MAX) {
+                log_error("Stream declares field with size %zu > DATA_SIZE_MAX = %u",
+                          imp->data_size, DATA_SIZE_MAX);
+                return -EINVAL;
+        }
+        if (imp->data_size == 0)
+                log_warning("Binary field with zero length");
+
+        return 1;
+}
+
+static int get_data_data(JournalImporter *imp, void **data) {
+        int r;
+
+        assert(imp);
+        assert(data);
+        assert(imp->state == IMPORTER_STATE_DATA);
+
+        r = fill_fixed_size(imp, data, imp->data_size);
+        if (r <= 0)
+                return r;
+
+        return 1;
+}
+
+static int get_data_newline(JournalImporter *imp) {
+        int r;
+        char *data;
+
+        assert(imp);
+        assert(imp->state == IMPORTER_STATE_DATA_FINISH);
+
+        r = fill_fixed_size(imp, (void**) &data, 1);
+        if (r <= 0)
+                return r;
+
+        assert(data);
+        if (*data != '\n') {
+                log_error("expected newline, got '%c'", *data);
+                return -EINVAL;
+        }
+
+        return 1;
+}
+
+static int process_dunder(JournalImporter *imp, char *line, size_t n) {
+        const char *timestamp;
+        int r;
+
+        assert(line);
+        assert(n > 0);
+        assert(line[n-1] == '\n');
+
+        /* XXX: is it worth to support timestamps in extended format?
+         * We don't produce them, but who knows... */
+
+        timestamp = startswith(line, "__CURSOR=");
+        if (timestamp)
+                /* ignore __CURSOR */
+                return 1;
+
+        timestamp = startswith(line, "__REALTIME_TIMESTAMP=");
+        if (timestamp) {
+                long long unsigned x;
+                line[n-1] = '\0';
+                r = safe_atollu(timestamp, &x);
+                if (r < 0)
+                        log_warning("Failed to parse __REALTIME_TIMESTAMP: '%s'", timestamp);
+                else
+                        imp->ts.realtime = x;
+                return r < 0 ? r : 1;
+        }
+
+        timestamp = startswith(line, "__MONOTONIC_TIMESTAMP=");
+        if (timestamp) {
+                long long unsigned x;
+                line[n-1] = '\0';
+                r = safe_atollu(timestamp, &x);
+                if (r < 0)
+                        log_warning("Failed to parse __MONOTONIC_TIMESTAMP: '%s'", timestamp);
+                else
+                        imp->ts.monotonic = x;
+                return r < 0 ? r : 1;
+        }
+
+        timestamp = startswith(line, "__");
+        if (timestamp) {
+                log_notice("Unknown dunder line %s", line);
+                return 1;
+        }
+
+        /* no dunder */
+        return 0;
+}
+
+int journal_importer_process_data(JournalImporter *imp) {
+        int r;
+
+        switch(imp->state) {
+        case IMPORTER_STATE_LINE: {
+                char *line, *sep;
+                size_t n = 0;
+
+                assert(imp->data_size == 0);
+
+                r = get_line(imp, &line, &n);
+                if (r < 0)
+                        return r;
+                if (r == 0) {
+                        imp->state = IMPORTER_STATE_EOF;
+                        return r;
+                }
+                assert(n > 0);
+                assert(line[n-1] == '\n');
+
+                if (n == 1) {
+                        log_trace("Received empty line, event is ready");
+                        return 1;
+                }
+
+                r = process_dunder(imp, line, n);
+                if (r != 0)
+                        return r < 0 ? r : 0;
+
+                /* MESSAGE=xxx\n
+                   or
+                   COREDUMP\n
+                   LLLLLLLL0011223344...\n
+                */
+                sep = memchr(line, '=', n);
+                if (sep) {
+                        /* chomp newline */
+                        n--;
+
+                        r = iovw_put(&imp->iovw, line, n);
+                        if (r < 0)
+                                return r;
+                } else {
+                        /* replace \n with = */
+                        line[n-1] = '=';
+
+                        imp->field_len = n;
+                        imp->state = IMPORTER_STATE_DATA_START;
+
+                        /* we cannot put the field in iovec until we have all data */
+                }
+
+                log_trace("Received: %.*s (%s)", (int) n, line, sep ? "text" : "binary");
+
+                return 0; /* continue */
+        }
+
+        case IMPORTER_STATE_DATA_START:
+                assert(imp->data_size == 0);
+
+                r = get_data_size(imp);
+                // log_debug("get_data_size() -> %d", r);
+                if (r < 0)
+                        return r;
+                if (r == 0) {
+                        imp->state = IMPORTER_STATE_EOF;
+                        return 0;
+                }
+
+                imp->state = imp->data_size > 0 ?
+                        IMPORTER_STATE_DATA : IMPORTER_STATE_DATA_FINISH;
+
+                return 0; /* continue */
+
+        case IMPORTER_STATE_DATA: {
+                void *data;
+                char *field;
+
+                assert(imp->data_size > 0);
+
+                r = get_data_data(imp, &data);
+                // log_debug("get_data_data() -> %d", r);
+                if (r < 0)
+                        return r;
+                if (r == 0) {
+                        imp->state = IMPORTER_STATE_EOF;
+                        return 0;
+                }
+
+                assert(data);
+
+                field = (char*) data - sizeof(uint64_t) - imp->field_len;
+                memmove(field + sizeof(uint64_t), field, imp->field_len);
+
+                r = iovw_put(&imp->iovw, field + sizeof(uint64_t), imp->field_len + imp->data_size);
+                if (r < 0)
+                        return r;
+
+                imp->state = IMPORTER_STATE_DATA_FINISH;
+
+                return 0; /* continue */
+        }
+
+        case IMPORTER_STATE_DATA_FINISH:
+                r = get_data_newline(imp);
+                // log_debug("get_data_newline() -> %d", r);
+                if (r < 0)
+                        return r;
+                if (r == 0) {
+                        imp->state = IMPORTER_STATE_EOF;
+                        return 0;
+                }
+
+                imp->data_size = 0;
+                imp->state = IMPORTER_STATE_LINE;
+
+                return 0; /* continue */
+        default:
+                assert_not_reached("wtf?");
+        }
+}
+
+int journal_importer_push_data(JournalImporter *imp, const char *data, size_t size) {
+        assert(imp);
+        assert(imp->state != IMPORTER_STATE_EOF);
+
+        if (!realloc_buffer(imp, imp->filled + size)) {
+                log_error("Failed to store received data of size %zu "
+                          "(in addition to existing %zu bytes with %zu filled): %s",
+                          size, imp->size, imp->filled, strerror(ENOMEM));
+                return -ENOMEM;
+        }
+
+        memcpy(imp->buf + imp->filled, data, size);
+        imp->filled += size;
+
+        return 0;
+}
+
+void journal_importer_drop_iovw(JournalImporter *imp) {
+        size_t remain, target;
+
+        /* This function drops processed data that along with the iovw that points at it */
+
+        iovw_free_contents(&imp->iovw);
+
+        /* possibly reset buffer position */
+        remain = imp->filled - imp->offset;
+
+        if (remain == 0) /* no brainer */
+                imp->offset = imp->scanned = imp->filled = 0;
+        else if (imp->offset > imp->size - imp->filled &&
+                 imp->offset > remain) {
+                memcpy(imp->buf, imp->buf + imp->offset, remain);
+                imp->offset = imp->scanned = 0;
+                imp->filled = remain;
+        }
+
+        target = imp->size;
+        while (target > 16 * LINE_CHUNK && imp->filled < target / 2)
+                target /= 2;
+        if (target < imp->size) {
+                char *tmp;
+
+                tmp = realloc(imp->buf, target);
+                if (!tmp)
+                        log_warning("Failed to reallocate buffer to (smaller) size %zu",
+                                    target);
+                else {
+                        log_debug("Reallocated buffer from %zu to %zu bytes",
+                                  imp->size, target);
+                        imp->buf = tmp;
+                        imp->size = target;
+                }
+        }
+}
+
+bool journal_importer_eof(const JournalImporter *imp) {
+        return imp->state == IMPORTER_STATE_EOF;
+}
diff --git a/src/basic/journal-importer.h b/src/basic/journal-importer.h
new file mode 100644
index 0000000000..b3e308dd6d
--- /dev/null
+++ b/src/basic/journal-importer.h
@@ -0,0 +1,70 @@
+/***
+  This file is part of systemd.
+
+  Copyright 2016 Zbigniew Jędrzejewski-Szmek
+
+  systemd is free software; you can redistribute it and/or modify it
+  under the terms of the GNU Lesser General Public License as published by
+  the Free Software Foundation; either version 2.1 of the License, or
+  (at your option) any later version.
+
+  systemd is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+  Lesser General Public License for more details.
+
+  You should have received a copy of the GNU Lesser General Public License
+  along with systemd; If not, see <http://www.gnu.org/licenses/>.
+***/
+
+#pragma once
+
+#include <stddef.h>
+#include <stdbool.h>
+#include <sys/uio.h>
+
+#include "time-util.h"
+
+/* Make sure not to make this smaller than the maximum coredump size.
+ * See COREDUMP_MAX in coredump.c */
+#define ENTRY_SIZE_MAX (1024*1024*770u)
+#define DATA_SIZE_MAX (1024*1024*768u)
+#define LINE_CHUNK 8*1024u
+
+struct iovec_wrapper {
+        struct iovec *iovec;
+        size_t size_bytes;
+        size_t count;
+};
+
+size_t iovw_size(struct iovec_wrapper *iovw);
+
+typedef struct JournalImporter {
+        int fd;
+        bool passive_fd;
+        char *name;
+
+        char *buf;
+        size_t size;       /* total size of the buffer */
+        size_t offset;     /* offset to the beginning of live data in the buffer */
+        size_t scanned;    /* number of bytes since the beginning of data without a newline */
+        size_t filled;     /* total number of bytes in the buffer */
+
+        size_t field_len;  /* used for binary fields: the field name length */
+        size_t data_size;  /* and the size of the binary data chunk being processed */
+
+        struct iovec_wrapper iovw;
+
+        int state;
+        dual_timestamp ts;
+} JournalImporter;
+
+void journal_importer_cleanup(JournalImporter *);
+int journal_importer_process_data(JournalImporter *);
+int journal_importer_push_data(JournalImporter *, const char *data, size_t size);
+void journal_importer_drop_iovw(JournalImporter *);
+bool journal_importer_eof(const JournalImporter *);
+
+static inline size_t journal_importer_bytes_remaining(const JournalImporter *imp) {
+        return imp->filled;
+}
diff --git a/src/coredump/coredump.c b/src/coredump/coredump.c
index 4d66b59fd6..2d9819e5b0 100644
--- a/src/coredump/coredump.c
+++ b/src/coredump/coredump.c
@@ -47,7 +47,7 @@
 #include "fileio.h"
 #include "fs-util.h"
 #include "io-util.h"
-#include "journald-native.h"
+#include "journal-importer.h"
 #include "log.h"
 #include "macro.h"
 #include "missing.h"
diff --git a/src/journal-remote/journal-remote-parse.c b/src/journal-remote/journal-remote-parse.c
index 9ba9ee3fc0..79afe6604c 100644
--- a/src/journal-remote/journal-remote-parse.c
+++ b/src/journal-remote/journal-remote-parse.c
@@ -24,20 +24,11 @@
 #include "parse-util.h"
 #include "string-util.h"
 
-#define LINE_CHUNK 8*1024u
-
 void source_free(RemoteSource *source) {
         if (!source)
                 return;
 
-        if (source->fd >= 0 && !source->passive_fd) {
-                log_debug("Closing fd:%d (%s)", source->fd, source->name);
-                safe_close(source->fd);
-        }
-
-        free(source->name);
-        free(source->buf);
-        iovw_free_contents(&source->iovw);
+        journal_importer_cleanup(&source->importer);
 
         log_debug("Writer ref count %i", source->writer->n_ref);
         writer_unref(source->writer);
@@ -65,442 +56,44 @@ RemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *writer) {
         if (!source)
                 return NULL;
 
-        source->fd = fd;
-        source->passive_fd = passive_fd;
-        source->name = name;
+        source->importer.fd = fd;
+        source->importer.passive_fd = passive_fd;
+        source->importer.name = name;
+
         source->writer = writer;
 
         return source;
 }
 
-static char* realloc_buffer(RemoteSource *source, size_t size) {
-        char *b, *old = source->buf;
-
-        b = GREEDY_REALLOC(source->buf, source->size, size);
-        if (!b)
-                return NULL;
-
-        iovw_rebase(&source->iovw, old, source->buf);
-
-        return b;
-}
-
-static int get_line(RemoteSource *source, char **line, size_t *size) {
-        ssize_t n;
-        char *c = NULL;
-
-        assert(source);
-        assert(source->state == STATE_LINE);
-        assert(source->offset <= source->filled);
-        assert(source->filled <= source->size);
-        assert(source->buf == NULL || source->size > 0);
-        assert(source->fd >= 0);
-
-        for (;;) {
-                if (source->buf) {
-                        size_t start = MAX(source->scanned, source->offset);
-
-                        c = memchr(source->buf + start, '\n',
-                                   source->filled - start);
-                        if (c != NULL)
-                                break;
-                }
-
-                source->scanned = source->filled;
-                if (source->scanned >= DATA_SIZE_MAX) {
-                        log_error("Entry is bigger than %u bytes.", DATA_SIZE_MAX);
-                        return -E2BIG;
-                }
-
-                if (source->passive_fd)
-                        /* we have to wait for some data to come to us */
-                        return -EAGAIN;
-
-                /* We know that source->filled is at most DATA_SIZE_MAX, so if
-                   we reallocate it, we'll increase the size at least a bit. */
-                assert_cc(DATA_SIZE_MAX < ENTRY_SIZE_MAX);
-                if (source->size - source->filled < LINE_CHUNK &&
-                    !realloc_buffer(source, MIN(source->filled + LINE_CHUNK, ENTRY_SIZE_MAX)))
-                                return log_oom();
-
-                assert(source->buf);
-                assert(source->size - source->filled >= LINE_CHUNK ||
-                       source->size == ENTRY_SIZE_MAX);
-
-                n = read(source->fd,
-                         source->buf + source->filled,
-                         source->size - source->filled);
-                if (n < 0) {
-                        if (errno != EAGAIN)
-                                log_error_errno(errno, "read(%d, ..., %zu): %m",
-                                                source->fd,
-                                                source->size - source->filled);
-                        return -errno;
-                } else if (n == 0)
-                        return 0;
-
-                source->filled += n;
-        }
-
-        *line = source->buf + source->offset;
-        *size = c + 1 - source->buf - source->offset;
-        source->offset += *size;
-
-        return 1;
-}
-
-int push_data(RemoteSource *source, const char *data, size_t size) {
-        assert(source);
-        assert(source->state != STATE_EOF);
-
-        if (!realloc_buffer(source, source->filled + size)) {
-                log_error("Failed to store received data of size %zu "
-                          "(in addition to existing %zu bytes with %zu filled): %s",
-                          size, source->size, source->filled, strerror(ENOMEM));
-                return -ENOMEM;
-        }
-
-        memcpy(source->buf + source->filled, data, size);
-        source->filled += size;
-
-        return 0;
-}
-
-static int fill_fixed_size(RemoteSource *source, void **data, size_t size) {
-
-        assert(source);
-        assert(source->state == STATE_DATA_START ||
-               source->state == STATE_DATA ||
-               source->state == STATE_DATA_FINISH);
-        assert(size <= DATA_SIZE_MAX);
-        assert(source->offset <= source->filled);
-        assert(source->filled <= source->size);
-        assert(source->buf != NULL || source->size == 0);
-        assert(source->buf == NULL || source->size > 0);
-        assert(source->fd >= 0);
-        assert(data);
-
-        while (source->filled - source->offset < size) {
-                int n;
-
-                if (source->passive_fd)
-                        /* we have to wait for some data to come to us */
-                        return -EAGAIN;
-
-                if (!realloc_buffer(source, source->offset + size))
-                        return log_oom();
-
-                n = read(source->fd, source->buf + source->filled,
-                         source->size - source->filled);
-                if (n < 0) {
-                        if (errno != EAGAIN)
-                                log_error_errno(errno, "read(%d, ..., %zu): %m", source->fd,
-                                                source->size - source->filled);
-                        return -errno;
-                } else if (n == 0)
-                        return 0;
-
-                source->filled += n;
-        }
-
-        *data = source->buf + source->offset;
-        source->offset += size;
-
-        return 1;
-}
-
-static int get_data_size(RemoteSource *source) {
-        int r;
-        void *data;
-
-        assert(source);
-        assert(source->state == STATE_DATA_START);
-        assert(source->data_size == 0);
-
-        r = fill_fixed_size(source, &data, sizeof(uint64_t));
-        if (r <= 0)
-                return r;
-
-        source->data_size = le64toh( *(uint64_t *) data );
-        if (source->data_size > DATA_SIZE_MAX) {
-                log_error("Stream declares field with size %zu > DATA_SIZE_MAX = %u",
-                          source->data_size, DATA_SIZE_MAX);
-                return -EINVAL;
-        }
-        if (source->data_size == 0)
-                log_warning("Binary field with zero length");
-
-        return 1;
-}
-
-static int get_data_data(RemoteSource *source, void **data) {
-        int r;
-
-        assert(source);
-        assert(data);
-        assert(source->state == STATE_DATA);
-
-        r = fill_fixed_size(source, data, source->data_size);
-        if (r <= 0)
-                return r;
-
-        return 1;
-}
-
-static int get_data_newline(RemoteSource *source) {
-        int r;
-        char *data;
-
-        assert(source);
-        assert(source->state == STATE_DATA_FINISH);
-
-        r = fill_fixed_size(source, (void**) &data, 1);
-        if (r <= 0)
-                return r;
-
-        assert(data);
-        if (*data != '\n') {
-                log_error("expected newline, got '%c'", *data);
-                return -EINVAL;
-        }
-
-        return 1;
-}
-
-static int process_dunder(RemoteSource *source, char *line, size_t n) {
-        const char *timestamp;
-        int r;
-
-        assert(line);
-        assert(n > 0);
-        assert(line[n-1] == '\n');
-
-        /* XXX: is it worth to support timestamps in extended format?
-         * We don't produce them, but who knows... */
-
-        timestamp = startswith(line, "__CURSOR=");
-        if (timestamp)
-                /* ignore __CURSOR */
-                return 1;
-
-        timestamp = startswith(line, "__REALTIME_TIMESTAMP=");
-        if (timestamp) {
-                long long unsigned x;
-                line[n-1] = '\0';
-                r = safe_atollu(timestamp, &x);
-                if (r < 0)
-                        log_warning("Failed to parse __REALTIME_TIMESTAMP: '%s'", timestamp);
-                else
-                        source->ts.realtime = x;
-                return r < 0 ? r : 1;
-        }
-
-        timestamp = startswith(line, "__MONOTONIC_TIMESTAMP=");
-        if (timestamp) {
-                long long unsigned x;
-                line[n-1] = '\0';
-                r = safe_atollu(timestamp, &x);
-                if (r < 0)
-                        log_warning("Failed to parse __MONOTONIC_TIMESTAMP: '%s'", timestamp);
-                else
-                        source->ts.monotonic = x;
-                return r < 0 ? r : 1;
-        }
-
-        timestamp = startswith(line, "__");
-        if (timestamp) {
-                log_notice("Unknown dunder line %s", line);
-                return 1;
-        }
-
-        /* no dunder */
-        return 0;
-}
-
-static int process_data(RemoteSource *source) {
-        int r;
-
-        switch(source->state) {
-        case STATE_LINE: {
-                char *line, *sep;
-                size_t n = 0;
-
-                assert(source->data_size == 0);
-
-                r = get_line(source, &line, &n);
-                if (r < 0)
-                        return r;
-                if (r == 0) {
-                        source->state = STATE_EOF;
-                        return r;
-                }
-                assert(n > 0);
-                assert(line[n-1] == '\n');
-
-                if (n == 1) {
-                        log_trace("Received empty line, event is ready");
-                        return 1;
-                }
-
-                r = process_dunder(source, line, n);
-                if (r != 0)
-                        return r < 0 ? r : 0;
-
-                /* MESSAGE=xxx\n
-                   or
-                   COREDUMP\n
-                   LLLLLLLL0011223344...\n
-                */
-                sep = memchr(line, '=', n);
-                if (sep) {
-                        /* chomp newline */
-                        n--;
-
-                        r = iovw_put(&source->iovw, line, n);
-                        if (r < 0)
-                                return r;
-                } else {
-                        /* replace \n with = */
-                        line[n-1] = '=';
-
-                        source->field_len = n;
-                        source->state = STATE_DATA_START;
-
-                        /* we cannot put the field in iovec until we have all data */
-                }
-
-                log_trace("Received: %.*s (%s)", (int) n, line, sep ? "text" : "binary");
-
-                return 0; /* continue */
-        }
-
-        case STATE_DATA_START:
-                assert(source->data_size == 0);
-
-                r = get_data_size(source);
-                // log_debug("get_data_size() -> %d", r);
-                if (r < 0)
-                        return r;
-                if (r == 0) {
-                        source->state = STATE_EOF;
-                        return 0;
-                }
-
-                source->state = source->data_size > 0 ?
-                        STATE_DATA : STATE_DATA_FINISH;
-
-                return 0; /* continue */
-
-        case STATE_DATA: {
-                void *data;
-                char *field;
-
-                assert(source->data_size > 0);
-
-                r = get_data_data(source, &data);
-                // log_debug("get_data_data() -> %d", r);
-                if (r < 0)
-                        return r;
-                if (r == 0) {
-                        source->state = STATE_EOF;
-                        return 0;
-                }
-
-                assert(data);
-
-                field = (char*) data - sizeof(uint64_t) - source->field_len;
-                memmove(field + sizeof(uint64_t), field, source->field_len);
-
-                r = iovw_put(&source->iovw, field + sizeof(uint64_t), source->field_len + source->data_size);
-                if (r < 0)
-                        return r;
-
-                source->state = STATE_DATA_FINISH;
-
-                return 0; /* continue */
-        }
-
-        case STATE_DATA_FINISH:
-                r = get_data_newline(source);
-                // log_debug("get_data_newline() -> %d", r);
-                if (r < 0)
-                        return r;
-                if (r == 0) {
-                        source->state = STATE_EOF;
-                        return 0;
-                }
-
-                source->data_size = 0;
-                source->state = STATE_LINE;
-
-                return 0; /* continue */
-        default:
-                assert_not_reached("wtf?");
-        }
-}
-
 int process_source(RemoteSource *source, bool compress, bool seal) {
-        size_t remain, target;
         int r;
 
         assert(source);
         assert(source->writer);
 
-        r = process_data(source);
+        r = journal_importer_process_data(&source->importer);
         if (r <= 0)
                 return r;
 
         /* We have a full event */
         log_trace("Received full event from source@%p fd:%d (%s)",
-                  source, source->fd, source->name);
+                  source, source->importer.fd, source->importer.name);
 
-        if (!source->iovw.count) {
+        if (source->importer.iovw.count == 0) {
                 log_warning("Entry with no payload, skipping");
                 goto freeing;
         }
 
-        assert(source->iovw.iovec);
-        assert(source->iovw.count);
+        assert(source->importer.iovw.iovec);
 
-        r = writer_write(source->writer, &source->iovw, &source->ts, compress, seal);
+        r = writer_write(source->writer, &source->importer.iovw, &source->importer.ts, compress, seal);
         if (r < 0)
                 log_error_errno(r, "Failed to write entry of %zu bytes: %m",
-                                iovw_size(&source->iovw));
+                                iovw_size(&source->importer.iovw));
         else
                 r = 1;
 
  freeing:
-        iovw_free_contents(&source->iovw);
-
-        /* possibly reset buffer position */
-        remain = source->filled - source->offset;
-
-        if (remain == 0) /* no brainer */
-                source->offset = source->scanned = source->filled = 0;
-        else if (source->offset > source->size - source->filled &&
-                 source->offset > remain) {
-                memcpy(source->buf, source->buf + source->offset, remain);
-                source->offset = source->scanned = 0;
-                source->filled = remain;
-        }
-
-        target = source->size;
-        while (target > 16 * LINE_CHUNK && source->filled < target / 2)
-                target /= 2;
-        if (target < source->size) {
-                char *tmp;
-
-                tmp = realloc(source->buf, target);
-                if (!tmp)
-                        log_warning("Failed to reallocate buffer to (smaller) size %zu",
-                                    target);
-                else {
-                        log_debug("Reallocated buffer from %zu to %zu bytes",
-                                  source->size, target);
-                        source->buf = tmp;
-                        source->size = target;
-                }
-        }
-
+        journal_importer_drop_iovw(&source->importer);
         return r;
 }
diff --git a/src/journal-remote/journal-remote-parse.h b/src/journal-remote/journal-remote-parse.h
index 1740a21f92..e3632528cf 100644
--- a/src/journal-remote/journal-remote-parse.h
+++ b/src/journal-remote/journal-remote-parse.h
@@ -21,34 +21,11 @@
 
 #include "sd-event.h"
 
+#include "journal-importer.h"
 #include "journal-remote-write.h"
 
-typedef enum {
-        STATE_LINE = 0,    /* waiting to read, or reading line */
-        STATE_DATA_START,  /* reading binary data header */
-        STATE_DATA,        /* reading binary data */
-        STATE_DATA_FINISH, /* expecting newline */
-        STATE_EOF,         /* done */
-} source_state;
-
 typedef struct RemoteSource {
-        char *name;
-        int fd;
-        bool passive_fd;
-
-        char *buf;
-        size_t size;       /* total size of the buffer */
-        size_t offset;     /* offset to the beginning of live data in the buffer */
-        size_t scanned;    /* number of bytes since the beginning of data without a newline */
-        size_t filled;     /* total number of bytes in the buffer */
-
-        size_t field_len;  /* used for binary fields: the field name length */
-        size_t data_size;  /* and the size of the binary data chunk being processed */
-
-        struct iovec_wrapper iovw;
-
-        source_state state;
-        dual_timestamp ts;
+        JournalImporter importer;
 
         Writer *writer;
 
@@ -57,13 +34,5 @@ typedef struct RemoteSource {
 } RemoteSource;
 
 RemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *writer);
-
-static inline size_t source_non_empty(RemoteSource *source) {
-        assert(source);
-
-        return source->filled;
-}
-
 void source_free(RemoteSource *source);
-int push_data(RemoteSource *source, const char *data, size_t size);
 int process_source(RemoteSource *source, bool compress, bool seal);
diff --git a/src/journal-remote/journal-remote-write.c b/src/journal-remote/journal-remote-write.c
index 8729372aa3..734cad333f 100644
--- a/src/journal-remote/journal-remote-write.c
+++ b/src/journal-remote/journal-remote-write.c
@@ -20,39 +20,6 @@
 #include "alloc-util.h"
 #include "journal-remote.h"
 
-int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len) {
-        if (!GREEDY_REALLOC(iovw->iovec, iovw->size_bytes, iovw->count + 1))
-                return log_oom();
-
-        iovw->iovec[iovw->count++] = (struct iovec) {data, len};
-        return 0;
-}
-
-void iovw_free_contents(struct iovec_wrapper *iovw) {
-        iovw->iovec = mfree(iovw->iovec);
-        iovw->size_bytes = iovw->count = 0;
-}
-
-size_t iovw_size(struct iovec_wrapper *iovw) {
-        size_t n = 0, i;
-
-        for (i = 0; i < iovw->count; i++)
-                n += iovw->iovec[i].iov_len;
-
-        return n;
-}
-
-void iovw_rebase(struct iovec_wrapper *iovw, char *old, char *new) {
-        size_t i;
-
-        for (i = 0; i < iovw->count; i++)
-                iovw->iovec[i].iov_base = (char*) iovw->iovec[i].iov_base - old + new;
-}
-
-/**********************************************************************
- **********************************************************************
- **********************************************************************/
-
 static int do_rotate(JournalFile **f, bool compress, bool seal) {
         int r = journal_file_rotate(f, compress, seal, NULL);
         if (r < 0) {
diff --git a/src/journal-remote/journal-remote-write.h b/src/journal-remote/journal-remote-write.h
index 53ba45fc04..e04af54e55 100644
--- a/src/journal-remote/journal-remote-write.h
+++ b/src/journal-remote/journal-remote-write.h
@@ -20,20 +20,10 @@
 ***/
 
 #include "journal-file.h"
+#include "journal-importer.h"
 
 typedef struct RemoteServer RemoteServer;
 
-struct iovec_wrapper {
-        struct iovec *iovec;
-        size_t size_bytes;
-        size_t count;
-};
-
-int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len);
-void iovw_free_contents(struct iovec_wrapper *iovw);
-size_t iovw_size(struct iovec_wrapper *iovw);
-void iovw_rebase(struct iovec_wrapper *iovw, char *old, char *new);
-
 typedef struct Writer {
         JournalFile *journal;
         JournalMetrics metrics;
diff --git a/src/journal-remote/journal-remote.c b/src/journal-remote/journal-remote.c
index d0d8d936e3..202a5a3f97 100644
--- a/src/journal-remote/journal-remote.c
+++ b/src/journal-remote/journal-remote.c
@@ -512,7 +512,8 @@ static int process_http_upload(
         if (*upload_data_size) {
                 log_trace("Received %zu bytes", *upload_data_size);
 
-                r = push_data(source, upload_data, *upload_data_size);
+                r = journal_importer_push_data(&source->importer,
+                                               upload_data, *upload_data_size);
                 if (r < 0)
                         return mhd_respond_oom(connection);
 
@@ -542,7 +543,7 @@ static int process_http_upload(
 
         /* The upload is finished */
 
-        remaining = source_non_empty(source);
+        remaining = journal_importer_bytes_remaining(&source->importer);
         if (remaining > 0) {
                 log_warning("Premature EOF byte. %zu bytes lost.", remaining);
                 return mhd_respondf(connection,
@@ -1036,19 +1037,19 @@ static int handle_raw_source(sd_event_source *event,
 
         assert(fd >= 0 && fd < (ssize_t) s->sources_size);
         source = s->sources[fd];
-        assert(source->fd == fd);
+        assert(source->importer.fd == fd);
 
         r = process_source(source, arg_compress, arg_seal);
-        if (source->state == STATE_EOF) {
+        if (journal_importer_eof(&source->importer)) {
                 size_t remaining;
 
-                log_debug("EOF reached with source fd:%d (%s)",
-                          source->fd, source->name);
+                log_debug("EOF reached with source %s (fd=%d)",
+                          source->importer.name, source->importer.fd);
 
-                remaining = source_non_empty(source);
+                remaining = journal_importer_bytes_remaining(&source->importer);
                 if (remaining > 0)
                         log_notice("Premature EOF. %zu bytes lost.", remaining);
-                remove_source(s, source->fd);
+                remove_source(s, source->importer.fd);
                 log_debug("%zu active sources remaining", s->active);
                 return 0;
         } else if (r == -E2BIG) {
@@ -1072,7 +1073,7 @@ static int dispatch_raw_source_until_block(sd_event_source *event,
         /* Make sure event stays around even if source is destroyed */
         sd_event_source_ref(event);
 
-        r = handle_raw_source(event, source->fd, EPOLLIN, server);
+        r = handle_raw_source(event, source->importer.fd, EPOLLIN, server);
         if (r != 1)
                 /* No more data for now */
                 sd_event_source_set_enabled(event, SD_EVENT_OFF);
@@ -1105,7 +1106,7 @@ static int dispatch_blocking_source_event(sd_event_source *event,
                                           void *userdata) {
         RemoteSource *source = userdata;
 
-        return handle_raw_source(event, source->fd, EPOLLIN, server);
+        return handle_raw_source(event, source->importer.fd, EPOLLIN, server);
 }
 
 static int accept_connection(const char* type, int fd,
diff --git a/src/journal/journald-native.c b/src/journal/journald-native.c
index 0a1ce205c2..3c03b83754 100644
--- a/src/journal/journald-native.c
+++ b/src/journal/journald-native.c
@@ -27,6 +27,7 @@
 #include "fd-util.h"
 #include "fs-util.h"
 #include "io-util.h"
+#include "journal-importer.h"
 #include "journald-console.h"
 #include "journald-kmsg.h"
 #include "journald-native.h"
diff --git a/src/journal/journald-native.h b/src/journal/journald-native.h
index c13b80aa4f..1ab415ac85 100644
--- a/src/journal/journald-native.h
+++ b/src/journal/journald-native.h
@@ -21,11 +21,6 @@
 
 #include "journald-server.h"
 
-/* Make sure not to make this smaller than the maximum coredump
- * size. See COREDUMP_MAX in coredump.c */
-#define ENTRY_SIZE_MAX (1024*1024*770u)
-#define DATA_SIZE_MAX (1024*1024*768u)
-
 bool valid_user_field(const char *p, size_t l, bool allow_protected);
 
 void server_process_native_message(Server *s, const void *buffer, size_t buffer_size, const struct ucred *ucred, const struct timeval *tv, const char *label, size_t label_len);