dropping inserts on connection errors

Author: d-led

src/influxdb-cpp-rest/influxdb_raw_db.cpp

@@ -74,10 +74,15 @@ void influxdb::raw::db::post(string_t const & query)
     // synchronous for now
     auto response = client.request(
         request_from(builder.to_string(), "", username, password)
-    ).get();
+    );
 
-    if (response.status_code() != status_codes::OK) {
-        throw_response(response);
+    try {
+        response.wait();
+        if (response.get().status_code() != status_codes::OK) {
+            throw_response(response.get());
+        }
+    } catch (const std::exception& e) {
+        throw std::runtime_error(e.what());
     }
 }
 
@@ -90,29 +95,42 @@ string_t influxdb::raw::db::get(string_t const & query)
     // synchronous for now
     auto response = client.request(
         request_from(builder.to_string(), "", username, password)
-    ).get();
-    if (response.status_code() == status_codes::OK)
-    {
-        return response.extract_string().get();
-    }
-    else
-    {
-        throw_response(response);
-        return string_t();
+    );
+
+    try {
+        response.wait();
+        if (response.get().status_code() == status_codes::OK)
+        {
+            return response.get().extract_string().get();
+        }
+        else
+        {
+            throw_response(response.get());
+            return string_t();
+        }
+    } catch (const std::exception& e) {
+        throw std::runtime_error(e.what());
     }
 }
 
 void influxdb::raw::db::insert(std::string const & lines)
 {
-    auto response = client.request(request_from(uri_with_db, lines, username, password)).get();
-    if (!(response.status_code() == status_codes::OK || response.status_code() == status_codes::NoContent)) {
-        throw_response(response);
+    auto response = client.request(request_from(uri_with_db, lines, username, password));
+
+    try {
+        response.wait();
+        if (!(response.get().status_code() == status_codes::OK || response.get().status_code() == status_codes::NoContent)) {
+            throw_response(response.get());
+        }
+    } catch (const std::exception& e) {
+        throw std::runtime_error(e.what());
     }
 }
 
+// synchronous for now
 void influxdb::raw::db::insert_async(std::string const & lines)
 {
-    client.request(request_from(uri_with_db, lines, username, password));
+    insert(lines);
 }
 
 void influxdb::raw::db::with_authentication(std::string const& username, std::string const& password)

src/influxdb-cpp-rest/influxdb_simple_async_api.cpp

@@ -67,7 +67,11 @@ struct influxdb::async_api::simple_db::impl {
                         .observe_on(rxcpp::synchronize_new_thread())
                         .subscribe([this](std::shared_ptr<fmt::MemoryWriter> const& w) {
                             if (w->size() > 0u) {
-                                db.insert_async(w->str());
+                                try {
+                                    db.insert_async(w->str());
+                                } catch (const std::runtime_error& e) {
+                                    std::cerr << "async_api::insert failed: " << e.what() << " -> Dropping " << w->size() << " bytes" << std::endl;
+                                }
                             }
                         },
                         [](std::exception_ptr ep) {