4 files changed, 157 insertions, 0 deletions
diff --git a/libnetdata/buffered_reader/Makefile.am b/libnetdata/buffered_reader/Makefile.am
new file mode 100644
index 00000000..161784b8
--- /dev/null
+++ b/libnetdata/buffered_reader/Makefile.am
@@ -0,0 +1,8 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+AUTOMAKE_OPTIONS = subdir-objects
+MAINTAINERCLEANFILES = $(srcdir)/Makefile.in
+
+dist_noinst_DATA = \
+    README.md \
+    $(NULL)
diff --git a/libnetdata/buffered_reader/README.md b/libnetdata/buffered_reader/README.md
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/libnetdata/buffered_reader/README.md
diff --git a/libnetdata/buffered_reader/buffered_reader.c b/libnetdata/buffered_reader/buffered_reader.c
new file mode 100644
index 00000000..7cd17abf
--- /dev/null
+++ b/libnetdata/buffered_reader/buffered_reader.c
@@ -0,0 +1,3 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "../libnetdata.h"
diff --git a/libnetdata/buffered_reader/buffered_reader.h b/libnetdata/buffered_reader/buffered_reader.h
new file mode 100644
index 00000000..4db57cd2
--- /dev/null
+++ b/libnetdata/buffered_reader/buffered_reader.h
@@ -0,0 +1,146 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "../libnetdata.h"
+
+#ifndef NETDATA_BUFFERED_READER_H
+#define NETDATA_BUFFERED_READER_H
+
+struct buffered_reader {
+    ssize_t read_len;
+    ssize_t pos;
+    char read_buffer[PLUGINSD_LINE_MAX + 1];
+};
+
+static inline void buffered_reader_init(struct buffered_reader *reader) {
+    reader->read_buffer[0] = '\0';
+    reader->read_len = 0;
+    reader->pos = 0;
+}
+
+typedef enum {
+    BUFFERED_READER_READ_OK = 0,
+    BUFFERED_READER_READ_FAILED = -1,
+    BUFFERED_READER_READ_BUFFER_FULL = -2,
+    BUFFERED_READER_READ_POLLERR = -3,
+    BUFFERED_READER_READ_POLLHUP = -4,
+    BUFFERED_READER_READ_POLLNVAL = -5,
+    BUFFERED_READER_READ_POLL_UNKNOWN = -6,
+    BUFFERED_READER_READ_POLL_TIMEOUT = -7,
+    BUFFERED_READER_READ_POLL_FAILED = -8,
+} buffered_reader_ret_t;
+
+
+static inline buffered_reader_ret_t buffered_reader_read(struct buffered_reader *reader, int fd) {
+#ifdef NETDATA_INTERNAL_CHECKS
+    if(reader->read_buffer[reader->read_len] != '\0')
+        fatal("read_buffer does not start with zero");
+#endif
+
+    char *read_at = reader->read_buffer + reader->read_len;
+    ssize_t remaining = sizeof(reader->read_buffer) - reader->read_len - 1;
+
+    if(unlikely(remaining <= 0))
+        return BUFFERED_READER_READ_BUFFER_FULL;
+
+    ssize_t bytes_read = read(fd, read_at, remaining);
+    if(unlikely(bytes_read <= 0))
+        return BUFFERED_READER_READ_FAILED;
+
+    reader->read_len += bytes_read;
+    reader->read_buffer[reader->read_len] = '\0';
+
+    return BUFFERED_READER_READ_OK;
+}
+
+static inline buffered_reader_ret_t buffered_reader_read_timeout(struct buffered_reader *reader, int fd, int timeout_ms, bool log_error) {
+    errno = 0;
+    struct pollfd fds[1];
+
+    fds[0].fd = fd;
+    fds[0].events = POLLIN;
+
+    int ret = poll(fds, 1, timeout_ms);
+
+    if (ret > 0) {
+        /* There is data to read */
+        if (fds[0].revents & POLLIN)
+            return buffered_reader_read(reader, fd);
+
+        else if(fds[0].revents & POLLERR) {
+            if(log_error)
+                netdata_log_error("PARSER: read failed: POLLERR.");
+            return BUFFERED_READER_READ_POLLERR;
+        }
+        else if(fds[0].revents & POLLHUP) {
+            if(log_error)
+                netdata_log_error("PARSER: read failed: POLLHUP.");
+            return BUFFERED_READER_READ_POLLHUP;
+        }
+        else if(fds[0].revents & POLLNVAL) {
+            if(log_error)
+                netdata_log_error("PARSER: read failed: POLLNVAL.");
+            return BUFFERED_READER_READ_POLLNVAL;
+        }
+
+        if(log_error)
+            netdata_log_error("PARSER: poll() returned positive number, but POLLIN|POLLERR|POLLHUP|POLLNVAL are not set.");
+        return BUFFERED_READER_READ_POLL_UNKNOWN;
+    }
+    else if (ret == 0) {
+        if(log_error)
+            netdata_log_error("PARSER: timeout while waiting for data.");
+        return BUFFERED_READER_READ_POLL_TIMEOUT;
+    }
+
+    if(log_error)
+        netdata_log_error("PARSER: poll() failed with code %d.", ret);
+    return BUFFERED_READER_READ_POLL_FAILED;
+}
+
+/* Produce a full line if one exists, statefully return where we start next time.
+ * When we hit the end of the buffer with a partial line move it to the beginning for the next fill.
+ */
+static inline bool buffered_reader_next_line(struct buffered_reader *reader, BUFFER *dst) {
+    buffer_need_bytes(dst, reader->read_len - reader->pos + 2);
+
+    size_t start = reader->pos;
+
+    char *ss = &reader->read_buffer[start];
+    char *se = &reader->read_buffer[reader->read_len];
+    char *ds = &dst->buffer[dst->len];
+    char *de = &ds[dst->size - dst->len - 2];
+
+    if(ss >= se) {
+        *ds = '\0';
+        reader->pos = 0;
+        reader->read_len = 0;
+        reader->read_buffer[reader->read_len] = '\0';
+        return false;
+    }
+
+    // copy all bytes to buffer
+    while(ss < se && ds < de && *ss != '\n') {
+        *ds++ = *ss++;
+        dst->len++;
+    }
+
+    // if we have a newline, return the buffer
+    if(ss < se && ds < de && *ss == '\n') {
+        // newline found in the r->read_buffer
+
+        *ds++ = *ss++; // copy the newline too
+        dst->len++;
+
+        *ds = '\0';
+
+        reader->pos = ss - reader->read_buffer;
+        return true;
+    }
+
+    reader->pos = 0;
+    reader->read_len = 0;
+    reader->read_buffer[reader->read_len] = '\0';
+    return false;
+}
+
+#endif //NETDATA_BUFFERED_READER_H