10 files changed, 29 insertions, 25 deletions
diff --git a/src/streaming/compression_brotli.c b/src/streaming/stream-compression/brotli.c
index cf52f3bca..c2c09cdc5 100644
--- a/src/streaming/compression_brotli.c
+++ b/src/streaming/stream-compression/brotli.c
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
-#include "compression_brotli.h"
+#include "brotli.h"
 
 #ifdef ENABLE_BROTLI
 #include <brotli/encode.h>
diff --git a/src/streaming/compression_brotli.h b/src/streaming/stream-compression/brotli.h
index 4955e5a82..4955e5a82 100644
--- a/src/streaming/compression_brotli.h
+++ b/src/streaming/stream-compression/brotli.h
diff --git a/src/streaming/compression.c b/src/streaming/stream-compression/compression.c
index a94c8a0a6..3c9930656 100644
--- a/src/streaming/compression.c
+++ b/src/streaming/stream-compression/compression.c
@@ -2,18 +2,18 @@
 
 #include "compression.h"
 
-#include "compression_gzip.h"
+#include "gzip.h"
 
 #ifdef ENABLE_LZ4
-#include "compression_lz4.h"
+#include "lz4.h"
 #endif
 
 #ifdef ENABLE_ZSTD
-#include "compression_zstd.h"
+#include "zstd.h"
 #endif
 
 #ifdef ENABLE_BROTLI
-#include "compression_brotli.h"
+#include "brotli.h"
 #endif
 
 int rrdpush_compression_levels[COMPRESSION_ALGORITHM_MAX] = {
@@ -32,7 +32,7 @@ void rrdpush_parse_compression_order(struct receiver_state *rpt, const char *ord
     char *s = strdupz(order);
 
     char *words[COMPRESSION_ALGORITHM_MAX + 100] = { NULL };
-    size_t num_words = quoted_strings_splitter_pluginsd(s, words, COMPRESSION_ALGORITHM_MAX + 100);
+    size_t num_words = quoted_strings_splitter_whitespace(s, words, COMPRESSION_ALGORITHM_MAX + 100);
     size_t slot = 0;
     STREAM_CAPABILITIES added = STREAM_CAP_NONE;
     for(size_t i = 0; i < num_words && slot < COMPRESSION_ALGORITHM_MAX ;i++) {
@@ -395,21 +395,17 @@ size_t rrdpush_decompress(struct decompressor_state *state, const char *compress
 // ----------------------------------------------------------------------------
 // unit test
 
-static inline long int my_random (void) {
-    return random();
-}
-
 void unittest_generate_random_name(char *dst, size_t size) {
     if(size < 7)
         size = 7;
 
-    size_t len = 5 + my_random() % (size - 6);
+    size_t len = 5 + os_random32() % (size - 6);
 
     for(size_t i = 0; i < len ; i++) {
-        if(my_random() % 2 == 0)
-            dst[i] = 'A' + my_random() % 26;
+        if(os_random8() % 2 == 0)
+            dst[i] = 'A' + os_random8() % 26;
         else
-            dst[i] = 'a' + my_random() % 26;
+            dst[i] = 'a' + os_random8() % 26;
     }
 
     dst[len] = '\0';
@@ -423,9 +419,9 @@ void unittest_generate_message(BUFFER *wb, time_t now_s, size_t counter) {
     time_t point_end_time_s = now_s;
     time_t wall_clock_time_s = now_s;
     size_t chart_slot = counter + 1;
-    size_t dimensions = 2 + my_random() % 5;
+    size_t dimensions = 2 + os_random8() % 5;
     char chart[RRD_ID_LENGTH_MAX + 1] = "name";
-    unittest_generate_random_name(chart, 5 + my_random() % 30);
+    unittest_generate_random_name(chart, 5 + os_random8() % 30);
 
     buffer_fast_strcat(wb, PLUGINSD_KEYWORD_BEGIN_V2, sizeof(PLUGINSD_KEYWORD_BEGIN_V2) - 1);
 
@@ -451,10 +447,10 @@ void unittest_generate_message(BUFFER *wb, time_t now_s, size_t counter) {
     for(size_t d = 0; d < dimensions ;d++) {
         size_t dim_slot = d + 1;
         char dim_id[RRD_ID_LENGTH_MAX + 1] = "dimension";
-        unittest_generate_random_name(dim_id, 10 + my_random() % 20);
-        int64_t last_collected_value = (my_random() % 2 == 0) ? (int64_t)(counter + d) : (int64_t)my_random();
-        NETDATA_DOUBLE value = (my_random() % 2 == 0) ? (NETDATA_DOUBLE)my_random() / ((NETDATA_DOUBLE)my_random() + 1) : (NETDATA_DOUBLE)last_collected_value;
-        SN_FLAGS flags = (my_random() % 1000 == 0) ? SN_FLAG_NONE : SN_FLAG_NOT_ANOMALOUS;
+        unittest_generate_random_name(dim_id, 10 + os_random8() % 20);
+        int64_t last_collected_value = (os_random8() % 2 == 0) ? (int64_t)(counter + d) : (int64_t)os_random32();
+        NETDATA_DOUBLE value = (os_random8() % 2 == 0) ? (NETDATA_DOUBLE)os_random64() / ((NETDATA_DOUBLE)os_random64() + 1) : (NETDATA_DOUBLE)last_collected_value;
+        SN_FLAGS flags = (os_random16() % 1000 == 0) ? SN_FLAG_NONE : SN_FLAG_NOT_ANOMALOUS;
 
         buffer_fast_strcat(wb, PLUGINSD_KEYWORD_SET_V2, sizeof(PLUGINSD_KEYWORD_SET_V2) - 1);
 
diff --git a/src/streaming/compression.h b/src/streaming/stream-compression/compression.h
index 285fb2cf6..37f589b85 100644
--- a/src/streaming/compression.h
+++ b/src/streaming/stream-compression/compression.h
@@ -1,10 +1,10 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
-#include "rrdpush.h"
-
 #ifndef NETDATA_RRDPUSH_COMPRESSION_H
 #define NETDATA_RRDPUSH_COMPRESSION_H 1
 
+#include "libnetdata/libnetdata.h"
+
 // signature MUST end with a newline
 
 #if COMPRESSION_MAX_MSG_SIZE >= (COMPRESSION_MAX_CHUNK - COMPRESSION_MAX_OVERHEAD)
@@ -172,4 +172,12 @@ static inline size_t rrdpush_decompressor_get(struct decompressor_state *state,
 
 // ----------------------------------------------------------------------------
 
+#include "../rrdpush.h"
+
+bool rrdpush_compression_initialize(struct sender_state *s);
+bool rrdpush_decompression_initialize(struct receiver_state *rpt);
+void rrdpush_parse_compression_order(struct receiver_state *rpt, const char *order);
+void rrdpush_select_receiver_compression_algorithm(struct receiver_state *rpt);
+void rrdpush_compression_deactivate(struct sender_state *s);
+
 #endif // NETDATA_RRDPUSH_COMPRESSION_H 1
diff --git a/src/streaming/compression_gzip.c b/src/streaming/stream-compression/gzip.c
index c4ef3af05..d63e9afbe 100644
--- a/src/streaming/compression_gzip.c
+++ b/src/streaming/stream-compression/gzip.c
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
-#include "compression_gzip.h"
+#include "gzip.h"
 #include <zlib.h>
 
 void rrdpush_compressor_init_gzip(struct compressor_state *state) {
diff --git a/src/streaming/compression_gzip.h b/src/streaming/stream-compression/gzip.h
index 85f34bc6d..85f34bc6d 100644
--- a/src/streaming/compression_gzip.h
+++ b/src/streaming/stream-compression/gzip.h
diff --git a/src/streaming/compression_lz4.c b/src/streaming/stream-compression/lz4.c
index f5174134e..284192153 100644
--- a/src/streaming/compression_lz4.c
+++ b/src/streaming/stream-compression/lz4.c
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
-#include "compression_lz4.h"
+#include "lz4.h"
 
 #ifdef ENABLE_LZ4
 #include "lz4.h"
diff --git a/src/streaming/compression_lz4.h b/src/streaming/stream-compression/lz4.h
index 69f0fadcc..69f0fadcc 100644
--- a/src/streaming/compression_lz4.h
+++ b/src/streaming/stream-compression/lz4.h
diff --git a/src/streaming/compression_zstd.c b/src/streaming/stream-compression/zstd.c
index dabc044f7..0ce27c0d3 100644
--- a/src/streaming/compression_zstd.c
+++ b/src/streaming/stream-compression/zstd.c
@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
-#include "compression_zstd.h"
+#include "zstd.h"
 
 #ifdef ENABLE_ZSTD
 #include <zstd.h>
diff --git a/src/streaming/compression_zstd.h b/src/streaming/stream-compression/zstd.h
index bfabbf89d..bfabbf89d 100644
--- a/src/streaming/compression_zstd.h
+++ b/src/streaming/stream-compression/zstd.h