diff --git a/include/snowplow/configuration/emitter_configuration.cpp b/include/snowplow/configuration/emitter_configuration.cpp index 2c33714..8bd3138 100644 --- a/include/snowplow/configuration/emitter_configuration.cpp +++ b/include/snowplow/configuration/emitter_configuration.cpp @@ -35,6 +35,7 @@ void EmitterConfiguration::shared_init() { m_batch_size = SNOWPLOW_EMITTER_DEFAULT_BATCH_SIZE; m_byte_limit_get = SNOWPLOW_EMITTER_DEFAULT_BYTE_LIMIT_GET; m_byte_limit_post = SNOWPLOW_EMITTER_DEFAULT_BYTE_LIMIT_POST; + m_flush_timeout_ms = 30000; } void EmitterConfiguration::set_event_store(shared_ptr event_store) { diff --git a/include/snowplow/configuration/emitter_configuration.hpp b/include/snowplow/configuration/emitter_configuration.hpp index 1ed5dfd..d354f9e 100644 --- a/include/snowplow/configuration/emitter_configuration.hpp +++ b/include/snowplow/configuration/emitter_configuration.hpp @@ -88,14 +88,23 @@ class EmitterConfiguration { /** * @brief Set a custom retry rule for when the HTTP status code is received in emit response from Collector. - * + * * This overrides default behavior for HTTP status codes greater than 300. - * + * * @param http_status_code HTTP status code * @param retry Whether events should be retried or not */ void set_custom_retry_for_status_code(int http_status_code, bool retry); + /** + * @brief Set the maximum time flush() will wait for the event queue to drain before stopping. + * + * Pass 0 for no timeout (waits indefinitely). Undelivered events remain in SQLite for the next session. + * + * @param flush_timeout_ms Maximum wait time in milliseconds (default: 30000). + */ + void set_flush_timeout_ms(int flush_timeout_ms) { m_flush_timeout_ms = flush_timeout_ms; } + /** * @brief Get the event store. * @@ -149,17 +158,25 @@ class EmitterConfiguration { /** * @brief Get the custom retry rule settings for HTTP status codes. - * + * * @return map Map of status code –> retry or not boolean. */ map get_custom_retry_for_status_codes() const { return m_custom_retry_for_status_codes; } + /** + * @brief Get the flush timeout in milliseconds. + * + * @return int Maximum time flush() waits for the queue to drain (0 = no timeout). + */ + int get_flush_timeout_ms() const { return m_flush_timeout_ms; } + private: void shared_init(); int m_batch_size; int m_byte_limit_post; int m_byte_limit_get; + int m_flush_timeout_ms; shared_ptr m_event_store; EmitterCallback m_callback; EmitStatus m_callback_emit_status; diff --git a/include/snowplow/emitter/emitter.cpp b/include/snowplow/emitter/emitter.cpp index b50cd72..20ecb3e 100644 --- a/include/snowplow/emitter/emitter.cpp +++ b/include/snowplow/emitter/emitter.cpp @@ -25,7 +25,6 @@ using std::to_string; using std::transform; using std::equal; using std::future; -using std::this_thread::sleep_for; const int post_wrapper_bytes = 88; // "schema":"iglu:com.snowplowanalytics.snowplow/payload_data/jsonschema/1-0-4","data":[] const int post_stm_bytes = 22; // "stm":"1443452851000" @@ -62,6 +61,7 @@ Emitter::Emitter(NetworkConfiguration &network_config, const EmitterConfiguratio m_callback = emitter_config.get_request_callback(); m_callback_emit_status = emitter_config.get_request_callback_emit_status(); m_custom_retry_for_status_codes = emitter_config.get_custom_retry_for_status_codes(); + m_flush_timeout_ms = emitter_config.get_flush_timeout_ms(); } Emitter::Emitter(shared_ptr event_store, const string &uri, Method method, Protocol protocol, int batch_size, @@ -86,6 +86,7 @@ Emitter::Emitter(shared_ptr event_store, const string &uri, Method m } this->m_running = false; + this->m_flush_timeout_ms = 30000; this->m_method = method; this->m_batch_size = batch_size; this->m_byte_limit_post = byte_limit_post; @@ -109,6 +110,8 @@ void Emitter::start() { if (this->m_running) { return; // refuse to start more than once } + this->m_stop_requested = false; + this->m_flush_done = false; this->m_running = true; this->m_daemon_thread = thread(&Emitter::run, this); } @@ -117,10 +120,18 @@ void Emitter::stop() { unique_lock locker(this->m_run_check); if (this->m_running == true) { this->m_running = false; + this->m_stop_requested = true; locker.unlock(); + // Close the lost-wakeup race: by acquiring and releasing m_db_select before + // notify_all, we guarantee the daemon is either already in wait_for (will + // receive the notify) or will see m_stop_requested=true at its next pre-check. + { unique_lock db_locker(this->m_db_select); } this->m_check_db.notify_all(); this->m_daemon_thread.join(); + + // Unblock flush() if it is waiting on m_check_fin (e.g. stop() called externally) + this->m_check_fin.notify_all(); } } @@ -136,10 +147,19 @@ void Emitter::flush() { } locker_1.unlock(); + // Reset flush_done before waking the daemon so that the predicate below + // reflects only queue state observed after this flush() call. + m_flush_done = false; this->m_check_db.notify_all(); unique_lock locker_2(this->m_flush_fin); - this->m_check_fin.wait(locker_2); + if (m_flush_timeout_ms > 0) { + this->m_check_fin.wait_for(locker_2, std::chrono::milliseconds(m_flush_timeout_ms), + [this]{ return m_flush_done.load() || m_stop_requested.load(); }); + } else { + this->m_check_fin.wait(locker_2, + [this]{ return m_flush_done.load() || m_stop_requested.load(); }); + } locker_2.unlock(); this->stop(); @@ -188,17 +208,25 @@ void Emitter::run() { m_retry_delay.wont_retry_emit(); } - // sleep for the retry delay if there is one + // sleep for the retry delay if there is one — interruptible by stop() auto retry_delay = m_retry_delay.get(); if (retry_delay.count() > 0) { - sleep_for(retry_delay); + unique_lock retry_locker(m_db_select); + if (!m_stop_requested.load()) { + m_check_db.wait_for(retry_locker, retry_delay); + } } } else { + // Queue is empty: signal flush() waiters + m_flush_done = true; m_check_fin.notify_all(); - // if there are no events to send, sleep for a while + // Idle sleep — pre-check m_stop_requested so stop() calling notify_all between + // here and wait_for is guaranteed visible via the m_db_select lock-handshake in stop() unique_lock locker(m_db_select); - m_check_db.wait_for(locker, std::chrono::seconds(5)); + if (!m_stop_requested.load()) { + m_check_db.wait_for(locker, std::chrono::seconds(5)); + } } } while (is_running()); } diff --git a/include/snowplow/emitter/emitter.hpp b/include/snowplow/emitter/emitter.hpp index 06b3144..a98eed8 100644 --- a/include/snowplow/emitter/emitter.hpp +++ b/include/snowplow/emitter/emitter.hpp @@ -16,6 +16,7 @@ See the Apache License Version 2.0 for the specific language governing permissio #include #include +#include #include #include #include @@ -193,6 +194,9 @@ class Emitter { mutex m_db_select; mutex m_run_check; bool m_running; + std::atomic m_stop_requested{false}; + std::atomic m_flush_done{false}; + int m_flush_timeout_ms; EmitterCallback m_callback; EmitStatus m_callback_emit_status; map m_custom_retry_for_status_codes; diff --git a/test/configuration/emitter_configuration_test.cpp b/test/configuration/emitter_configuration_test.cpp index 6c4b92a..4034444 100644 --- a/test/configuration/emitter_configuration_test.cpp +++ b/test/configuration/emitter_configuration_test.cpp @@ -45,4 +45,12 @@ TEST_CASE("emitter configuration") { REQUIRE(raised_error); REQUIRE(emitter_config.get_custom_retry_for_status_codes().size() == 0); } + + SECTION("flush timeout getter and setter") { + auto storage = std::make_shared("test-emitter.db"); + EmitterConfiguration emitter_config(storage); + REQUIRE(emitter_config.get_flush_timeout_ms() == 30000); + emitter_config.set_flush_timeout_ms(5000); + REQUIRE(emitter_config.get_flush_timeout_ms() == 5000); + } } diff --git a/test/emitter/emitter_test.cpp b/test/emitter/emitter_test.cpp index 6d4503f..071e81e 100644 --- a/test/emitter/emitter_test.cpp +++ b/test/emitter/emitter_test.cpp @@ -374,4 +374,69 @@ TEST_CASE("emitter") { emitter.stop(); } + + SECTION("stop() on idle emitter returns promptly") { + Emitter emitter(storage, "com.acme.collector", Method::POST, Protocol::HTTP, 500, 500, 500, unique_ptr(new TestHttpClient())); + emitter.start(); + auto t_start = std::chrono::high_resolution_clock::now(); + emitter.stop(); + auto t_end = std::chrono::high_resolution_clock::now(); + double elapsed_ms = std::chrono::duration(t_end - t_start).count(); + REQUIRE(elapsed_ms < 2000); + } + + SECTION("flush() with unreachable collector returns after flush timeout") { + TestHttpClient::set_http_response_code(503); // always 503 (will retry indefinitely without timeout) + + auto test_storage = std::make_shared("test-emitter-flushtimeout.db"); + NetworkConfiguration network_config("com.acme.unreachable.collector", POST); + network_config.set_http_client(unique_ptr(new TestHttpClient())); + EmitterConfiguration emitter_config(test_storage); + emitter_config.set_flush_timeout_ms(500); + + Emitter emitter(network_config, emitter_config); + emitter.start(); + + Payload payload; + payload.add("e", "pv"); + emitter.add(payload); + + auto t_start = std::chrono::high_resolution_clock::now(); + emitter.flush(); + auto t_end = std::chrono::high_resolution_clock::now(); + double elapsed_ms = std::chrono::duration(t_end - t_start).count(); + REQUIRE(elapsed_ms < 3000); + + // Events remain in SQLite — not dropped on timeout + list remaining; + test_storage->get_all_event_rows(&remaining); + REQUIRE(remaining.size() > 0); + + TestHttpClient::reset(); + remove("test-emitter-flushtimeout.db"); + } + + SECTION("stop() during retry sleep returns promptly") { + auto test_storage = std::make_shared("test-emitter-retrystop.db"); + TestHttpClient::set_temporary_response_code(503, 20); // 20 failures to keep daemon retrying + + Emitter emitter(test_storage, "com.acme.collector", Method::POST, Protocol::HTTP, 500, 500, 500, unique_ptr(new TestHttpClient())); + emitter.start(); + + Payload payload; + payload.add("e", "pv"); + emitter.add(payload); + + // Give the daemon time to attempt the first send and enter the retry sleep + sleep_for(milliseconds(100)); + + auto t_start = std::chrono::high_resolution_clock::now(); + emitter.stop(); + auto t_end = std::chrono::high_resolution_clock::now(); + double elapsed_ms = std::chrono::duration(t_end - t_start).count(); + REQUIRE(elapsed_ms < 2000); + + TestHttpClient::reset(); + remove("test-emitter-retrystop.db"); + } }