wakuv2 connectivity and peer management (#13)
This commit is contained in:
parent
537024fdc5
commit
e474468da3
|
@ -56,3 +56,24 @@ proc signTypedData*(data: string, address: string, password: string): string =
|
||||||
|
|
||||||
proc getBloomFilter*(): string =
|
proc getBloomFilter*(): string =
|
||||||
return $callPrivateRPC("bloomFilter".prefix, %* []).parseJSON()["result"].getStr
|
return $callPrivateRPC("bloomFilter".prefix, %* []).parseJSON()["result"].getStr
|
||||||
|
|
||||||
|
proc adminPeers*(): seq[string] =
|
||||||
|
let response = callPrivateRPC("admin_peers", %* []).parseJSON()["result"]
|
||||||
|
for jsonPeer in response:
|
||||||
|
result.add(jsonPeer["enode"].getStr)
|
||||||
|
|
||||||
|
proc wakuV2Peers*(): seq[string] =
|
||||||
|
let response = callPrivateRPC("peers".prefix, %* []).parseJSON()["result"]
|
||||||
|
for (id, proto) in response.pairs:
|
||||||
|
if proto.len != 0:
|
||||||
|
result.add(id)
|
||||||
|
|
||||||
|
proc dialPeer*(address: string):bool =
|
||||||
|
let response = callPrivateRPC("dialPeer".prefix, %* [address]).parseJSON()
|
||||||
|
if response.hasKey("error"):
|
||||||
|
error "waku peer could not be dialed", response
|
||||||
|
return false
|
||||||
|
return true
|
||||||
|
|
||||||
|
proc dropPeerByID*(peerID: string) =
|
||||||
|
echo callPrivateRPC("dropPeer".prefix, %* [peerID])
|
||||||
|
|
|
@ -1,11 +1,12 @@
|
||||||
import json, times
|
import json, times
|
||||||
import core, ../utils
|
import core, ../utils
|
||||||
|
|
||||||
proc ping*(mailservers: seq[string], timeoutMs: int): string =
|
proc ping*(mailservers: seq[string], timeoutMs: int, isWakuV2: bool = false): string =
|
||||||
var addresses: seq[string] = @[]
|
var addresses: seq[string] = @[]
|
||||||
for mailserver in mailservers:
|
for mailserver in mailservers:
|
||||||
addresses.add(mailserver)
|
addresses.add(mailserver)
|
||||||
result = callPrivateRPC("mailservers_ping", %* [
|
var rpcMethod = if isWakuV2: "mailservers_multiAddressPing" else: "mailservers_ping"
|
||||||
|
result = callPrivateRPC(rpcMethod, %* [
|
||||||
{ "addresses": addresses, "timeoutMs": timeoutMs }
|
{ "addresses": addresses, "timeoutMs": timeoutMs }
|
||||||
])
|
])
|
||||||
|
|
||||||
|
|
|
@ -1,10 +1,11 @@
|
||||||
import chronicles
|
import chronicles
|
||||||
import ../eventemitter
|
import ../eventemitter
|
||||||
import libstatus/settings
|
import libstatus/settings
|
||||||
|
import libstatus/core
|
||||||
import json
|
import json
|
||||||
import uuids
|
import uuids, strutils
|
||||||
import json_serialization
|
import json_serialization
|
||||||
import ./types/[setting]
|
import ./types/[setting, fleet]
|
||||||
|
|
||||||
logScope:
|
logScope:
|
||||||
topics = "network-model"
|
topics = "network-model"
|
||||||
|
@ -21,6 +22,16 @@ proc newNetworkModel*(events: EventEmitter): NetworkModel =
|
||||||
result.peers = @[]
|
result.peers = @[]
|
||||||
result.connected = false
|
result.connected = false
|
||||||
|
|
||||||
|
proc fetchPeers*(self: NetworkModel): seq[string] =
|
||||||
|
var fleetStr = getSetting[string](Setting.Fleet)
|
||||||
|
if fleetStr == "": fleetStr = "eth.prod"
|
||||||
|
let fleet = parseEnum[Fleet](fleetStr)
|
||||||
|
let isWakuV2 = if fleet == WakuV2Prod or fleet == WakuV2Test: true else: false
|
||||||
|
if isWakuV2:
|
||||||
|
return wakuV2Peers()
|
||||||
|
else:
|
||||||
|
return adminPeers()
|
||||||
|
|
||||||
proc peerSummaryChange*(self: NetworkModel, peers: seq[string]) =
|
proc peerSummaryChange*(self: NetworkModel, peers: seq[string]) =
|
||||||
if peers.len == 0 and self.connected:
|
if peers.len == 0 and self.connected:
|
||||||
self.connected = false
|
self.connected = false
|
||||||
|
|
|
@ -58,6 +58,9 @@ proc getCurrentNetwork*(self: SettingsModel): NetworkType =
|
||||||
proc setWakuVersion*(self: SettingsModel, newVersion: int) =
|
proc setWakuVersion*(self: SettingsModel, newVersion: int) =
|
||||||
libstatus_settings.setWakuVersion(newVersion)
|
libstatus_settings.setWakuVersion(newVersion)
|
||||||
|
|
||||||
|
proc getWakuVersion*(self: SettingsModel): int =
|
||||||
|
libstatus_settings.getWakuVersion()
|
||||||
|
|
||||||
proc setBloomFilterMode*(self: SettingsModel, bloomFilterMode: bool): StatusGoError =
|
proc setBloomFilterMode*(self: SettingsModel, bloomFilterMode: bool): StatusGoError =
|
||||||
libstatus_settings.setBloomFilterMode(bloomFilterMode)
|
libstatus_settings.setBloomFilterMode(bloomFilterMode)
|
||||||
|
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
import json, json_serialization, strutils
|
import json, json_serialization, strutils
|
||||||
import signals/[base, community, discovery_summary, envelope, expired, mailserver, messages, signal_type, stats, wallet, whisper_filter]
|
import signals/[base, community, discovery_summary, envelope, expired, mailserver, messages, signal_type, stats, wallet, whisper_filter, peerstats]
|
||||||
|
|
||||||
export base, community, discovery_summary, envelope, expired, mailserver, messages, signal_type, stats, wallet, whisper_filter
|
export base, community, discovery_summary, envelope, expired, mailserver, messages, signal_type, stats, wallet, whisper_filter, peerstats
|
||||||
|
|
||||||
proc decode*(jsonSignal: JsonNode): Signal =
|
proc decode*(jsonSignal: JsonNode): Signal =
|
||||||
let signalString = jsonSignal{"type"}.getStr
|
let signalString = jsonSignal{"type"}.getStr
|
||||||
|
@ -18,6 +18,7 @@ proc decode*(jsonSignal: JsonNode): Signal =
|
||||||
of SignalType.WhisperFilterAdded: WhisperFilterSignal.fromEvent(jsonSignal)
|
of SignalType.WhisperFilterAdded: WhisperFilterSignal.fromEvent(jsonSignal)
|
||||||
of SignalType.Wallet: WalletSignal.fromEvent(jsonSignal)
|
of SignalType.Wallet: WalletSignal.fromEvent(jsonSignal)
|
||||||
of SignalType.NodeLogin: Json.decode($jsonSignal, NodeSignal)
|
of SignalType.NodeLogin: Json.decode($jsonSignal, NodeSignal)
|
||||||
|
of SignalType.PeerStats: PeerStatsSignal.fromEvent(jsonSignal)
|
||||||
of SignalType.DiscoverySummary: DiscoverySummarySignal.fromEvent(jsonSignal)
|
of SignalType.DiscoverySummary: DiscoverySummarySignal.fromEvent(jsonSignal)
|
||||||
of SignalType.MailserverRequestCompleted: MailserverRequestCompletedSignal.fromEvent(jsonSignal)
|
of SignalType.MailserverRequestCompleted: MailserverRequestCompletedSignal.fromEvent(jsonSignal)
|
||||||
of SignalType.MailserverRequestExpired: MailserverRequestExpiredSignal.fromEvent(jsonSignal)
|
of SignalType.MailserverRequestExpired: MailserverRequestExpiredSignal.fromEvent(jsonSignal)
|
||||||
|
|
|
@ -0,0 +1,14 @@
|
||||||
|
import json
|
||||||
|
import base
|
||||||
|
import signal_type
|
||||||
|
|
||||||
|
type PeerStatsSignal* = ref object of Signal
|
||||||
|
peers*: seq[string]
|
||||||
|
|
||||||
|
proc fromEvent*(T: type PeerStatsSignal, jsonSignal: JsonNode): PeerStatsSignal =
|
||||||
|
result = PeerStatsSignal()
|
||||||
|
result.signalType = SignalType.PeerStats
|
||||||
|
if jsonSignal["event"].kind != JNull:
|
||||||
|
for (node, protocols) in jsonSignal["event"]["peers"].pairs():
|
||||||
|
if protocols.getElems.len != 0:
|
||||||
|
result.peers.add(node)
|
|
@ -19,6 +19,7 @@ type SignalType* {.pure.} = enum
|
||||||
SubscriptionsError = "subscriptions.error"
|
SubscriptionsError = "subscriptions.error"
|
||||||
WhisperFilterAdded = "whisper.filter.added"
|
WhisperFilterAdded = "whisper.filter.added"
|
||||||
CommunityFound = "community.found"
|
CommunityFound = "community.found"
|
||||||
|
PeerStats = "wakuv2.peerstats"
|
||||||
Stats = "stats"
|
Stats = "stats"
|
||||||
Unknown
|
Unknown
|
||||||
|
|
||||||
|
|
|
@ -44,9 +44,13 @@ proc getNodes*(self: FleetConfig, fleet: Fleet, nodeType: FleetNodes = FleetNode
|
||||||
if not self.fleet[$fleet].hasKey($nodeType): return
|
if not self.fleet[$fleet].hasKey($nodeType): return
|
||||||
result = toSeq(self.fleet[$fleet][$nodeType].values)
|
result = toSeq(self.fleet[$fleet][$nodeType].values)
|
||||||
|
|
||||||
proc getMailservers*(self: FleetConfig, fleet: Fleet): Table[string, string] =
|
proc getMailservers*(self: FleetConfig, fleet: Fleet, isWakuV2: bool): Table[string, string] =
|
||||||
if not self.fleet[$fleet].hasKey($FleetNodes.Mailservers):
|
# TODO: If using wakuV2, this assumes that Waku nodes in fleet.status.json are also store nodes.
|
||||||
|
# Maybe it make senses to add a "waku-store" section in case we want to have separate node types?
|
||||||
|
# Discuss with @iurimatias, @cammellos and Vac team
|
||||||
|
let fleetKey = if isWakuV2: $FleetNodes.Waku else: $FleetNodes.Mailservers
|
||||||
|
if not self.fleet[$fleet].hasKey(fleetKey) :
|
||||||
result = initTable[string,string]()
|
result = initTable[string,string]()
|
||||||
return
|
return
|
||||||
result = self.fleet[$fleet][$FleetNodes.Mailservers]
|
result = self.fleet[$fleet][fleetKey]
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue