mirror of
https://github.com/logos-messaging/logos-messaging-nim.git
synced 2026-01-31 04:03:12 +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
73 lines
2.4 KiB
Nim
73 lines
2.4 KiB
Nim
import chronicles, chronos, results
|
|
|
|
import waku/factory/waku
|
|
import waku/[requests/health_request, waku_core, waku_node]
|
|
import waku/node/delivery_service/send_service
|
|
import waku/node/delivery_service/subscription_service
|
|
import ./[api_conf, types]
|
|
|
|
logScope:
|
|
topics = "api"
|
|
|
|
# TODO: Specs says it should return a `WakuNode`. As `send` and other APIs are defined, we can align.
|
|
proc createNode*(config: NodeConfig): Future[Result[Waku, string]] {.async.} =
|
|
let wakuConf = toWakuConf(config).valueOr:
|
|
return err("Failed to handle the configuration: " & error)
|
|
|
|
## We are not defining app callbacks at node creation
|
|
let wakuRes = (await Waku.new(wakuConf)).valueOr:
|
|
error "waku initialization failed", error = error
|
|
return err("Failed setting up Waku: " & $error)
|
|
|
|
return ok(wakuRes)
|
|
|
|
proc checkApiAvailability(w: Waku): Result[void, string] =
|
|
if w.isNil():
|
|
return err("Waku node is not initialized")
|
|
|
|
# check if health is satisfactory
|
|
# If Node is not healthy, return err("Waku node is not healthy")
|
|
let healthStatus = RequestNodeHealth.request(w.brokerCtx)
|
|
|
|
if healthStatus.isErr():
|
|
warn "Failed to get Waku node health status: ", error = healthStatus.error
|
|
# Let's suppose the node is hesalthy enough, go ahead
|
|
else:
|
|
if healthStatus.get().healthStatus == NodeHealth.Unhealthy:
|
|
return err("Waku node is not healthy, has got no connections.")
|
|
|
|
return ok()
|
|
|
|
proc subscribe*(
|
|
w: Waku, contentTopic: ContentTopic
|
|
): Future[Result[void, string]] {.async.} =
|
|
?checkApiAvailability(w)
|
|
|
|
return w.deliveryService.subscriptionService.subscribe(contentTopic)
|
|
|
|
proc unsubscribe*(w: Waku, contentTopic: ContentTopic): Result[void, string] =
|
|
?checkApiAvailability(w)
|
|
|
|
return w.deliveryService.subscriptionService.unsubscribe(contentTopic)
|
|
|
|
proc send*(
|
|
w: Waku, envelope: MessageEnvelope
|
|
): Future[Result[RequestId, string]] {.async.} =
|
|
?checkApiAvailability(w)
|
|
|
|
let requestId = RequestId.new(w.rng)
|
|
|
|
let deliveryTask = DeliveryTask.new(requestId, envelope, w.brokerCtx).valueOr:
|
|
return err("API send: Failed to create delivery task: " & error)
|
|
|
|
info "API send: scheduling delivery task",
|
|
requestId = $requestId,
|
|
pubsubTopic = deliveryTask.pubsubTopic,
|
|
contentTopic = deliveryTask.msg.contentTopic,
|
|
msgHash = deliveryTask.msgHash.to0xHex(),
|
|
myPeerId = w.node.peerId()
|
|
|
|
asyncSpawn w.deliveryService.sendService.send(deliveryTask)
|
|
|
|
return ok(requestId)
|