mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-08-04 10:53:19 +00:00
18e17d2ddd
Move messaging_client.nim and the delivery_service machinery (send/recv services, processors, not_delivered_storage) out of waku/ and waku/node/ into a dedicated top-level messaging/ folder. The delivery services are a messaging-layer concern built on top of the node, not part of node core, so a sibling folder makes that boundary explicit. Shared message types (waku/api/types) and the node-owned subscription_manager are left in place to avoid churn on widely-imported modules. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
37 lines
1.1 KiB
Nim
37 lines
1.1 KiB
Nim
import chronos
|
|
import brokers/broker_context
|
|
import ./delivery_task
|
|
|
|
{.push raises: [].}
|
|
|
|
type BaseSendProcessor* = ref object of RootObj
|
|
fallbackProcessor*: BaseSendProcessor
|
|
brokerCtx*: BrokerContext
|
|
|
|
proc chain*(self: BaseSendProcessor, next: BaseSendProcessor) =
|
|
self.fallbackProcessor = next
|
|
|
|
method isValidProcessor*(
|
|
self: BaseSendProcessor, task: DeliveryTask
|
|
): bool {.base, gcsafe.} =
|
|
return false
|
|
|
|
method sendImpl*(
|
|
self: BaseSendProcessor, task: DeliveryTask
|
|
): Future[void] {.async, base.} =
|
|
assert false, "Not implemented"
|
|
|
|
method process*(
|
|
self: BaseSendProcessor, task: DeliveryTask
|
|
): Future[void] {.async, base.} =
|
|
var currentProcessor: BaseSendProcessor = self
|
|
var keepTrying = true
|
|
while not currentProcessor.isNil() and keepTrying:
|
|
if currentProcessor.isValidProcessor(task):
|
|
await currentProcessor.sendImpl(task)
|
|
currentProcessor = currentProcessor.fallbackProcessor
|
|
keepTrying = task.state == DeliveryState.FallbackRetry
|
|
|
|
if task.state == DeliveryState.FallbackRetry:
|
|
task.state = DeliveryState.NextRoundRetry
|