mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-03-31 05:13:07 +00:00
51ec09c39d
* SubscriptionManager tracks shard and content topic interest * RecvService emits MessageReceivedEvent on subscribed content topics * Route MAPI through old Kernel API relay unique-handler infra to avoid code duplication * Encode current gen-zero network policy: on Core node boot, subscribe to all pubsub topics (all shards) * Add test_api_subscriptions.nim (basic relay/core testing only) * Removed any MAPI Edge sub/unsub/receive support code that was there (will add in next PR) * Hook MessageSeenEvent to Kernel API bus * Fix MAPI vs Kernel API unique relay handler support * RecvService delegating topic subs to SubscriptionManager * RecvService emits MessageReceivedEvent (fully filtered) * Rename old SubscriptionManager to LegacySubscriptionManager
53 lines
1.5 KiB
Nim
53 lines
1.5 KiB
Nim
{.push raises: [].}
|
|
|
|
import std/tables, results, chronicles, chronos
|
|
|
|
import ./push_handler, ../topics, ../message
|
|
|
|
## Subscription manager
|
|
type LegacySubscriptionManager* = object
|
|
subscriptions: TableRef[(string, ContentTopic), FilterPushHandler]
|
|
|
|
proc init*(T: type LegacySubscriptionManager): T =
|
|
LegacySubscriptionManager(
|
|
subscriptions: newTable[(string, ContentTopic), FilterPushHandler]()
|
|
)
|
|
|
|
proc clear*(m: var LegacySubscriptionManager) =
|
|
m.subscriptions.clear()
|
|
|
|
proc registerSubscription*(
|
|
m: LegacySubscriptionManager,
|
|
pubsubTopic: PubsubTopic,
|
|
contentTopic: ContentTopic,
|
|
handler: FilterPushHandler,
|
|
) =
|
|
try:
|
|
# TODO: Handle over subscription surprises
|
|
m.subscriptions[(pubsubTopic, contentTopic)] = handler
|
|
except CatchableError:
|
|
error "failed to register filter subscription", error = getCurrentExceptionMsg()
|
|
|
|
proc removeSubscription*(
|
|
m: LegacySubscriptionManager, pubsubTopic: PubsubTopic, contentTopic: ContentTopic
|
|
) =
|
|
m.subscriptions.del((pubsubTopic, contentTopic))
|
|
|
|
proc notifySubscriptionHandler*(
|
|
m: LegacySubscriptionManager,
|
|
pubsubTopic: PubsubTopic,
|
|
contentTopic: ContentTopic,
|
|
message: WakuMessage,
|
|
) =
|
|
if not m.subscriptions.hasKey((pubsubTopic, contentTopic)):
|
|
return
|
|
|
|
try:
|
|
let handler = m.subscriptions[(pubsubTopic, contentTopic)]
|
|
asyncSpawn handler(pubsubTopic, message)
|
|
except CatchableError:
|
|
discard
|
|
|
|
proc getSubscriptionsCount*(m: LegacySubscriptionManager): int =
|
|
m.subscriptions.len()
|