Add notifly::exchange, fix post_and_wait's double-answer bug

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$<$<CONFIG:Debug>: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$<$<CONFIG:Debug>:Debug>
+        )
+
 # C Interface example
 set(C_EXAMPLE notifly_c_example)
 add_executable(${C_EXAMPLE} example/c_example.c)

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<int>(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<int>(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

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 <chrono>
+#include <cstdio>
+#include <string>
+#include <thread>
+#include <vector>
+
+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<int> 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<int>(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<int>(evt_job_complete, [&](const int a_transaction)
+      {
+          transaction = a_transaction;
+          return notifly_verdict::done;
+      })
+      .on<std::string>(evt_job_refused, [&](const std::string& a_reason)
+      {
+          refusal = a_reason;
+          return notifly_verdict::done;
+      })
+      .on<int>(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<std::string>(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<int> entries;
+
+    notifly::exchange ex(a_center);
+    ex.on<int>(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<std::string>(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, "<ticket/>");
+    print_a_ticket(center, "");
+    transfer_a_template(center);
+    read_the_note_table(center);
+    wait_for_a_note(center);
+
+    printf("\n");
+    return 0;
+}