mirror of
https://github.com/waku-org/nwaku.git
synced 2025-02-24 12:58:35 +00:00
Adding parsePeerInfo and deprecating 'parseRemotePeerInfo' (#1658)
This commit is contained in:
Ivan Folgueira Bande
GitHub
parent
d5979e94da
commit
b2dcb07751
@ -482,7 +482,12 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
|
||||
var storenode: Option[RemotePeerInfo]
|
||||
|
||||
if conf.storenode != "":
|
||||
storenode = some(parseRemotePeerInfo(conf.storenode))
|
||||
let peerInfo = parsePeerInfo(conf.storenode)
|
||||
if peerInfo.isOk():
|
||||
storenode = some(peerInfo.value)
|
||||
else:
|
||||
error "Incorrect conf.storenode", error = peerInfo.error
|
||||
|
||||
elif discoveredNodes.len > 0:
|
||||
echo "Store enabled, but no store nodes configured. Choosing one at random from discovered peers"
|
||||
storenode = some(discoveredNodes[rand(0..len(discoveredNodes) - 1)])
|
||||
@ -509,23 +514,31 @@ proc processInput(rfd: AsyncFD, rng: ref HmacDrbgContext) {.async.} =
|
||||
|
||||
# NOTE Must be mounted after relay
|
||||
if conf.lightpushnode != "":
|
||||
await mountLightPush(node)
|
||||
|
||||
node.mountLightPushClient()
|
||||
node.peerManager.addServicePeer(parseRemotePeerInfo(conf.lightpushnode), WakuLightpushCodec)
|
||||
let peerInfo = parsePeerInfo(conf.lightpushnode)
|
||||
if peerInfo.isOk():
|
||||
await mountLightPush(node)
|
||||
node.mountLightPushClient()
|
||||
node.peerManager.addServicePeer(peerInfo.value, WakuLightpushCodec)
|
||||
else:
|
||||
error "LightPush not mounted. Couldn't parse conf.lightpushnode",
|
||||
error = peerInfo.error
|
||||
|
||||
if conf.filternode != "":
|
||||
await node.mountFilter()
|
||||
await node.mountFilterClient()
|
||||
let peerInfo = parsePeerInfo(conf.filternode)
|
||||
if peerInfo.isOk():
|
||||
await node.mountFilter()
|
||||
await node.mountFilterClient()
|
||||
node.peerManager.addServicePeer(peerInfo.value, WakuFilterCodec)
|
||||
|
||||
node.peerManager.addServicePeer(parseRemotePeerInfo(conf.filternode), WakuFilterCodec)
|
||||
proc filterHandler(pubsubTopic: PubsubTopic, msg: WakuMessage) {.gcsafe.} =
|
||||
trace "Hit filter handler", contentTopic=msg.contentTopic
|
||||
chat.printReceivedMessage(msg)
|
||||
|
||||
proc filterHandler(pubsubTopic: PubsubTopic, msg: WakuMessage) {.gcsafe.} =
|
||||
trace "Hit filter handler", contentTopic=msg.contentTopic
|
||||
await node.subscribe(pubsubTopic=DefaultPubsubTopic, contentTopics=chat.contentTopic, filterHandler)
|
||||
|
||||
chat.printReceivedMessage(msg)
|
||||
|
||||
await node.subscribe(pubsubTopic=DefaultPubsubTopic, contentTopics=chat.contentTopic, filterHandler)
|
||||
else:
|
||||
error "Filter not mounted. Couldn't parse conf.filternode",
|
||||
error = peerInfo.error
|
||||
|
||||
# Subscribe to a topic, if relay is mounted
|
||||
if conf.relay:
|
||||
|
||||
@ -286,12 +286,18 @@ when isMainModule:
|
||||
waitFor connectToNodes(bridge.nodev2, conf.staticnodes)
|
||||
|
||||
if conf.storenode != "":
|
||||
let storePeer = parseRemotePeerInfo(conf.storenode)
|
||||
bridge.nodev2.peerManager.addServicePeer(storePeer, WakuStoreCodec)
|
||||
let storePeer = parsePeerInfo(conf.storenode)
|
||||
if storePeer.isOk():
|
||||
bridge.nodev2.peerManager.addServicePeer(storePeer.value, WakuStoreCodec)
|
||||
else:
|
||||
error "Error parsing conf.storenode", error = storePeer.error
|
||||
|
||||
if conf.filternode != "":
|
||||
let filterPeer = parseRemotePeerInfo(conf.filternode)
|
||||
bridge.nodev2.peerManager.addServicePeer(filterPeer, WakuFilterCodec)
|
||||
let filterPeer = parsePeerInfo(conf.filternode)
|
||||
if filterPeer.isOk():
|
||||
bridge.nodev2.peerManager.addServicePeer(filterPeer.value, WakuFilterCodec)
|
||||
else:
|
||||
error "Error parsing conf.filternode", error = filterPeer.error
|
||||
|
||||
if conf.rpc:
|
||||
let ta = initTAddress(conf.rpcAddress,
|
||||
|
||||
@ -431,13 +431,19 @@ when isMainModule:
|
||||
|
||||
if conf.storenode != "":
|
||||
mountStoreClient(bridge.nodev2)
|
||||
let storeNode = parseRemotePeerInfo(conf.storenode)
|
||||
bridge.nodev2.peerManager.addServicePeer(storeNode, WakuStoreCodec)
|
||||
let storeNode = parsePeerInfo(conf.storenode)
|
||||
if storeNode.isOk():
|
||||
bridge.nodev2.peerManager.addServicePeer(storeNode.value, WakuStoreCodec)
|
||||
else:
|
||||
error "Couldn't parse conf.storenode", error = storeNode.error
|
||||
|
||||
if conf.filternode != "":
|
||||
waitFor mountFilterClient(bridge.nodev2)
|
||||
let filterNode = parseRemotePeerInfo(conf.filternode)
|
||||
bridge.nodev2.peerManager.addServicePeer(filterNode, WakuFilterCodec)
|
||||
let filterNode = parsePeerInfo(conf.filternode)
|
||||
if filterNode.isOk():
|
||||
bridge.nodev2.peerManager.addServicePeer(filterNode.value, WakuFilterCodec)
|
||||
else:
|
||||
error "Couldn't parse conf.filternode", error = filterNode.error
|
||||
|
||||
if conf.rpc:
|
||||
let ta = initTAddress(conf.rpcAddress,
|
||||
|
||||
@ -491,11 +491,11 @@ proc setupProtocols(node: WakuNode, conf: WakuNodeConf,
|
||||
|
||||
mountStoreClient(node)
|
||||
if conf.storenode != "":
|
||||
try:
|
||||
let storenode = parseRemotePeerInfo(conf.storenode)
|
||||
node.peerManager.addServicePeer(storenode, WakuStoreCodec)
|
||||
except CatchableError:
|
||||
return err("failed to set node waku store peer: " & getCurrentExceptionMsg())
|
||||
let storeNode = parsePeerInfo(conf.storenode)
|
||||
if storeNode.isOk():
|
||||
node.peerManager.addServicePeer(storeNode.value, WakuStoreCodec)
|
||||
else:
|
||||
return err("failed to set node waku store peer: " & storeNode.error)
|
||||
|
||||
# NOTE Must be mounted after relay
|
||||
if conf.lightpush:
|
||||
@ -505,12 +505,12 @@ proc setupProtocols(node: WakuNode, conf: WakuNodeConf,
|
||||
return err("failed to mount waku lightpush protocol: " & getCurrentExceptionMsg())
|
||||
|
||||
if conf.lightpushnode != "":
|
||||
try:
|
||||
let lightPushNode = parsePeerInfo(conf.lightpushnode)
|
||||
if lightPushNode.isOk():
|
||||
mountLightPushClient(node)
|
||||
let lightpushnode = parseRemotePeerInfo(conf.lightpushnode)
|
||||
node.peerManager.addServicePeer(lightpushnode, WakuLightPushCodec)
|
||||
except CatchableError:
|
||||
return err("failed to set node waku lightpush peer: " & getCurrentExceptionMsg())
|
||||
node.peerManager.addServicePeer(lightPushNode.value, WakuLightPushCodec)
|
||||
else:
|
||||
return err("failed to set node waku lightpush peer: " & lightPushNode.error)
|
||||
|
||||
# Filter setup. NOTE Must be mounted after relay
|
||||
if conf.filter:
|
||||
@ -520,12 +520,12 @@ proc setupProtocols(node: WakuNode, conf: WakuNodeConf,
|
||||
return err("failed to mount waku filter protocol: " & getCurrentExceptionMsg())
|
||||
|
||||
if conf.filternode != "":
|
||||
try:
|
||||
let filterNode = parsePeerInfo(conf.filternode)
|
||||
if filterNode.isOk():
|
||||
await mountFilterClient(node)
|
||||
let filternode = parseRemotePeerInfo(conf.filternode)
|
||||
node.peerManager.addServicePeer(filternode, WakuFilterCodec)
|
||||
except CatchableError:
|
||||
return err("failed to set node waku filter peer: " & getCurrentExceptionMsg())
|
||||
node.peerManager.addServicePeer(filterNode.value, WakuFilterCodec)
|
||||
else:
|
||||
return err("failed to set node waku filter peer: " & filterNode.error)
|
||||
|
||||
# waku peer exchange setup
|
||||
if (conf.peerExchangeNode != "") or (conf.peerExchange):
|
||||
@ -535,11 +535,11 @@ proc setupProtocols(node: WakuNode, conf: WakuNodeConf,
|
||||
return err("failed to mount waku peer-exchange protocol: " & getCurrentExceptionMsg())
|
||||
|
||||
if conf.peerExchangeNode != "":
|
||||
try:
|
||||
let peerExchangeNode = parseRemotePeerInfo(conf.peerExchangeNode)
|
||||
node.peerManager.addServicePeer(peerExchangeNode, WakuPeerExchangeCodec)
|
||||
except CatchableError:
|
||||
return err("failed to set node waku peer-exchange peer: " & getCurrentExceptionMsg())
|
||||
let peerExchangeNode = parsePeerInfo(conf.peerExchangeNode)
|
||||
if peerExchangeNode.isOk():
|
||||
node.peerManager.addServicePeer(peerExchangeNode.value, WakuPeerExchangeCodec)
|
||||
else:
|
||||
return err("failed to set node waku peer-exchange peer: " & peerExchangeNode.error)
|
||||
|
||||
return ok()
|
||||
|
||||
@ -601,14 +601,12 @@ when defined(waku_exp_store_resume):
|
||||
if address != "":
|
||||
return err("empty peer multiaddres")
|
||||
|
||||
var remotePeer: RemotePeerInfo
|
||||
try:
|
||||
remotePeer = parseRemotePeerInfo(address)
|
||||
except CatchableError:
|
||||
return err("invalid peer multiaddress: " & getCurrentExceptionMsg())
|
||||
let remotePeer = parsePeerInfo(address)
|
||||
if remotePeer.isErr():
|
||||
return err("invalid peer multiaddress: " & remotePeer.error)
|
||||
|
||||
try:
|
||||
await node.resume(some(@[remotePeer]))
|
||||
await node.resume(some(@[remotePeer.value]))
|
||||
except CatchableError:
|
||||
return err("failed to resume messages history: " & getCurrentExceptionMsg())
|
||||
|
||||
|
||||
@ -4,7 +4,6 @@ import
|
||||
std/[options, sequtils],
|
||||
stew/shims/net as stewNet,
|
||||
testutils/unittests,
|
||||
chronicles,
|
||||
chronos,
|
||||
json_rpc/rpcserver,
|
||||
json_rpc/rpcclient,
|
||||
@ -76,7 +75,10 @@ procSuite "Peer Manager":
|
||||
await allFutures(nodes.mapIt(it.start()))
|
||||
await allFutures(nodes.mapIt(it.mountRelay()))
|
||||
|
||||
let nonExistentPeer = parseRemotePeerInfo("/ip4/0.0.0.0/tcp/1000/p2p/16Uiu2HAmL5okWopX7NqZWBUKVqW8iUxCEmd5GMHLVPwCgzYzQv3e")
|
||||
let nonExistentPeerRes = parsePeerInfo("/ip4/0.0.0.0/tcp/1000/p2p/16Uiu2HAmL5okWopX7NqZWBUKVqW8iUxCEmd5GMHLVPwCgzYzQv3e")
|
||||
require nonExistentPeerRes.isOk()
|
||||
|
||||
let nonExistentPeer = nonExistentPeerRes.value
|
||||
|
||||
# Dial non-existent peer from node1
|
||||
let conn1 = await nodes[0].peerManager.dialPeer(nonExistentPeer, WakuFilterCodec)
|
||||
@ -136,9 +138,14 @@ procSuite "Peer Manager":
|
||||
nodes[0].peerManager.peerStore.connectedness(nodes[1].peerInfo.peerId) == NotConnected
|
||||
|
||||
# Failed connection
|
||||
let nonExistentPeer = parseRemotePeerInfo("/ip4/0.0.0.0/tcp/1000/p2p/16Uiu2HAmL5okWopX7NqZWBUKVqW8iUxCEmd5GMHLVPwCgzYzQv3e")
|
||||
let nonExistentPeerRes = parsePeerInfo("/ip4/0.0.0.0/tcp/1000/p2p/16Uiu2HAmL5okWopX7NqZWBUKVqW8iUxCEmd5GMHLVPwCgzYzQv3e")
|
||||
require:
|
||||
nonExistentPeerRes.isOk()
|
||||
|
||||
let nonExistentPeer = nonExistentPeerRes.value
|
||||
require:
|
||||
(await nodes[0].peerManager.connectRelay(nonExistentPeer)) == false
|
||||
|
||||
check:
|
||||
# Cannot connect to node2
|
||||
nodes[0].peerManager.peerStore.connectedness(nonExistentPeer.peerId) == CannotConnect
|
||||
@ -165,7 +172,10 @@ procSuite "Peer Manager":
|
||||
await allFutures(nodes.mapIt(it.start()))
|
||||
await allFutures(nodes.mapIt(it.mountRelay()))
|
||||
|
||||
let nonExistentPeer = parseRemotePeerInfo("/ip4/0.0.0.0/tcp/1000/p2p/16Uiu2HAmL5okWopX7NqZWBUKVqW8iUxCEmd5GMHLVPwCgzYzQv3e")
|
||||
let nonExistentPeerRes = parsePeerInfo("/ip4/0.0.0.0/tcp/1000/p2p/16Uiu2HAmL5okWopX7NqZWBUKVqW8iUxCEmd5GMHLVPwCgzYzQv3e")
|
||||
require nonExistentPeerRes.isOk()
|
||||
|
||||
let nonExistentPeer = nonExistentPeerRes.value
|
||||
|
||||
nodes[0].peerManager.addPeer(nonExistentPeer)
|
||||
|
||||
@ -413,37 +423,41 @@ procSuite "Peer Manager":
|
||||
|
||||
let
|
||||
node = newTestWakuNode(generateSecp256k1Key(), ValidIpAddress.init("0.0.0.0"), Port(0))
|
||||
peer1 = parseRemotePeerInfo("/ip4/0.0.0.0/tcp/30300/p2p/" & basePeerId & "1")
|
||||
peer2 = parseRemotePeerInfo("/ip4/0.0.0.0/tcp/30301/p2p/" & basePeerId & "2")
|
||||
peer3 = parseRemotePeerInfo("/ip4/0.0.0.0/tcp/30302/p2p/" & basePeerId & "3")
|
||||
peer4 = parseRemotePeerInfo("/ip4/0.0.0.0/tcp/30303/p2p/" & basePeerId & "4")
|
||||
peer5 = parseRemotePeerInfo("/ip4/0.0.0.0/tcp/30303/p2p/" & basePeerId & "5")
|
||||
peers = toSeq(1..5)
|
||||
.mapIt(
|
||||
parsePeerInfo("/ip4/0.0.0.0/tcp/30300/p2p/" & basePeerId & $it)
|
||||
)
|
||||
.filterIt(it.isOk())
|
||||
.mapIt(it.value)
|
||||
|
||||
require:
|
||||
peers.len == 5
|
||||
|
||||
# service peers
|
||||
node.peerManager.addServicePeer(peer1, WakuStoreCodec)
|
||||
node.peerManager.addServicePeer(peer2, WakuFilterCodec)
|
||||
node.peerManager.addServicePeer(peer3, WakuLightPushCodec)
|
||||
node.peerManager.addServicePeer(peer4, WakuPeerExchangeCodec)
|
||||
node.peerManager.addServicePeer(peers[0], WakuStoreCodec)
|
||||
node.peerManager.addServicePeer(peers[1], WakuFilterCodec)
|
||||
node.peerManager.addServicePeer(peers[2], WakuLightPushCodec)
|
||||
node.peerManager.addServicePeer(peers[3], WakuPeerExchangeCodec)
|
||||
|
||||
# relay peers (should not be added)
|
||||
node.peerManager.addServicePeer(peer5, WakuRelayCodec)
|
||||
node.peerManager.addServicePeer(peers[4], WakuRelayCodec)
|
||||
|
||||
# all peers are stored in the peerstore
|
||||
check:
|
||||
node.peerManager.peerStore.peers().anyIt(it.peerId == peer1.peerId)
|
||||
node.peerManager.peerStore.peers().anyIt(it.peerId == peer2.peerId)
|
||||
node.peerManager.peerStore.peers().anyIt(it.peerId == peer3.peerId)
|
||||
node.peerManager.peerStore.peers().anyIt(it.peerId == peer4.peerId)
|
||||
node.peerManager.peerStore.peers().anyIt(it.peerId == peers[0].peerId)
|
||||
node.peerManager.peerStore.peers().anyIt(it.peerId == peers[1].peerId)
|
||||
node.peerManager.peerStore.peers().anyIt(it.peerId == peers[2].peerId)
|
||||
node.peerManager.peerStore.peers().anyIt(it.peerId == peers[3].peerId)
|
||||
|
||||
# but the relay peer is not
|
||||
node.peerManager.peerStore.peers().anyIt(it.peerId == peer5.peerId) == false
|
||||
node.peerManager.peerStore.peers().anyIt(it.peerId == peers[4].peerId) == false
|
||||
|
||||
# all service peers are added to its service slot
|
||||
check:
|
||||
node.peerManager.serviceSlots[WakuStoreCodec].peerId == peer1.peerId
|
||||
node.peerManager.serviceSlots[WakuFilterCodec].peerId == peer2.peerId
|
||||
node.peerManager.serviceSlots[WakuLightPushCodec].peerId == peer3.peerId
|
||||
node.peerManager.serviceSlots[WakuPeerExchangeCodec].peerId == peer4.peerId
|
||||
node.peerManager.serviceSlots[WakuStoreCodec].peerId == peers[0].peerId
|
||||
node.peerManager.serviceSlots[WakuFilterCodec].peerId == peers[1].peerId
|
||||
node.peerManager.serviceSlots[WakuLightPushCodec].peerId == peers[2].peerId
|
||||
node.peerManager.serviceSlots[WakuPeerExchangeCodec].peerId == peers[3].peerId
|
||||
|
||||
# but the relay peer is not
|
||||
node.peerManager.serviceSlots.hasKey(WakuRelayCodec) == false
|
||||
@ -458,7 +472,12 @@ procSuite "Peer Manager":
|
||||
storage = nil)
|
||||
|
||||
# Create 3 peer infos
|
||||
let peers = toSeq(1..3).mapIt(parseRemotePeerInfo("/ip4/0.0.0.0/tcp/30300/p2p/" & basePeerId & $it))
|
||||
let peers = toSeq(1..3)
|
||||
.mapIt(parsePeerInfo("/ip4/0.0.0.0/tcp/30300/p2p/" & basePeerId & $it))
|
||||
.filterIt(it.isOk())
|
||||
.mapIt(it.value)
|
||||
require:
|
||||
peers.len == 3
|
||||
|
||||
# Add a peer[0] to the peerstore
|
||||
pm.peerStore[AddressBook][peers[0].peerId] = peers[0].addrs
|
||||
@ -520,8 +539,12 @@ procSuite "Peer Manager":
|
||||
storage = nil)
|
||||
|
||||
# Create 15 peers and add them to the peerstore
|
||||
let peers = toSeq(1..15).mapIt(parseRemotePeerInfo("/ip4/0.0.0.0/tcp/0/p2p/" & $PeerId.random().get()))
|
||||
for p in peers: pm.addPeer(p)
|
||||
let peers = toSeq(1..15)
|
||||
.mapIt(parsePeerInfo("/ip4/0.0.0.0/tcp/0/p2p/" & $PeerId.random().get()))
|
||||
.filterIt(it.isOk())
|
||||
.mapIt(it.value)
|
||||
for p in peers:
|
||||
pm.addPeer(p)
|
||||
|
||||
# Check that we have 15 peers in the peerstore
|
||||
check:
|
||||
|
||||
@ -16,8 +16,11 @@ suite "Utils - Peers":
|
||||
let address = "/ip4/127.0.0.1/tcp/65002/p2p/16Uuu2HBmAcHvhLqQKwSSbX6BG5JLWUDRcaLVrehUVqpw7fz1hbYc"
|
||||
|
||||
## When
|
||||
let remotePeerInfo = parseRemotePeerInfo(address)
|
||||
|
||||
let remotePeerInfoRes = parsePeerInfo(address)
|
||||
require remotePeerInfoRes.isOk()
|
||||
|
||||
let remotePeerInfo = remotePeerInfoRes.value
|
||||
|
||||
## Then
|
||||
check:
|
||||
$(remotePeerInfo.peerId) == "16Uuu2HBmAcHvhLqQKwSSbX6BG5JLWUDRcaLVrehUVqpw7fz1hbYc"
|
||||
@ -29,7 +32,10 @@ suite "Utils - Peers":
|
||||
let address = "/dns/localhost/tcp/65012/p2p/16Uuu2HBmAcHvhLqQKwSSbX6BG5JLWUDRcaLVrehUVqpw7fz1hbYc"
|
||||
|
||||
## When
|
||||
let dnsPeer = parseRemotePeerInfo(address)
|
||||
let dnsPeerRes = parsePeerInfo(address)
|
||||
require dnsPeerRes.isOk()
|
||||
|
||||
let dnsPeer = dnsPeerRes.value
|
||||
|
||||
## Then
|
||||
check:
|
||||
@ -42,7 +48,10 @@ suite "Utils - Peers":
|
||||
let address = "/dnsaddr/localhost/tcp/65022/p2p/16Uuu2HBmAcHvhLqQKwSSbX6BG5JLWUDRcaLVrehUVqpw7fz1hbYc"
|
||||
|
||||
## When
|
||||
let dnsAddrPeer = parseRemotePeerInfo(address)
|
||||
let dnsAddrPeerRes = parsePeerInfo(address)
|
||||
require dnsAddrPeerRes.isOk()
|
||||
|
||||
let dnsAddrPeer = dnsAddrPeerRes.value
|
||||
|
||||
## Then
|
||||
check:
|
||||
@ -55,7 +64,10 @@ suite "Utils - Peers":
|
||||
let address = "/dns4/localhost/tcp/65032/p2p/16Uuu2HBmAcHvhLqQKwSSbX6BG5JLWUDRcaLVrehUVqpw7fz1hbYc"
|
||||
|
||||
## When
|
||||
let dns4Peer = parseRemotePeerInfo(address)
|
||||
let dns4PeerRes = parsePeerInfo(address)
|
||||
require dns4PeerRes.isOk()
|
||||
|
||||
let dns4Peer = dns4PeerRes.value
|
||||
|
||||
# Then
|
||||
check:
|
||||
@ -68,7 +80,10 @@ suite "Utils - Peers":
|
||||
let address = "/dns6/localhost/tcp/65042/p2p/16Uuu2HBmAcHvhLqQKwSSbX6BG5JLWUDRcaLVrehUVqpw7fz1hbYc"
|
||||
|
||||
## When
|
||||
let dns6Peer = parseRemotePeerInfo(address)
|
||||
let dns6PeerRes = parsePeerInfo(address)
|
||||
require dns6PeerRes.isOk()
|
||||
|
||||
let dns6Peer = dns6PeerRes.value
|
||||
|
||||
## Then
|
||||
check:
|
||||
@ -81,46 +96,46 @@ suite "Utils - Peers":
|
||||
let address = "/p2p/$UCH GIBBER!SH"
|
||||
|
||||
## Then
|
||||
expect LPError:
|
||||
discard parseRemotePeerInfo(address)
|
||||
check:
|
||||
parsePeerInfo(address).isErr()
|
||||
|
||||
test "Multiaddr parsing should fail with leading whitespace":
|
||||
## Given
|
||||
let address = " /ip4/127.0.0.1/tcp/65062/p2p/16Uuu2HBmAcHvhLqQKwSSbX6BG5JLWUDRcaLVrehUVqpw7fz1hbYc"
|
||||
|
||||
|
||||
## Then
|
||||
expect LPError:
|
||||
discard parseRemotePeerInfo(address)
|
||||
check:
|
||||
parsePeerInfo(address).isErr()
|
||||
|
||||
test "Multiaddr parsing should fail with trailing whitespace":
|
||||
## Given
|
||||
let address = "/ip4/127.0.0.1/tcp/65072/p2p/16Uuu2HBmAcHvhLqQKwSSbX6BG5JLWUDRcaLVrehUVqpw7fz1hbYc "
|
||||
|
||||
## Then
|
||||
expect LPError:
|
||||
discard parseRemotePeerInfo(address)
|
||||
check:
|
||||
parsePeerInfo(address).isErr()
|
||||
|
||||
test "Multiaddress parsing should fail with invalid IP address":
|
||||
## Given
|
||||
let address = "/ip4/127.0.0.0.1/tcp/65082/p2p/16Uuu2HBmAcHvhLqQKwSSbX6BG5JLWUDRcaLVrehUVqpw7fz1hbYc"
|
||||
|
||||
## Then
|
||||
expect LPError:
|
||||
discard parseRemotePeerInfo(address)
|
||||
check:
|
||||
parsePeerInfo(address).isErr()
|
||||
|
||||
test "Multiaddress parsing should fail with no peer ID":
|
||||
## Given
|
||||
let address = "/ip4/127.0.0.1/tcp/65092"
|
||||
|
||||
# Then
|
||||
expect LPError:
|
||||
discard parseRemotePeerInfo(address)
|
||||
check:
|
||||
parsePeerInfo(address).isErr()
|
||||
|
||||
test "Multiaddress parsing should fail with unsupported transport":
|
||||
## Given
|
||||
let address = "/ip4/127.0.0.1/udp/65102/p2p/16Uuu2HBmAcHvhLqQKwSSbX6BG5JLWUDRcaLVrehUVqpw7fz1hbYc"
|
||||
|
||||
## Then
|
||||
expect ValueError:
|
||||
discard parseRemotePeerInfo(address)
|
||||
check:
|
||||
parsePeerInfo(address).isErr()
|
||||
|
||||
|
||||
@ -391,7 +391,7 @@ procSuite "Waku v2 Rest API - Store":
|
||||
$response.contentType == $MIMETYPE_TEXT
|
||||
response.data.messages.len == 0
|
||||
response.data.error_message.get ==
|
||||
"Failed parsing remote peer info [multiaddress: Invalid MultiAddress, must start with `/`]"
|
||||
"Failed parsing remote peer info [MultiAddress.init [multiaddress: Invalid MultiAddress, must start with `/`]]"
|
||||
|
||||
await restServer.stop()
|
||||
await restServer.closeWait()
|
||||
|
||||
@ -122,8 +122,14 @@ proc main(rng: ref HmacDrbgContext): Future[int] {.async.} =
|
||||
protocols = conf.protocols,
|
||||
logLevel = conf.logLevel
|
||||
|
||||
let peerRes = parsePeerInfo(conf.address)
|
||||
if peerRes.isErr():
|
||||
error "Couldn't parse 'conf.address'", error = peerRes.error
|
||||
return 1
|
||||
|
||||
let peer = peerRes.value
|
||||
|
||||
let
|
||||
peer: RemotePeerInfo = parseRemotePeerInfo(conf.address)
|
||||
nodeKey = crypto.PrivateKey.random(Secp256k1, rng[])[]
|
||||
bindIp = ValidIpAddress.init("0.0.0.0")
|
||||
wsBindPort = Port(conf.nodePort + WebSocketPortOffset)
|
||||
|
||||
@ -45,7 +45,11 @@ proc installAdminApiHandlers*(node: WakuNode, rpcsrv: RpcServer) =
|
||||
debug "post_waku_v2_admin_v1_peers"
|
||||
|
||||
for i, peer in peers:
|
||||
let connOk = await node.peerManager.connectRelay(parseRemotePeerInfo(peer), source="rpc")
|
||||
let peerInfo = parsePeerInfo(peer)
|
||||
if peerInfo.isErr():
|
||||
raise newException(ValueError, "Couldn't parse remote peer info: " & peerInfo.error)
|
||||
|
||||
let connOk = await node.peerManager.connectRelay(peerInfo.value, source="rpc")
|
||||
if not connOk:
|
||||
raise newException(ValueError, "Failed to connect to peer at index: " & $i & " " & $peer)
|
||||
|
||||
|
||||
@ -411,9 +411,11 @@ proc connectToNodes*(pm: PeerManager,
|
||||
|
||||
var futConns: seq[Future[bool]]
|
||||
for node in nodes:
|
||||
let node = when node is string: parseRemotePeerInfo(node)
|
||||
else: node
|
||||
futConns.add(pm.connectRelay(node))
|
||||
let node = parsePeerInfo(node)
|
||||
if node.isOk():
|
||||
futConns.add(pm.connectRelay(node.value))
|
||||
else:
|
||||
error "Couldn't parse node info", error = node.error
|
||||
|
||||
await allFutures(futConns)
|
||||
let successfulConns = futConns.mapIt(it.read()).countIt(true)
|
||||
|
||||
@ -439,8 +439,12 @@ proc filterSubscribe*(node: WakuNode, pubsubTopic: PubsubTopic, contentTopics: C
|
||||
error "cannot register filter subscription to topic", error="waku filter client is nil"
|
||||
return
|
||||
|
||||
let remotePeer = when peer is string: parseRemotePeerInfo(peer)
|
||||
else: peer
|
||||
let remotePeerRes = parsePeerInfo(peer)
|
||||
if remotePeerRes.isErr():
|
||||
error "Couldn't parse the peer info properly", error = remotePeerRes.error
|
||||
return
|
||||
|
||||
let remotePeer = remotePeerRes.value
|
||||
|
||||
info "registering filter subscription to content", pubsubTopic=pubsubTopic, contentTopics=contentTopics, peer=remotePeer.peerId
|
||||
|
||||
@ -466,8 +470,12 @@ proc filterUnsubscribe*(node: WakuNode, pubsubTopic: PubsubTopic, contentTopics:
|
||||
error "cannot unregister filter subscription to content", error="waku filter client is nil"
|
||||
return
|
||||
|
||||
let remotePeer = when peer is string: parseRemotePeerInfo(peer)
|
||||
else: peer
|
||||
let remotePeerRes = parsePeerInfo(peer)
|
||||
if remotePeerRes.isErr():
|
||||
error "couldn't parse remotePeerInfo", error = remotePeerRes.error
|
||||
return
|
||||
|
||||
let remotePeer = remotePeerRes.value
|
||||
|
||||
info "deregistering filter subscription to content", pubsubTopic=pubsubTopic, contentTopics=contentTopics, peer=remotePeer.peerId
|
||||
|
||||
@ -783,16 +791,19 @@ proc fetchPeerExchangePeers*(node: Wakunode, amount: uint64) {.async, raises: [D
|
||||
warn "Failed to retrieve peer info via peer exchange protocol", error = pxPeersRes.error
|
||||
|
||||
# TODO: Move to application module (e.g., wakunode2.nim)
|
||||
proc setPeerExchangePeer*(node: WakuNode, peer: RemotePeerInfo|string) {.raises: [Defect, ValueError, LPError].} =
|
||||
proc setPeerExchangePeer*(node: WakuNode, peer: RemotePeerInfo|string) =
|
||||
if node.wakuPeerExchange.isNil():
|
||||
error "could not set peer, waku peer-exchange is nil"
|
||||
return
|
||||
|
||||
info "Set peer-exchange peer", peer=peer
|
||||
|
||||
let remotePeer = when peer is string: parseRemotePeerInfo(peer)
|
||||
else: peer
|
||||
node.peerManager.addPeer(remotePeer, WakuPeerExchangeCodec)
|
||||
let remotePeerRes = parsePeerInfo(peer)
|
||||
if remotePeerRes.isErr():
|
||||
error "could not parse peer info", error = remotePeerRes.error
|
||||
return
|
||||
|
||||
node.peerManager.addPeer(remotePeerRes.value, WakuPeerExchangeCodec)
|
||||
waku_px_peers.inc()
|
||||
|
||||
|
||||
|
||||
@ -110,37 +110,42 @@ func getTransportProtocol(typedR: TypedRecord): Option[IpTransportProtocol] =
|
||||
|
||||
return none(IpTransportProtocol)
|
||||
|
||||
## Parses a fully qualified peer multiaddr, in the
|
||||
## format `(ip4|ip6)/tcp/p2p`, into dialable PeerInfo
|
||||
proc parseRemotePeerInfo*(address: string): RemotePeerInfo {.raises: [Defect, ValueError, LPError].}=
|
||||
let multiAddr = MultiAddress.init(address).tryGet()
|
||||
|
||||
var
|
||||
nwPart, tcpPart, p2pPart, wsPart, wssPart: MultiAddress
|
||||
proc parsePeerInfo*(peer: RemotePeerInfo|string):
|
||||
Result[RemotePeerInfo, string] =
|
||||
## Parses a fully qualified peer multiaddr, in the
|
||||
## format `(ip4|ip6)/tcp/p2p`, into dialable PeerInfo
|
||||
|
||||
if peer is RemotePeerInfo:
|
||||
return ok(cast[RemotePeerInfo](peer))
|
||||
|
||||
let multiAddr = ? MultiAddress.init(cast[string](peer))
|
||||
.mapErr(proc(err: string):
|
||||
string = "MultiAddress.init [" & err & "]")
|
||||
|
||||
var p2pPart: MultiAddress
|
||||
var wireAddr = MultiAddress()
|
||||
for addrPart in multiAddr.items():
|
||||
case addrPart[].protoName()[]
|
||||
# All protocols listed here: https://github.com/multiformats/multiaddr/blob/b746a7d014e825221cc3aea6e57a92d78419990f/protocols.csv
|
||||
of "ip4", "ip6", "dns", "dnsaddr", "dns4", "dns6":
|
||||
nwPart = addrPart.tryGet()
|
||||
of "tcp":
|
||||
tcpPart = addrPart.tryGet()
|
||||
of "p2p":
|
||||
p2pPart = addrPart.tryGet()
|
||||
of "ws":
|
||||
wsPart = addrPart.tryGet()
|
||||
of "wss":
|
||||
wssPart = addrPart.tryGet()
|
||||
p2pPart = ? addrPart.mapErr(proc(err: string):string = "Error getting p2pPart [" & err & "]")
|
||||
of "ip4", "ip6", "dns", "dnsaddr", "dns4", "dns6", "tcp", "ws", "wss":
|
||||
let val = ? addrPart.mapErr(proc(err: string):string = "Error getting addrPart [" & err & "]")
|
||||
? wireAddr.append(val).mapErr(proc(err: string):string = "Error appending addrPart [" & err & "]")
|
||||
|
||||
# nim-libp2p dialing requires remote peers to be initialised with a peerId and a wire address
|
||||
let
|
||||
peerIdStr = p2pPart.toString()[].split("/")[^1]
|
||||
let p2pPartStr = p2pPart.toString()[]
|
||||
if not p2pPartStr.contains("/"):
|
||||
return err("Error in parsePeerInfo: p2p part should contain /")
|
||||
|
||||
wireAddr = nwPart & tcpPart & wsPart & wssPart
|
||||
if (not wireAddr.validWireAddr()):
|
||||
raise newException(ValueError, "Invalid node multi-address")
|
||||
let peerId = ? PeerID.init(p2pPartStr.split("/")[^1])
|
||||
.mapErr(proc (e:cstring):string = cast[string](e))
|
||||
|
||||
if not wireAddr.validWireAddr():
|
||||
return err("Error in parsePeerInfo: Invalid node multiaddress")
|
||||
|
||||
return ok(RemotePeerInfo.init(peerId, @[wireAddr]))
|
||||
|
||||
return RemotePeerInfo.init(peerIdStr, @[wireAddr])
|
||||
|
||||
# Checks whether the peerAddr parameter represents a valid p2p multiaddress.
|
||||
# The param must be in the format `(ip4|ip6)/tcp/p2p/$peerId` but URL-encoded
|
||||
@ -150,12 +155,14 @@ proc parseUrlPeerAddr*(peerAddr: Option[string]):
|
||||
if not peerAddr.isSome() or peerAddr.get() == "":
|
||||
return ok(none(RemotePeerInfo))
|
||||
|
||||
try:
|
||||
let parsedAddr = decodeUrl(peerAddr.get())
|
||||
return ok(some(parseRemotePeerInfo(parsedAddr)))
|
||||
except Exception:
|
||||
let parsedAddr = decodeUrl(peerAddr.get())
|
||||
let parsedPeerInfo = parsePeerInfo(parsedAddr)
|
||||
|
||||
if parsedPeerInfo.isOk():
|
||||
return ok(some(parsedPeerInfo.value))
|
||||
else:
|
||||
return err("Failed parsing remote peer info [" &
|
||||
getCurrentExceptionMsg() & "]")
|
||||
parsedPeerInfo.error & "]")
|
||||
|
||||
## Converts an ENR to dialable RemotePeerInfo
|
||||
proc toRemotePeerInfo*(enr: enr.Record): Result[RemotePeerInfo, cstring] =
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user