Merge branch 'release/v1.1.7'

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 1.1.7 (2021-06-30)
+
+### Bug Fixes
+
+* Fix mem leakage in static_allocator used for HTTP requests
+
 ## 1.1.6 (2021-06-29)
 
 ### Bug Fixes

CMakeLists.txt

@@ -2,7 +2,7 @@ cmake_minimum_required(VERSION 3.0 FATAL_ERROR)
 
 set(STREAMCLIENT_VERSION_MAJOR "1")
 set(STREAMCLIENT_VERSION_MINOR "1")
-set(STREAMCLIENT_VERSION_RELEASE "6")
+set(STREAMCLIENT_VERSION_RELEASE "7")
 set(STREAMCLIENT_VERSION_STRING "${STREAMCLIENT_VERSION_MAJOR}.${STREAMCLIENT_VERSION_MINOR}.${STREAMCLIENT_VERSION_RELEASE}")
 set(STREAMCLIENT_LIB_VERSION ${STREAMCLIENT_VERSION_STRING})
 mark_as_advanced(STREAMCLIENT_VERSION_MAJOR STREAMCLIENT_VERSION_MINOR STREAMCLIENT_VERSION_RELEASE STREAMCLIENT_VERSION_STRING STREAMCLIENT_LIB_VERSION)

example/http/CMakeLists.txt

@@ -2,7 +2,10 @@ cmake_minimum_required(VERSION 3.0 FATAL_ERROR)
 
 find_package(Threads)
 
-add_executable(example_client example_client.cpp)
+add_executable(example_connector example_connector.cpp)
+target_link_libraries(example_connector PRIVATE stream-client::stream-client)
+target_link_libraries(example_connector PRIVATE Threads::Threads)
 
-target_link_libraries(example_client PRIVATE stream-client::stream-client)
-target_link_libraries(example_client PRIVATE Threads::Threads)
+add_executable(example_pool example_pool.cpp)
+target_link_libraries(example_pool PRIVATE stream-client::stream-client)
+target_link_libraries(example_pool PRIVATE Threads::Threads)

example/http/example_client.cpp → example/http/example_connector.cpp

@@ -60,7 +60,7 @@ struct ParsedURI
 int main(int argc, char* argv[])
 {
     if (argc != 4) {
-        std::cerr << "Usage: client <url> <threads> <requests per thread>\n";
+        std::cerr << "Usage: " << argv[0] << " <url> <threads> <requests per thread>\n";
         return 1;
     }
     int num_threads = std::atoi(argv[2]);

example/http/example_pool.cpp

@@ -0,0 +1,108 @@
+#include "stream-client/stream-client.hpp"
+
+#include <boost/algorithm/string/case_conv.hpp>
+#include <iostream>
+#include <regex>
+
+template <typename T>
+std::vector<std::thread> start_threads(const std::string& host, std::string& port,
+                                       const boost::beast::http::request<boost::beast::http::string_body>& request,
+                                       int threads_num, int req_per_thread)
+{
+    static const auto connection_pool = std::make_shared<T>(threads_num,
+                                                            std::chrono::milliseconds(500), // idle_timeout
+                                                            host, port,
+                                                            std::chrono::milliseconds(5000), // resolve_timeout
+                                                            std::chrono::milliseconds(1000), // connect_timeout
+                                                            std::chrono::milliseconds(500), // operation_timeout
+                                                            stream_client::resolver::ip_family::ipv4);
+
+    std::vector<std::thread> threads;
+    for (int i = 0; i < threads_num; ++i) {
+        threads.emplace_back([&request, req_per_thread]() {
+            // give each thread its' own session and reuse it to send-out some requests
+            auto session = connection_pool->get_session();
+
+            boost::system::error_code err;
+            for (int j = 0; j < req_per_thread; ++j) {
+                const auto resp = session->perform(request, err);
+                if (err) {
+                    std::cout << err.message();
+                } else {
+                    std::cout << *resp;
+                }
+
+                std::this_thread::sleep_for(std::chrono::milliseconds(500));
+            }
+
+            connection_pool->return_session(std::move(session));
+        });
+    }
+    return threads;
+}
+
+struct ParsedURI
+{
+    std::string protocol;
+    std::string domain;
+    std::string port;
+    std::string resource;
+    std::string query;
+
+    ParsedURI(const std::string& url)
+    {
+        static const std::regex PARSE_URL{R"(((http|https)://)?([^/ :]+)(:(\d+))?(/([^ ?]+)?)?/?\??([^/ ]+\=[^/ ]+)?)",
+                                          std::regex_constants::ECMAScript | std::regex_constants::icase};
+
+        auto value_or = [](const std::string& value, std::string&& deflt) { return (value.empty() ? deflt : value); };
+        // Note: only "http" and "https" protocols are supported
+        std::smatch match;
+        if (std::regex_match(url, match, PARSE_URL) && match.size() == 9) {
+            domain = match[3];
+            protocol = value_or(boost::algorithm::to_lower_copy(std::string(match[2])), "http");
+            port = value_or(match[5], (protocol == "https") ? "443" : "80");
+            resource = value_or(match[6], "/");
+            query = match[8];
+        }
+    }
+};
+
+int main(int argc, char* argv[])
+{
+    if (argc != 4) {
+        std::cerr << "Usage: " << argv[0] << " <url> <threads> <requests per thread>\n";
+        return 1;
+    }
+    int num_threads = std::atoi(argv[2]);
+    int requests = std::atoi(argv[3]);
+    ParsedURI uri(argv[1]);
+
+    boost::beast::http::request<boost::beast::http::string_body> req;
+    req.version(11);
+    req.method(boost::beast::http::verb::post);
+    req.target(uri.resource);
+    req.set(boost::beast::http::field::host, uri.domain);
+    req.body() = "{test}";
+    req.set(boost::beast::http::field::content_type, "application/json");
+    req.set(boost::beast::http::field::accept, "*/*");
+    req.set(boost::beast::http::field::user_agent, "beast/stream_client");
+    req.prepare_payload();
+
+    std::cout << std::boolalpha;
+    std::cout << req;
+
+    std::vector<std::thread> threads;
+    if (uri.protocol == "http") {
+        threads = start_threads<stream_client::connector::http_pool>(uri.domain, uri.port, req, num_threads, requests);
+    } else if (uri.protocol == "https") {
+        threads = start_threads<stream_client::connector::https_pool>(uri.domain, uri.port, req, num_threads, requests);
+    } else {
+        std::cerr << "protocol should be either 'http' or 'https'" << std::endl;
+        return 1;
+    }
+    for (auto& t : threads) {
+        t.join();
+    }
+
+    return 0;
+}

include/stream-client/stream/detail/static_allocator.hpp

@@ -24,7 +24,7 @@ class static_pool
 
     void destroy()
     {
-        if (refs_--) {
+        if (--refs_) {
             return;
         }
         this->~static_pool();