mirror of https://github.com/status-im/nim-eth.git
132 lines
4.3 KiB
Nim
132 lines
4.3 KiB
Nim
# Copyright (c) 2021-2022 Status Research & Development GmbH
|
|
# Licensed and distributed under either of
|
|
# * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
|
|
# * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
|
|
# at your option. This file may not be copied, modified, or distributed except according to those terms.
|
|
|
|
{.push raises: [Defect].}
|
|
|
|
import
|
|
std/[hashes, sugar],
|
|
chronos, chronicles,
|
|
../p2p/discoveryv5/[protocol, messages, encoding],
|
|
./utp_router,
|
|
../keys
|
|
|
|
export utp_router, protocol
|
|
|
|
logScope:
|
|
topics = "utp_discv5_protocol"
|
|
|
|
type
|
|
NodeAddress* = object
|
|
nodeId*: NodeId
|
|
address*: Address
|
|
|
|
UtpDiscv5Protocol* = ref object of TalkProtocol
|
|
prot: protocol.Protocol
|
|
router: UtpRouter[NodeAddress]
|
|
|
|
proc init*(T: type NodeAddress, nodeId: NodeId, address: Address): NodeAddress =
|
|
NodeAddress(nodeId: nodeId, address: address)
|
|
|
|
proc init*(T: type NodeAddress, node: Node): Option[NodeAddress] =
|
|
node.address.map((address: Address) => NodeAddress(nodeId: node.id, address: address))
|
|
|
|
proc hash(x: NodeAddress): Hash =
|
|
var h = 0
|
|
h = h !& x.nodeId.hash
|
|
h = h !& x.address.hash
|
|
!$h
|
|
|
|
proc hash(x: UtpSocketKey[NodeAddress]): Hash =
|
|
var h = 0
|
|
h = h !& x.remoteAddress.hash
|
|
h = h !& x.rcvId.hash
|
|
!$h
|
|
|
|
func `$`*(x: UtpSocketKey[NodeAddress]): string =
|
|
"(remoteId: " & $x.remoteAddress.nodeId &
|
|
", remoteAddress: " & $x.remoteAddress.address &
|
|
", rcvId: "& $x.rcvId &
|
|
")"
|
|
|
|
proc talkReqDirect(p: protocol.Protocol, n: NodeAddress, protocol, request: seq[byte]): Future[void] =
|
|
let
|
|
reqId = RequestId.init(p.rng[])
|
|
message = encodeMessage(TalkReqMessage(protocol: protocol, request: request), reqId)
|
|
|
|
(data, nonce) = encodeMessagePacket(p.rng[], p.codec, n.nodeId, n.address, message)
|
|
|
|
trace "Send message packet", dstId = n.nodeId, address = n.address, kind = MessageKind.talkreq
|
|
p.send(n.address, data)
|
|
|
|
proc initSendCallback(
|
|
t: protocol.Protocol, subProtocolName: seq[byte]): SendCallback[NodeAddress] =
|
|
return (
|
|
proc (to: NodeAddress, data: seq[byte]): Future[void] =
|
|
let fut = newFuture[void]()
|
|
# hidden assumption here is that nodes already have established discv5 session
|
|
# between each other. In our use case this should be true as openning stream
|
|
# is only done after succesful OFFER/ACCEPT or FINDCONTENT/CONTENT exchange
|
|
# which forces nodes to establish session between each other.
|
|
discard t.talkReqDirect(to, subProtocolName, data)
|
|
fut.complete()
|
|
return fut
|
|
)
|
|
|
|
proc messageHandler(protocol: TalkProtocol, request: seq[byte],
|
|
srcId: NodeId, srcUdpAddress: Address): seq[byte] =
|
|
let
|
|
p = UtpDiscv5Protocol(protocol)
|
|
nodeAddress = NodeAddress.init(srcId, srcUdpAddress)
|
|
debug "Received utp payload from known node. Start processing",
|
|
nodeId = nodeAddress.nodeId, address = nodeAddress.address
|
|
asyncSpawn p.router.processIncomingBytes(request, nodeAddress)
|
|
|
|
proc new*(
|
|
T: type UtpDiscv5Protocol,
|
|
p: protocol.Protocol,
|
|
subProtocolName: seq[byte],
|
|
acceptConnectionCb: AcceptConnectionCallback[NodeAddress],
|
|
allowConnectionCb: AllowConnectionCallback[NodeAddress] = nil,
|
|
socketConfig: SocketConfig = SocketConfig.init()): UtpDiscv5Protocol =
|
|
doAssert(not(isNil(acceptConnectionCb)))
|
|
|
|
let router = UtpRouter[NodeAddress].new(
|
|
acceptConnectionCb,
|
|
allowConnectionCb,
|
|
socketConfig,
|
|
p.rng
|
|
)
|
|
router.sendCb = initSendCallback(p, subProtocolName)
|
|
|
|
let prot = UtpDiscv5Protocol(
|
|
protocolHandler: messageHandler,
|
|
prot: p,
|
|
router: router
|
|
)
|
|
|
|
p.registerTalkProtocol(subProtocolName, prot).expect(
|
|
"Only one protocol should have this id"
|
|
)
|
|
prot
|
|
|
|
proc connectTo*(r: UtpDiscv5Protocol, address: NodeAddress):
|
|
Future[ConnectionResult[NodeAddress]] =
|
|
return r.router.connectTo(address)
|
|
|
|
proc connectTo*(r: UtpDiscv5Protocol, address: NodeAddress, connectionId: uint16):
|
|
Future[ConnectionResult[NodeAddress]] =
|
|
return r.router.connectTo(address, connectionId)
|
|
|
|
proc shutdown*(r: UtpDiscv5Protocol) =
|
|
## Closes all managed utp connections in background (does not close discovery,
|
|
## this is up to user)
|
|
r.router.shutdown()
|
|
|
|
proc shutdownWait*(r: UtpDiscv5Protocol) {.async.} =
|
|
## Closes all managed utp connections in background (does not close discovery,
|
|
## this is up to user)
|
|
await r.router.shutdownWait()
|