mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-07-23 05:00:21 +00:00
ed060553b0
New messaging option send-confirmation = store (default) | propagation. In propagation mode MessageSent fires from the publish path (relay mesh / lightpush relayPeerCount) right after MessagePropagated, with no store polling; finalized tasks no longer re-enter the task cache, so the confirmation is emitted exactly once. Store mode is bit-identical to before. The option flows through both the structured and legacy flat JSON config shapes. Part of the "Store as a startup-only dependency" experiment (step 9). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
55 lines
1.8 KiB
Nim
55 lines
1.8 KiB
Nim
## Messaging layer core: the `MessagingClient` type plus its construction and
|
|
## lifecycle. The public operations (subscribe / unsubscribe / send) live in
|
|
## `messaging/api.nim`.
|
|
import std/strutils
|
|
import results, chronos, chronicles
|
|
import
|
|
logos_delivery/api/conf/messaging_conf,
|
|
logos_delivery/api/messaging_client_api,
|
|
logos_delivery/waku/waku,
|
|
logos_delivery/waku/factory/conf_builder/waku_conf_builder,
|
|
logos_delivery/messaging/delivery_service/[recv_service, send_service]
|
|
|
|
export messaging_client_api, messaging_conf
|
|
|
|
type MessagingClient* = ref object
|
|
brokerCtx*: BrokerContext
|
|
waku*: Waku ## The Waku kernel this layer drives; read by `messaging/api/*`.
|
|
sendService*: SendService
|
|
recvService*: RecvService
|
|
started*: bool
|
|
|
|
proc new*(
|
|
T: type MessagingClient, conf: MessagingClientConf, waku: Waku
|
|
): Result[T, string] =
|
|
## The messaging layer chains onto Waku: it drives the underlying Waku kernel
|
|
## for transport while exposing its own send/recv API.
|
|
let reliability = conf.reliabilityEnabled.get(DefaultP2pReliability)
|
|
|
|
let confirmationMode =
|
|
case conf.sendConfirmation.get("store").strip().toLowerAscii()
|
|
of "store":
|
|
SendConfirmationMode.Store
|
|
of "propagation":
|
|
SendConfirmationMode.Propagation
|
|
else:
|
|
return err("invalid send-confirmation mode: " & conf.sendConfirmation.get(""))
|
|
|
|
let sendService = ?SendService.new(reliability, waku, confirmationMode)
|
|
let recvService = RecvService.new(waku)
|
|
return ok(
|
|
T(
|
|
waku: waku,
|
|
sendService: sendService,
|
|
recvService: recvService,
|
|
brokerCtx: waku.brokerCtx,
|
|
)
|
|
)
|
|
|
|
proc checkApiAvailability*(self: MessagingClient): Result[void, string] =
|
|
## Shared guard for the api operation module.
|
|
if self.isNil():
|
|
return err("MessagingClient is not initialized")
|
|
|
|
return ok()
|