Merge pull request #10 from peroxyacyl/2022.02.24-deflate-influx1

deflate line protocol on influx1

Author: peroxyacyl

src/influxdb-cpp-rest/deflate.cpp

@@ -0,0 +1,55 @@
+#include <cpprest/http_client.h>
+#include <cpprest/json.h>
+#include <cpprest/rawptrstream.h>
+#include <fmt/ostream.h>
+#include <rx.hpp>
+#include <atomic>
+#include <chrono>
+
+using namespace web::http::client;
+using namespace web::http;
+using namespace utility;
+using namespace web::http::compression::builtin;
+
+namespace influxdb {
+    namespace utility {
+        int compress(std::shared_ptr<fmt::MemoryWriter> const& w, std::vector<uint8_t> &compression_buffer) {
+            auto c = make_compressor(algorithm::GZIP);
+            const int rawdata_size = w->size();
+            const uint8_t* rawdata_ptr =
+                reinterpret_cast<const uint8_t*>(w->data());
+            int rawdata_cursor = 0;
+
+            int compressed_size = 0;
+            bool done = false;
+
+            compression_buffer.resize(rawdata_size);
+            while (!done)
+            {
+                size_t used = 0;
+                if (compressed_size >= rawdata_size * 16)
+                {
+                    throw std::runtime_error(
+                        "gzip data is 16 times larger than raw data. abort "
+                        "compression");
+                }
+                if (compressed_size == compression_buffer.size())
+                {
+                    compression_buffer.resize(compressed_size * 2);
+                }
+                const int got = c->compress(
+                    rawdata_ptr + rawdata_cursor,   // rawdata ptr
+                    rawdata_size - rawdata_cursor,  // remaining rawdata size
+                    compression_buffer.data() +
+                        compressed_size,  // buffer ptr to append
+                    compression_buffer.size() -
+                        compressed_size,  // buffer available size
+                    web::http::compression::operation_hint::is_last, used, done);
+                compressed_size += got;
+                rawdata_cursor += used;
+            }
+            compression_buffer.resize(compressed_size);
+            return compressed_size;
+        }
+    }
+}

src/influxdb-cpp-rest/deflate.h

@@ -0,0 +1,12 @@
+#pragma once
+
+#include <vector>
+#include <fmt/ostream.h>
+
+namespace influxdb {
+    namespace utility {
+        int compress(std::shared_ptr<fmt::MemoryWriter> const &w, std::vector<uint8_t> &compression_buffer);
+    }
+}
+
+

src/influxdb-cpp-rest/influxdb2_simple_async_api.cpp

@@ -1,6 +1,7 @@
 #include "influxdb2_simple_async_api.h"
 #include "influxdb_line.h"
 #include "input_sanitizer.h"
+#include "deflate.h"
 
 #include <cpprest/http_client.h>
 #include <cpprest/json.h>
