chore: Introduce EventBroker, RequestBroker and MultiRequestBroker (#3644)

* Introduce EventBroker and RequestBroker as decoupling helpers that represent reactive (event-driven) and proactive (request/response) patterns without tight coupling between modules
* Address copilot observation. error log if failed listener call exception, handling listener overuse - run out of IDs
* Address review observations: no exception to leak, listeners must raise no exception, adding listener now reports error with Result.
* Added MultiRequestBroker utility to collect results from many providers
* Support an arbitrary number of arguments for RequestBroker's request/provider signature
* MultiRequestBroker allows provider procs to throw exceptions, which will be handled during request processing.
* MultiRequestBroker supports one zero arg signature and/or multi arg signature
* test no exception leaks from RequestBroker and MultiRequestBroker
* Embed MultiRequestBroker tests into common
* EventBroker: removed all ...Broker typed public procs to simplify EventBroker interface, forger is renamed to dropListener
* Make Request's broker type private
* MultiRequestBroker: Use explicit returns in generated procs
* Updated descriptions of EventBroker and RequestBroker, updated RequestBroker.setProvider, returns error if already set.
* Better description for MultiRequestBroker and its usage
* Add EventBroker support for ref objects, fix emit variant with event object ctor
* Add RequestBroker support for ref objects
* Add MultiRequestBroker support for ref objects
* Mover brokers under waku/common

Author: NagyZoltanPeter

tests/common/test_all.nim

@@ -9,4 +9,7 @@ import
   ./test_tokenbucket,
   ./test_requestratelimiter,
   ./test_ratelimit_setting,
-  ./test_timed_map
+  ./test_timed_map,
+  ./test_event_broker,
+  ./test_request_broker,
+  ./test_multi_request_broker

tests/common/test_event_broker.nim

@@ -0,0 +1,125 @@
+import chronos
+import std/sequtils
+import testutils/unittests
+
+import waku/common/broker/event_broker
+
+EventBroker:
+  type SampleEvent = object
+    value*: int
+    label*: string
+
+EventBroker:
+  type BinaryEvent = object
+    flag*: bool
+
+EventBroker:
+  type RefEvent = ref object
+    payload*: seq[int]
+
+template waitForListeners() =
+  waitFor sleepAsync(1.milliseconds)
+
+suite "EventBroker":
+  test "delivers events to all listeners":
+    var seen: seq[(int, string)] = @[]
+
+    discard SampleEvent.listen(
+      proc(evt: SampleEvent): Future[void] {.async: (raises: []).} =
+        seen.add((evt.value, evt.label))
+    )
+
+    discard SampleEvent.listen(
+      proc(evt: SampleEvent): Future[void] {.async: (raises: []).} =
+        seen.add((evt.value * 2, evt.label & "!"))
+    )
+
+    let evt = SampleEvent(value: 5, label: "hi")
+    SampleEvent.emit(evt)
+    waitForListeners()
+
+    check seen.len == 2
+    check seen.anyIt(it == (5, "hi"))
+    check seen.anyIt(it == (10, "hi!"))
+
+    SampleEvent.dropAllListeners()
+
+  test "forget removes a single listener":
+    var counter = 0
+
+    let handleA = SampleEvent.listen(
+      proc(evt: SampleEvent): Future[void] {.async: (raises: []).} =
+        inc counter
+    )
+
+    let handleB = SampleEvent.listen(
+      proc(evt: SampleEvent): Future[void] {.async: (raises: []).} =
+        inc(counter, 2)
+    )
+
+    SampleEvent.dropListener(handleA.get())
+    let eventVal = SampleEvent(value: 1, label: "one")
+    SampleEvent.emit(eventVal)
+    waitForListeners()
+    check counter == 2
+
+    SampleEvent.dropAllListeners()
+
+  test "forgetAll clears every listener":
+    var triggered = false
+
+    let handle1 = SampleEvent.listen(
+      proc(evt: SampleEvent): Future[void] {.async: (raises: []).} =
+        triggered = true
+    )
+    let handle2 = SampleEvent.listen(
+      proc(evt: SampleEvent): Future[void] {.async: (raises: []).} =
+        discard
+    )
+
+    SampleEvent.dropAllListeners()
+    SampleEvent.emit(42, "noop")
+    SampleEvent.emit(label = "noop", value = 42)
+    waitForListeners()
+    check not triggered
+
+    let freshHandle = SampleEvent.listen(
+      proc(evt: SampleEvent): Future[void] {.async: (raises: []).} =
+        discard
+    )
+    check freshHandle.get().id > 0'u64
+    SampleEvent.dropListener(freshHandle.get())
+
+  test "broker helpers operate via typedesc":
+    var toggles: seq[bool] = @[]
+
+    let handle = BinaryEvent.listen(
+      proc(evt: BinaryEvent): Future[void] {.async: (raises: []).} =
+        toggles.add(evt.flag)
+    )
+
+    BinaryEvent(flag: true).emit()
+    waitForListeners()
+    let binaryEvent = BinaryEvent(flag: false)
+    BinaryEvent.emit(binaryEvent)
+    waitForListeners()
+
+    check toggles == @[true, false]
+    BinaryEvent.dropAllListeners()
+
+  test "ref typed event":
+    var counter: int = 0
+
+    let handle = RefEvent.listen(
+      proc(evt: RefEvent): Future[void] {.async: (raises: []).} =
+        for n in evt.payload:
+          counter += n
+    )
+
+    RefEvent(payload: @[1, 2, 3]).emit()
+    waitForListeners()
+    RefEvent.emit(payload = @[4, 5, 6])
+    waitForListeners()
+
+    check counter == 21 # 1+2+3 + 4+5+6
+    RefEvent.dropAllListeners()

tests/common/test_multi_request_broker.nim

@@ -0,0 +1,234 @@
+{.used.}
+
+import testutils/unittests
+import chronos
+import std/sequtils
+import std/strutils
+
+import waku/common/broker/multi_request_broker
+
+MultiRequestBroker:
+  type NoArgResponse = object
+    label*: string
+
+  proc signatureFetch*(): Future[Result[NoArgResponse, string]] {.async.}
+
+MultiRequestBroker:
+  type ArgResponse = object
+    id*: string
+
+  proc signatureFetch*(
+    suffix: string, numsuffix: int
+  ): Future[Result[ArgResponse, string]] {.async.}
+
+MultiRequestBroker:
+  type DualResponse = ref object
+    note*: string
+    suffix*: string
+
+  proc signatureBase*(): Future[Result[DualResponse, string]] {.async.}
+  proc signatureWithInput*(
+    suffix: string
+  ): Future[Result[DualResponse, string]] {.async.}
+
+suite "MultiRequestBroker":
+  test "aggregates zero-argument providers":
+    discard NoArgResponse.setProvider(
+      proc(): Future[Result[NoArgResponse, string]] {.async.} =
+        ok(NoArgResponse(label: "one"))
+    )
+
+    discard NoArgResponse.setProvider(
+      proc(): Future[Result[NoArgResponse, string]] {.async.} =
+        discard catch:
+          await sleepAsync(1.milliseconds)
+        ok(NoArgResponse(label: "two"))
+    )
+
+    let responses = waitFor NoArgResponse.request()
+    check responses.get().len == 2
+    check responses.get().anyIt(it.label == "one")
+    check responses.get().anyIt(it.label == "two")
+
+    NoArgResponse.clearProviders()
+
+  test "aggregates argument providers":
+    discard ArgResponse.setProvider(
+      proc(suffix: string, num: int): Future[Result[ArgResponse, string]] {.async.} =
+        ok(ArgResponse(id: suffix & "-a-" & $num))
+    )
+
+    discard ArgResponse.setProvider(
+      proc(suffix: string, num: int): Future[Result[ArgResponse, string]] {.async.} =
+        ok(ArgResponse(id: suffix & "-b-" & $num))
+    )
+
+    let keyed = waitFor ArgResponse.request("topic", 1)
+    check keyed.get().len == 2
+    check keyed.get().anyIt(it.id == "topic-a-1")
+    check keyed.get().anyIt(it.id == "topic-b-1")
+
+    ArgResponse.clearProviders()
+
+  test "clearProviders resets both provider lists":
+    discard DualResponse.setProvider(
+      proc(): Future[Result[DualResponse, string]] {.async.} =
+        ok(DualResponse(note: "base", suffix: ""))
+    )
+
+    discard DualResponse.setProvider(
+      proc(suffix: string): Future[Result[DualResponse, string]] {.async.} =
+        ok(DualResponse(note: "base" & suffix, suffix: suffix))
+    )
+
+    let noArgs = waitFor DualResponse.request()
+    check noArgs.get().len == 1
+
+    let param = waitFor DualResponse.request("-extra")
+    check param.get().len == 1
+    check param.get()[0].suffix == "-extra"
+
+    DualResponse.clearProviders()
+
+    let emptyNoArgs = waitFor DualResponse.request()
+    check emptyNoArgs.get().len == 0
+
+    let emptyWithArgs = waitFor DualResponse.request("-extra")
+    check emptyWithArgs.get().len == 0
+
+  test "request returns empty seq when no providers registered":
+    let empty = waitFor NoArgResponse.request()
+    check empty.get().len == 0
+
+  test "failed providers will fail the request":
+    NoArgResponse.clearProviders()
+    discard NoArgResponse.setProvider(
+      proc(): Future[Result[NoArgResponse, string]] {.async.} =
+        err("boom")
+    )
+
+    discard NoArgResponse.setProvider(
+      proc(): Future[Result[NoArgResponse, string]] {.async.} =
+        ok(NoArgResponse(label: "survivor"))
+    )
+
+    let filtered = waitFor NoArgResponse.request()
+    check filtered.isErr()
+
+    NoArgResponse.clearProviders()
+
+  test "deduplicates identical zero-argument providers":
+    NoArgResponse.clearProviders()
+    var invocations = 0
+    let sharedHandler = proc(): Future[Result[NoArgResponse, string]] {.async.} =
+      inc invocations
+      ok(NoArgResponse(label: "dup"))
+
+    let first = NoArgResponse.setProvider(sharedHandler)
+    let second = NoArgResponse.setProvider(sharedHandler)
+
+    check first.get().id == second.get().id
+    check first.get().kind == second.get().kind
+
+    let dupResponses = waitFor NoArgResponse.request()
+    check dupResponses.get().len == 1
+    check invocations == 1
+
+    NoArgResponse.clearProviders()
+
+  test "removeProvider deletes registered handlers":
+    var removedCalled = false
+    var keptCalled = false
+
+    let removable = NoArgResponse.setProvider(
+      proc(): Future[Result[NoArgResponse, string]] {.async.} =
+        removedCalled = true
+        ok(NoArgResponse(label: "removed"))
+    )
+
+    discard NoArgResponse.setProvider(
+      proc(): Future[Result[NoArgResponse, string]] {.async.} =
+        keptCalled = true
+        ok(NoArgResponse(label: "kept"))
+    )
+
+    NoArgResponse.removeProvider(removable.get())
+
+    let afterRemoval = (waitFor NoArgResponse.request()).valueOr:
+      assert false, "request failed"
+      @[]
+    check afterRemoval.len == 1
+    check afterRemoval[0].label == "kept"
+    check not removedCalled
+    check keptCalled
+
+    NoArgResponse.clearProviders()
+
+  test "removeProvider works for argument signatures":
+    var invoked: seq[string] = @[]
+
+    discard ArgResponse.setProvider(
+      proc(suffix: string, num: int): Future[Result[ArgResponse, string]] {.async.} =
+        invoked.add("first" & suffix)
+        ok(ArgResponse(id: suffix & "-one-" & $num))
+    )
+
+    let handle = ArgResponse.setProvider(
+      proc(suffix: string, num: int): Future[Result[ArgResponse, string]] {.async.} =
+        invoked.add("second" & suffix)
+        ok(ArgResponse(id: suffix & "-two-" & $num))
+    )
+
+    ArgResponse.removeProvider(handle.get())
+
+    let single = (waitFor ArgResponse.request("topic", 1)).valueOr:
+      assert false, "request failed"
+      @[]
+    check single.len == 1
+    check single[0].id == "topic-one-1"
+    check invoked == @["firsttopic"]
+
+    ArgResponse.clearProviders()
+
+  test "catches exception from providers and report error":
+    let firstHandler = NoArgResponse.setProvider(
+      proc(): Future[Result[NoArgResponse, string]] {.async.} =
+        raise newException(ValueError, "first handler raised")
+        ok(NoArgResponse(label: "any"))
+    )
+
+    discard NoArgResponse.setProvider(
+      proc(): Future[Result[NoArgResponse, string]] {.async.} =
+        ok(NoArgResponse(label: "just ok"))
+    )
+
+    let afterException = waitFor NoArgResponse.request()
+    check afterException.isErr()
+    check afterException.error().contains("first handler raised")
+
+    NoArgResponse.clearProviders()
+
+  test "ref providers returning nil fail request":
+    DualResponse.clearProviders()
+
+    discard DualResponse.setProvider(
+      proc(): Future[Result[DualResponse, string]] {.async.} =
+        let nilResponse: DualResponse = nil
+        ok(nilResponse)
+    )
+
+    let zeroArg = waitFor DualResponse.request()
+    check zeroArg.isErr()
+
+    DualResponse.clearProviders()
+
+    discard DualResponse.setProvider(
+      proc(suffix: string): Future[Result[DualResponse, string]] {.async.} =
+        let nilResponse: DualResponse = nil
+        ok(nilResponse)
+    )
+
+    let withInput = waitFor DualResponse.request("-extra")
+    check withInput.isErr()
+
+    DualResponse.clearProviders()