mirror of
https://github.com/logos-messaging/logos-messaging-nim.git
synced 2026-02-13 10:23:08 +00:00
1fd25355e0
* Introduce api/send Added events and requests for support. Reworked delivery_monitor into a featured devlivery_service, that - supports relay publish and lightpush depending on configuration but with fallback options - if available and configured it utilizes store api to confirm message delivery - emits message delivery events accordingly prepare for use in api_example * Fix edge mode config and test added * Fix some import issues, start and stop waku shall not throw exception but return with result properly * Utlize sync RequestBroker, adapt to non-async broker usage and gcsafe where appropriate, removed leftover * add api_example app to examples2 * Adapt after merge from master * Adapt code for using broker context * Fix brokerCtx settings for all usedbrokers, cover locked node init * Various fixes upon test failures. Added initial of subscribe API and auto-subscribe for send api * More test added * Fix multi propagate event emit, fix fail send test case * Fix rebase * Fix PushMessageHandlers in tests * adapt libwaku to api changes * Fix relay test by adapting publish return error in case NoPeersToPublish * Addressing all remaining review findings. Removed leftovers. Fixed loggings and typos * Fix rln relay broker, missed brokerCtx * Fix rest relay test failed, due to publish will fail if no peer avail * ignore anvil test state file * Make terst_wakunode_rln_relay broker context aware to fix * Fix waku rln tests by having them broker context aware * fix typo in test_app.nim
37 lines
1.1 KiB
Nim
37 lines
1.1 KiB
Nim
import chronos
|
|
import ./delivery_task
|
|
import waku/common/broker/broker_context
|
|
|
|
{.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
|