diff --git a/CMakeLists.txt b/CMakeLists.txt index 1257863..82b02fd 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -53,7 +53,7 @@ set_target_properties(gtest_main PROPERTIES MSVC_RUNTIME_LIBRARY MultiThreaded$< # 'Unit_Tests_run' is the target name # 'UNIT_SOURCE' are source files with tests -file(GLOB UNIT_SOURCE test/unit_test.cpp test/unit_test.h src/*.cpp include/*.h) +file(GLOB UNIT_SOURCE test/unit_test.cpp test/unit_test.h test/exchange_test.cpp src/*.cpp include/*.h) set(UNIT_TEST notifly_unit_test) add_executable(${UNIT_TEST} ${UNIT_SOURCE}) @@ -69,6 +69,19 @@ set_target_properties(${UNIT_TEST} MSVC_RUNTIME_LIBRARY MultiThreaded$<$:Debug> ) +# Exchange example +set(EXCHANGE_EXAMPLE notifly_exchange_example) +add_executable(${EXCHANGE_EXAMPLE} example/exchange_example.cpp) +target_include_directories(${EXCHANGE_EXAMPLE} PRIVATE include) +target_compile_features(${EXCHANGE_EXAMPLE} PUBLIC cxx_std_20) +target_link_libraries(${EXCHANGE_EXAMPLE} PRIVATE ${CMAKE_THREAD_LIBS_INIT}) +set_target_properties(${EXCHANGE_EXAMPLE} + PROPERTIES + OUTPUT_NAME ${EXCHANGE_EXAMPLE} + LINKER_LANGUAGE CXX + MSVC_RUNTIME_LIBRARY MultiThreaded$<$:Debug> + ) + # C Interface example set(C_EXAMPLE notifly_c_example) add_executable(${C_EXAMPLE} example/c_example.c) diff --git a/README.md b/README.md index 97631f9..3d89436 100644 --- a/README.md +++ b/README.md @@ -89,6 +89,70 @@ notifly::default_notifly().post_notification_async(MY_NOTIFICATION_ID); Asynchronous notifications are executed in separate threads, allowing your application to continue processing without waiting for observers to complete their work. +### Waiting For A Reply + +`post_notification` is one-way. When something is expected to answer, `post_and_wait` posts a request and blocks until +the reply arrives or the timeout expires. It subscribes before posting, so a reply that comes back before the post call +returns is still caught: + +```C++ +int reply = 0; +const auto result = notifly::default_notifly().post_and_wait( + REQUEST_ID, // post this + REPLY_ID, // wait for this + 500, // timeout, milliseconds + reply, // where the payload lands (a value, or a std::tuple) + arg1, arg2); // request payload + +if (result == notifly_result::timeout) { /* nobody answered */ } +``` + +For anything more involved, `notifly::exchange` is the same machinery with the pieces exposed. Each handler returns a +`notifly_verdict` saying what the delivery was — `skip` to ignore it and stay subscribed, `keep` for one piece of a +streamed reply, `done` to end the wait: + +```C++ +notifly::exchange ex(notifly::default_notifly()); + +// Ignore the state the sender is leaving; only the one asked for ends the wait. +ex.on(STATUS_ID, [](const int state) +{ + return state == 1 ? notifly_verdict::done : notifly_verdict::skip; +}); + +notifly::default_notifly().post_notification(ENABLE_ID, 0); + +if (ex.wait(std::chrono::milliseconds(500)) < 0) { /* timed out */ } +``` + +That covers the shapes a single request/reply pair cannot: + +| Shape | How | +|---|---| +| Ignore deliveries that are not the awaited one | return `notifly_verdict::skip` | +| Several possible answers, and which one arrived matters | chain `on()` calls; `wait()` returns the notification that fired | +| A reply streamed in pieces of unstated length | return `notifly_verdict::keep`, end with `drain(quiet, deadline)` | +| Silence is the successful outcome | `silent_for(window)` | +| Nothing to post — waiting on an external event | subscribe and `wait()`, post nothing | +| One of several alternatives is a plain value, not worth a handler | `capture(id, out)` — `out` is a value or a `std::tuple` | + +`capture()` is what `post_and_wait()` is built on internally; it is also public on its own, for a branch of a +multi-alternative exchange that just needs the payload and nothing else: + +```C++ +notifly::exchange ex(notifly::default_notifly()); +int status = -1; +ex.capture(STATUS_ID, status); // instead of ex.on(STATUS_ID, [&](int v) { status = v; return notifly_verdict::done; }) +``` + +Once a handler returns `done` the exchange is complete and later deliveries are ignored, so a sender that repeats itself +cannot disturb what the winning handler stored. The destructor unsubscribes; never destroy an exchange from inside a +handler. + +Note that a notification's payload shape must match exactly, references included: `post_notification` deduces its +arguments by value, so an observer declared as `[](const std::string&)` registers a different shape than one declared +as `[](std::string)` and will not be called. + ### Avoiding Unnecessary Lookups Notifications can be posted and modified by using the unique identifier returned when add observer is called: @@ -103,11 +167,14 @@ notifly::default_notifly().remove_observer(observerId); You can also use more than one instance of NotificationCenter. Although a default notification center is provided, you can also create your own notification centers for whatever purpose you may require them for. -### Example Program +### Example Programs The included example program shows you the basics of how to use NotificationCenter. It's not intended to be sophisticated by any means, just to showcase the basics. +`example/exchange_example.cpp` covers the request/reply side: it drives a simulated device through each of the shapes in +the table above, one per numbered section. + ### Bugs I don't expect this to work flawlessly for all applications, and thread safety isn't something that I've tested diff --git a/example/exchange_example.cpp b/example/exchange_example.cpp new file mode 100644 index 0000000..fd4a160 --- /dev/null +++ b/example/exchange_example.cpp @@ -0,0 +1,259 @@ +/* + * exchange_example.cpp + * notifly + * + * Driving a request/reply protocol with notifly::exchange. + * + * post_notification() is one-way and add_observer() is open-ended, so talking + * to something that answers means hand-rolling the join: subscribe, send, wait, + * unsubscribe -- and get the ordering right, because a reply can come back + * before the send call has even returned. + * + * post_and_wait() does that for the common shape: one request, one reply, take + * the first that arrives. This example covers the shapes it cannot express, + * each one modelled on something a real device does: + * + * 1. a reply worth ignoring -- the device reports the state it is leaving + * before the state it is entering + * 2. several possible answers -- a job ends in completion, refusal or a jam, + * and the caller needs to know which + * 3. silence means success -- the device only speaks up to refuse + * 4. a reply sent in pieces -- a table whose length the protocol never states + * 5. nothing to send at all -- waiting on a person, not on a command + */ + +#include "notifly.h" + +#include +#include +#include +#include +#include + +using namespace std::chrono_literals; + +namespace +{ + enum : int + { + cmd_enable = 100, + cmd_print, + cmd_read_table, + + evt_status, + evt_job_complete, + evt_job_refused, + evt_paper_jam, + evt_table_entry, + evt_note_inserted + }; + + /// Stands in for the device: it answers commands on a thread of its own, + /// the way a real one answers over a wire. + class fake_device + { + public: + explicit fake_device(notifly& a_center) : m_center(a_center) + { + m_observers.push_back(m_center.add_observer(cmd_enable, [this](int) + { + // Real hardware announces the state it is leaving first, then + // the one it settles into. + m_center.post_notification(evt_status, 0); + std::this_thread::sleep_for(30ms); + m_center.post_notification(evt_status, 1); + })); + + // Note the by-value parameter: post_notification() deduces its + // arguments by value, and a notification's payload shape has to + // match exactly, references included -- an observer taking + // "const std::string&" registers a different shape and is never + // called. See get_type_string() in notifly.h. + m_observers.push_back(m_center.add_observer(cmd_print, [this](const std::string a_ticket) + { + std::this_thread::sleep_for(20ms); + if(a_ticket.empty()) m_center.post_notification(evt_job_refused, std::string("empty ticket")); + else m_center.post_notification(evt_job_complete, 7); + })); + + m_observers.push_back(m_center.add_observer(cmd_read_table, [this](int) + { + // The protocol never says how many entries there are; the + // device simply stops sending. + for(int i = 1; i <= 4; ++i) + { + std::this_thread::sleep_for(15ms); + m_center.post_notification(evt_table_entry, i * 5); + } + })); + } + + ~fake_device() + { + for(const int observer: m_observers) m_center.remove_observer(observer); + } + + fake_device(const fake_device&) = delete; + fake_device& operator=(const fake_device&) = delete; + + private: + notifly& m_center; + std::vector m_observers; + }; +} + +// --------------------------------------------------------------------------- +// 1. A reply worth ignoring. +// --------------------------------------------------------------------------- + +void enable_the_device(notifly& a_center) +{ + printf("\n1. enable -- ignoring the state the device is leaving\n"); + + notifly::exchange ex(a_center); + + // post_and_wait() would take that first "0" as the answer and report the + // device enabled while it is still on its way there. A verdict lets the + // handler say "not this one" and stay subscribed. + ex.on(evt_status, [](const int a_state) + { + printf(" device reports state %d\n", a_state); + return a_state == 1 ? notifly_verdict::done : notifly_verdict::skip; + }); + + a_center.post_notification(cmd_enable, 0); + + if(ex.wait(2000ms) < 0) printf(" -> timed out\n"); + else printf(" -> enabled (took %zu of the deliveries)\n", ex.accepted()); +} + +// --------------------------------------------------------------------------- +// 2. Several possible answers, and the caller needs to know which one came. +// --------------------------------------------------------------------------- + +void print_a_ticket(notifly& a_center, const std::string& a_ticket) +{ + printf("\n2. print -- three ways for one job to end\n"); + + notifly::exchange ex(a_center); + + int transaction = 0; + std::string refusal; + + ex.on(evt_job_complete, [&](const int a_transaction) + { + transaction = a_transaction; + return notifly_verdict::done; + }) + .on(evt_job_refused, [&](const std::string& a_reason) + { + refusal = a_reason; + return notifly_verdict::done; + }) + .on(evt_paper_jam, [](int) { return notifly_verdict::done; }); + + a_center.post_notification(cmd_print, a_ticket); + + // A print job runs for seconds, so it is given far longer than a command. + switch(const int fired = ex.wait(30000ms)) + { + case evt_job_complete: printf(" -> printed, transaction %d\n", transaction); break; + case evt_job_refused: printf(" -> refused: %s\n", refusal.c_str()); break; + case evt_paper_jam: printf(" -> paper jam\n"); break; + default: printf(" -> timed out (fired=%d)\n", fired); break; + } +} + +// --------------------------------------------------------------------------- +// 3. Silence means success. +// --------------------------------------------------------------------------- + +void transfer_a_template(notifly& a_center) +{ + printf("\n3. transfer -- the device only speaks up to refuse\n"); + + notifly::exchange ex(a_center); + ex.on(evt_job_refused); + + // Nothing is sent here: this stands for a transfer the device accepts + // silently. Waiting for a reply that only exists on failure would always + // time out, so the question is inverted -- did anything object? + if(ex.silent_for(200ms)) printf(" -> accepted (nothing objected)\n"); + else printf(" -> refused\n"); +} + +// --------------------------------------------------------------------------- +// 4. A reply sent in pieces. +// --------------------------------------------------------------------------- + +void read_the_note_table(notifly& a_center) +{ + printf("\n4. read table -- a reply of unstated length\n"); + + std::vector entries; + + notifly::exchange ex(a_center); + ex.on(evt_table_entry, [&](const int a_value) + { + entries.push_back(a_value); + // keep, not done: there is no way to know which entry is the last. + return notifly_verdict::keep; + }); + + a_center.post_notification(cmd_read_table, 0); + + // Ends once the device has been quiet for 100ms, or at 5s regardless. + const auto count = ex.drain(100ms, 5000ms); + + printf(" -> %zu entries:", count); + for(const int entry: entries) printf(" %d", entry); + printf("\n"); +} + +// --------------------------------------------------------------------------- +// 5. Nothing to send at all. +// --------------------------------------------------------------------------- + +void wait_for_a_note(notifly& a_center) +{ + printf("\n5. wait for a note -- no command to post\n"); + + std::string note; + + notifly::exchange ex(a_center); + ex.on(evt_note_inserted, [&](const std::string& a_note) + { + note = a_note; + return notifly_verdict::done; + }); + + // Whoever is at the machine, not a command, decides when this happens. + std::thread player([&a_center] + { + std::this_thread::sleep_for(50ms); + a_center.post_notification(evt_note_inserted, std::string("EUR 20")); + }); + + if(ex.wait(30000ms) < 0) printf(" -> nobody inserted anything\n"); + else printf(" -> got %s\n", note.c_str()); + + player.join(); +} + +// --------------------------------------------------------------------------- + +int main() +{ + notifly center; + const fake_device device(center); + + enable_the_device(center); + print_a_ticket(center, ""); + print_a_ticket(center, ""); + transfer_a_template(center); + read_the_note_table(center); + wait_for_a_note(center); + + printf("\n"); + return 0; +} diff --git a/include/notifly.h b/include/notifly.h index 9ff2665..f913c85 100644 --- a/include/notifly.h +++ b/include/notifly.h @@ -30,6 +30,8 @@ #include #include #include +#include +#include #include #include #include @@ -49,7 +51,7 @@ #endif #define NOTIFLY_VERSION_MAJOR 3 -#define NOTIFLY_VERSION_MINOR 5 +#define NOTIFLY_VERSION_MINOR 6 #define NOTIFLY_VERSION_PATCH 0 #define NOTIFLY_VERSION (NOTIFLY_VERSION_MAJOR << 16 | NOTIFLY_VERSION_MINOR << 8 | NOTIFLY_VERSION_PATCH) @@ -67,6 +69,32 @@ enum class notifly_result timeout = -5 }; +/** + * @brief What a notifly::exchange handler decides about one delivery. + * + * A handler sees every delivery of the notification it subscribed to and says + * what it is: something to ignore, one of several pieces of a streamed reply, + * or the one that ends the wait. + */ +enum class notifly_verdict +{ + skip, ///< Not what is being waited for. Stay subscribed, record nothing. + keep, ///< Part of a streamed reply. Record it and keep waiting. + done ///< This delivery completes the exchange. +}; + +namespace notifly_detail +{ + /** + * @brief Helper trait to detect tuple types. + */ + template + struct is_tuple : std::false_type {}; + + template + struct is_tuple> : std::true_type {}; +} + /** * @brief This class is an observer that is used to observe notifications. */ @@ -306,41 +334,248 @@ class notifly ResultTuple& a_result, PostArgs... post_args) { - // Create promise/future for synchronization - auto response_promise = std::make_shared>(); - std::future response_future = response_promise->get_future(); + // Subscribing before posting is what makes this safe: a reply that comes + // back before the post call has even returned is still caught. + exchange ex(*this); + ex.capture(a_wait_notification, a_result); + if (ex.status() != notifly_result::success) return ex.status(); + + const int post_result = post_notification(a_post_notification, std::forward(post_args)...); + + if (post_result < 0) return static_cast(post_result); + if (post_result == 0) return notifly_result::notification_not_found; + + return ex.wait(std::chrono::milliseconds(a_timeout_ms)) < 0 + ? notifly_result::timeout + : notifly_result::success; + } + + /** + * @brief A scoped set of subscriptions that a thread can block on. + * + * Where post_and_wait() covers the common shape -- post one notification, + * wait for one reply, take the first that arrives -- an exchange covers the + * rest: waiting on several notifications at once and learning which one + * answered, ignoring deliveries that are not the one being waited for, + * collecting a reply the sender streams in pieces, posting nothing at all, + * or treating silence as the successful outcome. + * + * Subscribe with on(), then post whatever the reply is expected to answer, + * then block on wait(), drain() or silent_for(). The destructor + * unsubscribes, and remove_observer() waits for a dispatch already in + * flight, so a handler may safely refer to the caller's own locals. + * + * Once a handler returns notifly_verdict::done the exchange is complete and + * later deliveries are ignored, so a sender that repeats itself -- or + * answers on two of the subscribed notifications -- cannot disturb what the + * winning handler stored. + * + * @warning Never destroy an exchange, or call remove_observer(), from + * inside a handler: dispatch runs with the notification centre + * locked, and unsubscribing there would wait on the very dispatch + * that is running. + */ + class exchange + { + public: + explicit exchange(notifly& a_center) : m_center(a_center) {} - // Register temporary observer for the response - int observer_id = add_response_observer(a_wait_notification, response_promise); + ~exchange() + { + for (const int observer: m_observers) m_center.remove_observer(observer); + } - if (observer_id < 0) return static_cast(observer_id); + exchange(const exchange&) = delete; + exchange& operator=(const exchange&) = delete; + + /** + * @brief Subscribe to a notification, letting a handler judge each delivery. + * @param a_notification The notification to observe. + * @param a_handler Callable (Args...) -> notifly_verdict. + * @return This exchange, so subscriptions can be chained. + */ + template + exchange& on(const int a_notification, Handler a_handler) + { + const int id = m_center.add_observer(a_notification, + std::function( + [this, a_notification, a_handler](Args... args) + { + offer(a_notification, [&] { return a_handler(args...); }); + })); - // Post the request - int post_result = post_notification(a_post_notification, std::forward(post_args)...); + if (id > 0) m_observers.push_back(id); + else if (m_status == notifly_result::success) m_status = static_cast(id); - if (post_result < 0) + return *this; + } + + /** + * @brief Subscribe to a notification, accepting its first delivery and reading nothing out of it. + * @param a_notification The notification to observe. + * @return This exchange, so subscriptions can be chained. + */ + template + exchange& on(const int a_notification) { - remove_observer(observer_id); - return static_cast(post_result); + return on(a_notification, [](Args...) { return notifly_verdict::done; }); } - if (post_result == 0) + /** + * @brief Subscribe to a notification and store its first delivery into a tuple. + * @param a_notification The notification to observe. + * @param a_out Where to store the payload. + * @return This exchange, so subscriptions can be chained. + */ + template + exchange& capture(const int a_notification, std::tuple& a_out) { - remove_observer(observer_id); - return notifly_result::notification_not_found; + return on(a_notification, [&a_out](Args... args) + { + a_out = std::make_tuple(args...); + return notifly_verdict::done; + }); } - // Wait for response with timeout - auto status = response_future.wait_for(std::chrono::milliseconds(a_timeout_ms)); + /** + * @brief Subscribe to a notification and store its first delivery into a single value. + * @param a_notification The notification to observe. + * @param a_out Where to store the payload. + * @return This exchange, so subscriptions can be chained. + */ + template::value, int> = 0> + exchange& capture(const int a_notification, T& a_out) + { + return on(a_notification, [&a_out](T a_value) + { + a_out = a_value; + return notifly_verdict::done; + }); + } - // Remove the temporary observer - remove_observer(observer_id); + /** + * @brief Block until a handler returns done, or the timeout expires. + * @param a_timeout How long to wait. + * @return The notification that completed the exchange, or -1 on timeout. + */ + [[nodiscard]] int wait(const std::chrono::milliseconds a_timeout) + { + std::unique_lock lock(m_mutex); + if(!m_cv.wait_for(lock, a_timeout, [this] { return m_fired >= 0; })) return -1; + return m_fired; + } + + /** + * @brief Block for the whole window and report whether nothing arrived. + * + * For protocols where the sender only answers when a command fails, so + * silence is the successful outcome. + * + * @param a_window How long the sender is given to object. + * @return True if no handler returned done within the window. + */ + [[nodiscard]] bool silent_for(const std::chrono::milliseconds a_window) + { + return wait(a_window) < 0; + } - if (status == std::future_status::timeout) return notifly_result::timeout; + /** + * @brief Block while a reply is streamed in pieces, until the sender goes quiet. + * + * Ends as soon as a handler returns done, or once nothing has been + * accepted for a_quiet, or at a_deadline -- whichever comes first. For + * replies whose length the protocol never states. + * + * @param a_quiet Idle time that marks the end of the stream. + * @param a_deadline Upper bound on the whole wait. + * @return How many deliveries were accepted (keep or done). + */ + [[nodiscard]] std::size_t drain(const std::chrono::milliseconds a_quiet, + const std::chrono::milliseconds a_deadline) + { + const auto deadline = std::chrono::steady_clock::now() + a_deadline; + std::unique_lock lock(m_mutex); - a_result = response_future.get(); - return notifly_result::success; - } + while(m_fired < 0) + { + const auto now = std::chrono::steady_clock::now(); + if(now >= deadline) break; + if(m_accepted > 0 && now - m_last_accept >= a_quiet) break; + + // Wake at whichever comes first: the quiet window closing on the + // last delivery, or the deadline. + auto wake = deadline; + if(m_accepted > 0) + { + if(const auto quiet_at = m_last_accept + a_quiet; quiet_at < wake) wake = quiet_at; + } + m_cv.wait_until(lock, wake); + } + + return m_accepted; + } + + /** + * @brief The first subscription error, or success if every on() call took. + */ + [[nodiscard]] notifly_result status() const { return m_status; } + + /** + * @brief The notification that completed the exchange, or -1 if none has. + */ + [[nodiscard]] int fired() + { + std::lock_guard lock(m_mutex); + return m_fired; + } + + /** + * @brief How many deliveries have been accepted so far. + */ + [[nodiscard]] std::size_t accepted() + { + std::lock_guard lock(m_mutex); + return m_accepted; + } + + private: + /** + * @brief Offer one delivery to its handler and record the verdict. + * + * The handler runs under the exchange's own lock so that "the first done + * wins" is atomic against a second delivery arriving on another thread. + */ + void offer(const int a_notification, const std::function& a_evaluate) + { + { + std::lock_guard lock(m_mutex); + + // Already complete: leave whatever the winning handler stored alone. + if(m_fired >= 0) return; + + const notifly_verdict verdict = a_evaluate(); + if(verdict == notifly_verdict::skip) return; + + ++m_accepted; + m_last_accept = std::chrono::steady_clock::now(); + + if(verdict == notifly_verdict::done) m_fired = a_notification; + } + // Wakes wait() on done, and drain() on either verdict so it can + // re-measure the quiet window. + m_cv.notify_all(); + } + + notifly& m_center; + std::vector m_observers; + notifly_result m_status = notifly_result::success; + + std::mutex m_mutex; + std::condition_variable m_cv; + int m_fired = -1; + std::size_t m_accepted = 0; + std::chrono::steady_clock::time_point m_last_accept{}; + }; /** * @brief Get the default global notification center. @@ -364,42 +599,6 @@ class notifly } private: - /** - * @brief Helper trait to detect tuple types - */ - template - struct is_tuple : std::false_type {}; - - template - struct is_tuple> : std::true_type {}; - - /** - * @brief Add observer for response - specialization for tuple types - */ - template - int add_response_observer(const int a_notification, std::shared_ptr>> promise) - { - // Tuple response - return add_observer(a_notification, [promise](Args... args) - { - promise->set_value(std::make_tuple(args...)); - }); - } - - /** - * @brief Add observer for response - for single types using SFINAE - */ - template - std::enable_if_t::value, int> - add_response_observer(const int a_notification, std::shared_ptr> promise) - { - // Single value response - return add_observer(a_notification, [promise](T value) - { - promise->set_value(value); - }); - } - // Structure to group observer data for a notification struct NotificationData { diff --git a/test/exchange_test.cpp b/test/exchange_test.cpp new file mode 100644 index 0000000..8186080 --- /dev/null +++ b/test/exchange_test.cpp @@ -0,0 +1,444 @@ +// +// Tests for notifly::exchange. +// +// post_and_wait() covers the common shape: post one notification, wait for one +// reply, take the first that arrives. Every test here is a shape that shape +// cannot express -- the ones a real request/reply protocol keeps producing. +// +// Each test builds its own notifly so observer ids and per-notification type +// signatures can never collide with another test's. +// +#include + +#include +#include +#include +#include +#include +#include + +#include "notifly.h" + +namespace +{ + using namespace std::chrono_literals; + + enum : int + { + kCommand = 2000, + kStatus, + kJobComplete, + kJobCancelled, + kTransferError, + kTableEntry, + kDocumentInserted + }; +} + +// --------------------------------------------------------------------------- +// The plain case, the one post_and_wait() already covered. +// --------------------------------------------------------------------------- + +TEST(exchange, the_first_delivery_completes_the_wait) +{ + notifly center; + notifly::exchange ex(center); + ex.on(kStatus); + + std::thread device([&] + { + std::this_thread::sleep_for(20ms); + center.post_notification(kStatus, 1); + }); + + EXPECT_EQ(ex.wait(2000ms), kStatus); + device.join(); +} + +// --------------------------------------------------------------------------- +// A handler that judges each delivery instead of taking the first one. +// --------------------------------------------------------------------------- + +TEST(exchange, a_predicate_ignores_the_state_the_sender_is_leaving) +{ + notifly center; + int observed = -1; + + notifly::exchange ex(center); + // Only a status reporting the state that was asked for ends the wait: a + // device reports the state it is leaving first. + ex.on(kStatus, [&](const int a_state) + { + if(a_state != 1) return notifly_verdict::skip; + observed = a_state; + return notifly_verdict::done; + }); + + std::thread device([&] + { + std::this_thread::sleep_for(10ms); + center.post_notification(kStatus, 0); // leaving + std::this_thread::sleep_for(10ms); + center.post_notification(kStatus, 1); // entering + }); + + EXPECT_EQ(ex.wait(2000ms), kStatus); + EXPECT_EQ(observed, 1); + EXPECT_EQ(ex.accepted(), 1u); // the skipped delivery was not recorded + device.join(); +} + +// --------------------------------------------------------------------------- +// A sender that repeats itself cannot disturb what the winning handler stored. +// --------------------------------------------------------------------------- + +TEST(exchange, a_repeated_reply_cannot_disturb_the_winning_delivery) +{ + notifly center; + int captured = 0; + + notifly::exchange ex(center); + ex.on(kStatus, [&](const int a_value) + { + captured = a_value; + return notifly_verdict::done; + }); + + center.post_notification(kStatus, 7); + + // The sender answers again before the exchange is torn down. + EXPECT_NO_THROW(center.post_notification(kStatus, 99)); + EXPECT_NO_THROW(center.post_notification(kStatus, 123)); + + EXPECT_EQ(ex.wait(0ms), kStatus); + EXPECT_EQ(captured, 7); + EXPECT_EQ(ex.accepted(), 1u); +} + +// --------------------------------------------------------------------------- +// Several notifications at once, and the caller learns which one answered. +// --------------------------------------------------------------------------- + +TEST(exchange, the_caller_learns_which_notification_answered) +{ + notifly center; + std::string outcome; + + notifly::exchange ex(center); + ex.on(kJobComplete, [&](int) + { + outcome = "complete"; + return notifly_verdict::done; + }) + .on(kJobCancelled, [&](const std::string& a_message) + { + outcome = a_message; + return notifly_verdict::done; + }) + .on(kTransferError, [&](int) + { + outcome = "transfer error"; + return notifly_verdict::done; + }); + + center.post_notification(kJobCancelled, std::string("out of paper")); + + EXPECT_EQ(ex.wait(2000ms), kJobCancelled); + EXPECT_EQ(outcome, "out of paper"); +} + +TEST(exchange, the_alternatives_that_did_not_answer_stay_out_of_the_way) +{ + notifly center; + + notifly::exchange ex(center); + ex.on(kJobComplete) + .on(kTransferError); + + center.post_notification(kJobComplete, 1); + // The other alternative answers too, late. The first one still won. + center.post_notification(kTransferError, 1); + + EXPECT_EQ(ex.wait(0ms), kJobComplete); + EXPECT_EQ(ex.accepted(), 1u); +} + +// --------------------------------------------------------------------------- +// Protocols where the sender only speaks up to refuse: silence is success. +// --------------------------------------------------------------------------- + +TEST(exchange, silence_is_the_successful_outcome_when_only_failures_are_reported) +{ + notifly center; + notifly::exchange ex(center); + ex.on(kTransferError); + + EXPECT_TRUE(ex.silent_for(80ms)); + EXPECT_EQ(ex.fired(), -1); +} + +TEST(exchange, a_failure_arriving_within_the_window_breaks_the_silence) +{ + notifly center; + notifly::exchange ex(center); + ex.on(kTransferError); + + std::thread device([&] + { + std::this_thread::sleep_for(10ms); + center.post_notification(kTransferError, 42); + }); + + EXPECT_FALSE(ex.silent_for(2000ms)); + device.join(); +} + +// --------------------------------------------------------------------------- +// A reply the sender streams in pieces, whose length the protocol never states. +// --------------------------------------------------------------------------- + +TEST(exchange, drain_collects_a_streamed_reply_until_the_sender_goes_quiet) +{ + notifly center; + std::mutex entries_mutex; + std::vector entries; + + notifly::exchange ex(center); + ex.on(kTableEntry, [&](const int a_entry) + { + std::lock_guard lock(entries_mutex); + entries.push_back(a_entry); + // Never completes: only the sender going quiet ends this. + return notifly_verdict::keep; + }); + + std::thread device([&] + { + for(int i = 1; i <= 5; ++i) + { + std::this_thread::sleep_for(10ms); + center.post_notification(kTableEntry, i); + } + }); + + const auto count = ex.drain(150ms, 5000ms); + device.join(); + + EXPECT_EQ(count, 5u); + + std::lock_guard lock(entries_mutex); + EXPECT_EQ(entries, (std::vector{1, 2, 3, 4, 5})); +} + +TEST(exchange, drain_gives_up_at_the_deadline_when_the_sender_never_goes_quiet) +{ + notifly center; + std::atomic_bool stop{false}; + + notifly::exchange ex(center); + ex.on(kTableEntry, [](int) { return notifly_verdict::keep; }); + + std::thread device([&] + { + while(!stop.load()) + { + center.post_notification(kTableEntry, 1); + std::this_thread::sleep_for(5ms); + } + }); + + const auto started = std::chrono::steady_clock::now(); + // The quiet window is wide enough that it never closes; the deadline is + // what has to stop this. + const auto count = ex.drain(5000ms, 150ms); + const auto elapsed = std::chrono::steady_clock::now() - started; + + stop.store(true); + device.join(); + + EXPECT_GT(count, 0u); + EXPECT_LT(elapsed, 3000ms); +} + +TEST(exchange, drain_ends_early_when_a_handler_says_done) +{ + notifly center; + + notifly::exchange ex(center); + ex.on(kTableEntry, [](const int a_entry) + { + // A sender that does mark its last entry. + return a_entry < 0 ? notifly_verdict::done : notifly_verdict::keep; + }); + + std::thread device([&] + { + center.post_notification(kTableEntry, 1); + center.post_notification(kTableEntry, 2); + center.post_notification(kTableEntry, -1); // end of table + }); + + const auto started = std::chrono::steady_clock::now(); + const auto count = ex.drain(5000ms, 10000ms); + const auto elapsed = std::chrono::steady_clock::now() - started; + device.join(); + + EXPECT_EQ(count, 3u); + EXPECT_EQ(ex.fired(), kTableEntry); + EXPECT_LT(elapsed, 3000ms); // neither the quiet window nor the deadline was needed +} + +// --------------------------------------------------------------------------- +// Waiting for something no command asked for. +// --------------------------------------------------------------------------- + +TEST(exchange, an_exchange_can_wait_for_something_no_command_asked_for) +{ + notifly center; + std::string document; + + notifly::exchange ex(center); + ex.on(kDocumentInserted, [&](const std::string& a_document) + { + document = a_document; + return notifly_verdict::done; + }); + + // Nothing is posted by the waiter: the document turns up when whoever is at + // the machine decides to insert one. + std::thread player([&] + { + std::this_thread::sleep_for(30ms); + center.post_notification(kDocumentInserted, std::string("ticket-42")); + }); + + EXPECT_EQ(ex.wait(5000ms), kDocumentInserted); + EXPECT_EQ(document, "ticket-42"); + player.join(); +} + +// --------------------------------------------------------------------------- +// Lifetime and error reporting. +// --------------------------------------------------------------------------- + +TEST(exchange, destruction_unsubscribes_every_handler) +{ + notifly center; + int deliveries = 0; + + { + notifly::exchange ex(center); + ex.on(kStatus, [&](int) + { + ++deliveries; + return notifly_verdict::keep; + }); + + center.post_notification(kStatus, 1); + } + + // With the exchange gone the notification has no observers left at all. + EXPECT_EQ(center.post_notification(kStatus, 2), + static_cast(notifly_result::notification_not_found)); + EXPECT_EQ(deliveries, 1); +} + +TEST(exchange, a_type_mismatch_is_reported_instead_of_waiting_for_a_reply_that_cannot_arrive) +{ + notifly center; + + const auto existing = center.add_observer(kStatus, [](const std::string&) {}); + ASSERT_GT(existing, 0); + + notifly::exchange ex(center); + ex.on(kStatus); // wrong payload shape for this notification + + EXPECT_EQ(ex.status(), notifly_result::payload_type_not_match); + + center.remove_observer(existing); +} + +TEST(exchange, a_healthy_set_of_subscriptions_reports_success) +{ + notifly center; + + notifly::exchange ex(center); + ex.on(kJobComplete) + .on(kJobCancelled) + .on(kTransferError); + + EXPECT_EQ(ex.status(), notifly_result::success); +} + +// --------------------------------------------------------------------------- +// capture() is what post_and_wait() is built on: on() plus writing the first +// delivery straight into a variable instead of a handler. It is public on its +// own, for a caller building a multi-alternative exchange who wants that +// shorthand for one of the branches without writing out the lambda. +// --------------------------------------------------------------------------- + +TEST(exchange, capture_writes_the_first_delivery_into_a_single_value) +{ + notifly center; + int status = -1; + + notifly::exchange ex(center); + ex.capture(kStatus, status); + + center.post_notification(kStatus, 7); + + EXPECT_EQ(ex.wait(0ms), kStatus); + EXPECT_EQ(status, 7); +} + +TEST(exchange, capture_writes_a_multi_argument_delivery_into_a_tuple) +{ + notifly center; + std::tuple job; + + notifly::exchange ex(center); + ex.capture(kJobComplete, job); + + center.post_notification(kJobComplete, 42, std::string("ticket-42")); + + EXPECT_EQ(ex.wait(0ms), kJobComplete); + EXPECT_EQ(std::get<0>(job), 42); + EXPECT_EQ(std::get<1>(job), "ticket-42"); +} + +// --------------------------------------------------------------------------- +// post_and_wait() rides on the same machinery, so it inherits the guard. +// --------------------------------------------------------------------------- + +TEST(exchange, post_and_wait_keeps_the_first_answer_when_the_sender_answers_twice) +{ + notifly center; + + const auto responder = center.add_observer(kCommand, [&](const int a_value) + { + // Answers twice, the way a device reporting the state it leaves and + // then the state it enters does. + center.post_notification(kStatus, a_value * 2); + center.post_notification(kStatus, a_value * 3); + }); + ASSERT_GT(responder, 0); + + int result = 0; + notifly_result ret{}; + EXPECT_NO_THROW(ret = center.post_and_wait(kCommand, kStatus, 2000, result, 21)); + + EXPECT_EQ(ret, notifly_result::success); + EXPECT_EQ(result, 42); // the first answer, not the second + + center.remove_observer(responder); +} + +TEST(exchange, post_and_wait_still_reports_a_missing_responder) +{ + notifly center; + + int result = 0; + const auto ret = center.post_and_wait(kCommand, kStatus, 100, result, 1); + + EXPECT_EQ(ret, notifly_result::notification_not_found); +}