@@ -91,42 +92,8 @@ struct simple_db::impl
                         {
                           try
                           {
-                            auto c = make_compressor(algorithm::GZIP);
-
-                            const int rawdata_size = w->size();
-                            const uint8_t* rawdata_ptr =
-                                reinterpret_cast<const uint8_t*>(w->data());
-                            int rawdata_cursor = 0;
-
                             std::vector<uint8_t> compression_buffer;
-                            int compressed_size = 0;
-                            bool done = false;
-
-                            compression_buffer.resize(rawdata_size);
-                            while (!done)
-                            {
-                              size_t used = 0;
-                              if (compressed_size >= rawdata_size * 16)
-                              {
-                                throw std::runtime_error(
-                                    "gzip data is 16 times larger than raw data. abort "
-                                    "compression");
-                              }
-                              if (compressed_size == compression_buffer.size())
-                              {
-                                compression_buffer.resize(compressed_size * 2);
-                              }
-                              const int got = c->compress(
-                                  rawdata_ptr + rawdata_cursor,   // rawdata ptr
-                                  rawdata_size - rawdata_cursor,  // remaining rawdata size
-                                  compression_buffer.data() +
-                                      compressed_size,  // buffer ptr to append
-                                  compression_buffer.size() -
-                                      compressed_size,  // buffer available size
-                                  web::http::compression::operation_hint::is_last, used, done);
-                              compressed_size += got;
-                              rawdata_cursor += used;
-                            }
+                            compress(w, compression_buffer);
 
                             http_request request;
                             request.set_request_uri(uri_with_db);
@@ -136,7 +103,7 @@ struct simple_db::impl
 
                             request.set_body(
                                 concurrency::streams::rawptr_stream<uint8_t>::open_istream(
-                                    compression_buffer.data(), compressed_size));
+                                    compression_buffer.data(), compression_buffer.size()));
                             auto response = client.request(request);
                             try
                             {
@@ -304,7 +271,7 @@ simple_db::simple_db(std::string const& url, std::string const& org, std::string
                      std::string const& token, const int duration_seconds,
                      const int shard_duration_seconds, const int window_max_lines,
                      const int window_max_ms)
-    : influxdb::async_api::simple_db(url, bucket, window_max_lines, window_max_ms),
+    : influxdb::async_api::simple_db(url, bucket, window_max_lines, window_max_ms, true),
       pimpl(std::make_unique<impl>(url, org, bucket, token, duration_seconds,
                                    shard_duration_seconds, window_max_lines, window_max_ms))
 {

src/influxdb-cpp-rest/influxdb_raw_db.cpp

@@ -7,13 +7,17 @@
 //
 
 #include "influxdb_raw_db.h"
+#include "deflate.h"
 
+#include <fmt/ostream.h>
 #include <cpprest/streams.h>
 #include <cpprest/http_client.h>
+#include <cpprest/rawptrstream.h>
 
 using namespace utility;
 using namespace web;
 using namespace web::http;
+using namespace web::http::compression::builtin;
 
 namespace {
     inline void throw_response(http_response const& response) {
@@ -26,9 +30,9 @@ namespace {
 
     inline http_request request_from(
             uri const& uri_with_db,
-            std::string const& lines,
             std::string const& username,
             std::string const& password,
+            bool deflate,
             std::string const& retention_policy = "",
             web::http::method const& m = methods::POST
     ) {
@@ -58,16 +62,14 @@ namespace {
                 )
             ;
         }
-
-        request.set_body(lines);
-
+        
         return request;
     }
 }
 
-influxdb::raw::db::db(string_t const & url, string_t const & name)
+influxdb::raw::db::db(string_t const & url, string_t const & name, bool deflate)
     :
-    client(url)
+    client(url), deflate(deflate)
 {
     uri_builder builder(client.base_uri());
     builder.append(U("/write"));
@@ -82,9 +84,9 @@ void influxdb::raw::db::post(string_t const & query)
     builder.append_query(U("q"), query);
 
     // synchronous for now
-    auto response = client.request(
-        request_from(builder.to_string(), "", username, password, retention_policy)
-    );
+    auto request = request_from(builder.to_string(), username, password, deflate, retention_policy);
+    request.set_body("");
+    auto response = client.request(request);
 
     try {
         response.wait();
@@ -102,10 +104,10 @@ string_t influxdb::raw::db::get(string_t const & query)
 
     builder.append_query(U("q"), query);
 
+    auto request = request_from(builder.to_string(), username, password, deflate, retention_policy);
+    request.set_body("");
     // synchronous for now
-    auto response = client.request(
-        request_from(builder.to_string(), "", username, password, retention_policy)
-    );
+    auto response = client.request(request);
 
     try {
         response.wait();
@@ -123,9 +125,19 @@ string_t influxdb::raw::db::get(string_t const & query)
     }
 }
 
-void influxdb::raw::db::insert(std::string const & lines)
+void influxdb::raw::db::insert(std::shared_ptr<fmt::MemoryWriter> const& lines)
 {
-    auto response = client.request(request_from(uri_with_db, lines, username, password, retention_policy));
+    auto request = request_from(uri_with_db, username, password, deflate, retention_policy);
+    std::vector<uint8_t> buffer; // needs to live until sending the request
+    if (lines->size() > 0 && deflate) {
+        int size = influxdb::utility::compress(lines, buffer);
+        request.headers().add(header_names::content_encoding, algorithm::GZIP);
+        request.set_body(concurrency::streams::rawptr_stream<uint8_t>::open_istream(
+                                buffer.data(), size));
+    } else {
+        request.set_body(lines->str());
+    }
+    auto response = client.request(request);
 
     try {
         response.wait();
@@ -138,7 +150,7 @@ void influxdb::raw::db::insert(std::string const & lines)
 }
 
 // synchronous for now
-void influxdb::raw::db::insert_async(std::string const & lines)
+void influxdb::raw::db::insert_async(std::shared_ptr<fmt::MemoryWriter> const& lines)
 {
     insert(lines);
 }

src/influxdb-cpp-rest/influxdb_raw_db.h

@@ -6,6 +6,7 @@
 #pragma once
 
 #include <cpprest/http_client.h>
+#include <fmt/ostream.h>
 #include <string>
 
 using utility::string_t;
@@ -21,9 +22,10 @@ namespace influxdb {
             std::string username;
             std::string password;
             std::string retention_policy;
+            bool deflate;
 
         public:
-            db(string_t const& url, string_t const& name);
+            db(string_t const& url, string_t const& name, bool deflate);
 
             /// post queries
             void post(string_t const& query);
@@ -32,10 +34,10 @@ namespace influxdb {
             string_t get(string_t const& query);
 
             /// post measurements
-            void insert(std::string const& lines);
+            void insert(std::shared_ptr<fmt::MemoryWriter> const& lines);
 
             /// post measurements and do not wait
-            void insert_async(std::string const& lines);
+            void insert_async(std::shared_ptr<fmt::MemoryWriter> const& lines);
 
             /// set username & password for basic authentication
             void with_authentication(std::string const& username, std::string const& password);

src/influxdb-cpp-rest/influxdb_raw_db_utf8.cpp

@@ -17,18 +17,18 @@ struct influxdb::raw::db_utf8::impl {
     db db_utf16;
 
 public:
-    impl(std::string const& url,std::string const& name)
+    impl(std::string const& url,std::string const& name, bool deflate)
         :
 #ifndef _MSC_VER
-        db_utf16(url, name)
+        db_utf16(url, name, deflate)
 #else
-        db_utf16(conversions::utf8_to_utf16(url), conversions::utf8_to_utf16(name))
+        db_utf16(conversions::utf8_to_utf16(url), conversions::utf8_to_utf16(name), deflate)
 #endif
     {}
 };
 
-influxdb::raw::db_utf8::db_utf8(std::string const & url, std::string const& name) :
-    pimpl(std::make_unique<impl>(url, name))
+influxdb::raw::db_utf8::db_utf8(std::string const & url, std::string const& name, bool deflate) :
+    pimpl(std::make_unique<impl>(url, name, deflate))
 {
 }
 
@@ -56,11 +56,11 @@ std::string influxdb::raw::db_utf8::get(std::string const& query) {
 #endif
 }
 
-void influxdb::raw::db_utf8::insert(std::string const & lines) {
+void influxdb::raw::db_utf8::insert(std::shared_ptr<fmt::MemoryWriter> const& lines) {
     pimpl->db_utf16.insert(lines);
 }
 
-void influxdb::raw::db_utf8::insert_async(std::string const & lines)
+void influxdb::raw::db_utf8::insert_async(std::shared_ptr<fmt::MemoryWriter> const& lines)
 {
     pimpl->db_utf16.insert_async(lines);
 }

src/influxdb-cpp-rest/influxdb_raw_db_utf8.h

@@ -5,6 +5,7 @@
 
 #pragma once
 
+#include <fmt/ostream.h>
 #include <string>
 #include <memory>
 
@@ -15,7 +16,7 @@ namespace influxdb {
             std::unique_ptr<impl> pimpl;
 
         public:
-            db_utf8(std::string const& url, std::string const& name);
+            db_utf8(std::string const& url, std::string const& name, bool deflate);
             ~db_utf8();
 
             /// post queries
@@ -25,10 +26,10 @@ namespace influxdb {
             std::string get(std::string const& query);
 
             /// post measurements
-            void insert(std::string const& lines);
+            void insert(std::shared_ptr<fmt::MemoryWriter> const& lines);
 
             /// post measurements without waiting for an answer
-            void insert_async(std::string const& lines);
+            void insert_async(std::shared_ptr<fmt::MemoryWriter> const& lines);
 
             /// set username & password for basic authentication
             void with_authentication(std::string const& username, std::string const& password);

src/influxdb-cpp-rest/influxdb_simple_api.cpp

@@ -17,16 +17,16 @@ struct influxdb::api::simple_db::impl {
     std::string name;
     influxdb::raw::db_utf8 db;
 
-    impl(std::string const& url, std::string const& name) :
-        db(url, name),
+    impl(std::string const& url, std::string const& name, bool deflate) :
+        db(url, name, deflate),
         name(name)
     {
         throw_on_invalid_identifier(name);
     }
 };
 
-influxdb::api::simple_db::simple_db(std::string const& url, std::string const& name) :
-    pimpl(std::make_unique<impl>(url, name))
+influxdb::api::simple_db::simple_db(std::string const& url, std::string const& name, bool deflate) :
+    pimpl(std::make_unique<impl>(url, name, deflate))
 {
 }
 
@@ -46,7 +46,9 @@ void influxdb::api::simple_db::drop()
 
 void influxdb::api::simple_db::insert(line const & lines)
 {
-    pimpl->db.insert(lines.get());
+    auto w = std::make_shared<fmt::MemoryWriter>();
+    *w << lines.get();
+    pimpl->db.insert(w);
 }
 
 void influxdb::api::simple_db::with_authentication(std::string const& username, std::string const& password)

src/influxdb-cpp-rest/influxdb_simple_api.h

@@ -19,7 +19,7 @@ namespace influxdb {
             std::unique_ptr<impl> pimpl;
 
         public:
-            simple_db(std::string const& url, std::string const& name);
+            simple_db(std::string const& url, std::string const& name, bool deflate);
             ~simple_db();
 
         public: