mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-07-19 11:09:26 +00:00
refactor(apps): adopt WakuEnvelope in relay handlers and message-path bench
chat2 and networkmonitor relay handlers take the envelope; the micro/macro benchmark's app handlers follow the new signature (paced/windowed publishing and counters unchanged). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
84a42a3eb4
commit
e1ca47ce33
@ -211,7 +211,7 @@ proc emit(res: ScenarioResult) =
|
||||
# Node setup
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
proc dummyHandler(topic: PubsubTopic, msg: WakuMessage) {.async, gcsafe.} =
|
||||
proc dummyHandler(envelope: WakuEnvelope) {.async, gcsafe.} =
|
||||
discard
|
||||
|
||||
proc buildIngestNode(
|
||||
@ -304,7 +304,7 @@ proc runMacro(shard: PubsubTopic, work: Workload): Future[ScenarioResult] {.asyn
|
||||
times: newSeqOfCap[MonoTime](work.msgs.len),
|
||||
)
|
||||
|
||||
proc countingHandler(topic: PubsubTopic, msg: WakuMessage) {.async, gcsafe.} =
|
||||
proc countingHandler(envelope: WakuEnvelope) {.async, gcsafe.} =
|
||||
arrivals.times.add(getMonoTime())
|
||||
arrivals.count += 1
|
||||
if arrivals.count >= arrivals.target:
|
||||
|
||||
@ -508,7 +508,9 @@ proc processInput(rfd: AsyncFD, rng: crypto.Rng) {.async.} =
|
||||
|
||||
# Subscribe to a topic, if relay is mounted
|
||||
if conf.relay:
|
||||
proc handler(topic: PubsubTopic, msg: WakuMessage): Future[void] {.async, gcsafe.} =
|
||||
proc handler(envelope: WakuEnvelope): Future[void] {.async, gcsafe.} =
|
||||
let topic = envelope.pubsubTopic
|
||||
let msg = envelope.msg
|
||||
trace "Hit subscribe handler", topic
|
||||
|
||||
if msg.contentTopic == chat.contentTopic:
|
||||
|
||||
@ -518,9 +518,9 @@ proc subscribeAndHandleMessages(
|
||||
msgPerContentTopic: ContentTopicMessageTableRef,
|
||||
) =
|
||||
# handle function
|
||||
proc handler(
|
||||
pubsubTopic: PubsubTopic, msg: WakuMessage
|
||||
): Future[void] {.async, gcsafe.} =
|
||||
proc handler(envelope: WakuEnvelope): Future[void] {.async, gcsafe.} =
|
||||
let pubsubTopic = envelope.pubsubTopic
|
||||
let msg = envelope.msg
|
||||
trace "rx message", pubsubTopic = pubsubTopic, contentTopic = msg.contentTopic
|
||||
|
||||
# If we reach a table limit size, remove c topics with the least messages.
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user