Add peer lifetime feature for PeerInfo. (#77)
* Add peer lifetime feature for PeerInfo. Refactor peerinfo to use openarrays instead of sequences. Fix tests and examples to use arrays instead of sequences. * Add access to lifetime Future[T] itself.
This commit is contained in:
parent
23712ecf3b
commit
540e79a430
|
@ -68,7 +68,7 @@ proc dialPeer(p: ChatProto, address: string) {.async, gcsafe.} =
|
|||
quit("invalid or incompelete peerId")
|
||||
|
||||
var remotePeer = PeerInfo.init(parts[^1],
|
||||
@[MultiAddress.init(address)])
|
||||
[MultiAddress.init(address)])
|
||||
|
||||
echo &"dialing peer: {address}"
|
||||
p.conn = await p.switch.dial(remotePeer, ChatCodec)
|
||||
|
|
|
@ -8,6 +8,7 @@
|
|||
## those terms.
|
||||
|
||||
import options
|
||||
import chronos
|
||||
import peer, multiaddress, crypto/crypto
|
||||
|
||||
## A peer can be constructed in one of tree ways:
|
||||
|
@ -27,53 +28,70 @@ type
|
|||
peerId*: PeerID
|
||||
addrs*: seq[MultiAddress]
|
||||
protocols*: seq[string]
|
||||
lifefut: Future[void]
|
||||
case keyType*: KeyType:
|
||||
of HasPrivate:
|
||||
privateKey*: PrivateKey
|
||||
of HasPublic:
|
||||
key: Option[PublicKey]
|
||||
|
||||
proc init*(p: typedesc[PeerInfo],
|
||||
key: PrivateKey,
|
||||
addrs: seq[MultiAddress] = @[],
|
||||
protocols: seq[string] = @[]): PeerInfo {.inline.} =
|
||||
template postInit(peerinfo: PeerInfo,
|
||||
addrs: openarray[MultiAddress],
|
||||
protocols: openarray[string]) =
|
||||
if len(addrs) > 0:
|
||||
peerinfo.addrs = @addrs
|
||||
if len(protocols) > 0:
|
||||
peerinfo.protocols = @protocols
|
||||
peerinfo.lifefut = newFuture[void]("libp2p.peerinfo.lifetime")
|
||||
|
||||
result = PeerInfo(keyType: HasPrivate,
|
||||
peerId: PeerID.init(key),
|
||||
privateKey: key,
|
||||
addrs: addrs,
|
||||
protocols: protocols)
|
||||
proc init*(p: typedesc[PeerInfo], key: PrivateKey,
|
||||
addrs: openarray[MultiAddress] = [],
|
||||
protocols: openarray[string] = []): PeerInfo {.inline.} =
|
||||
result = PeerInfo(keyType: HasPrivate, peerId: PeerID.init(key),
|
||||
privateKey: key)
|
||||
result.postInit(addrs, protocols)
|
||||
|
||||
proc init*(p: typedesc[PeerInfo],
|
||||
peerId: PeerID,
|
||||
addrs: seq[MultiAddress] = @[],
|
||||
protocols: seq[string] = @[]): PeerInfo {.inline.} =
|
||||
proc init*(p: typedesc[PeerInfo], peerId: PeerID,
|
||||
addrs: openarray[MultiAddress] = [],
|
||||
protocols: openarray[string] = []): PeerInfo {.inline.} =
|
||||
result = PeerInfo(keyType: HasPublic, peerId: peerId)
|
||||
result.postInit(addrs, protocols)
|
||||
|
||||
PeerInfo(keyType: HasPublic,
|
||||
peerId: peerId,
|
||||
addrs: addrs,
|
||||
protocols: protocols)
|
||||
proc init*(p: typedesc[PeerInfo], peerId: string,
|
||||
addrs: openarray[MultiAddress] = [],
|
||||
protocols: openarray[string] = []): PeerInfo {.inline.} =
|
||||
result = PeerInfo(keyType: HasPublic, peerId: PeerID.init(peerId))
|
||||
result.postInit(addrs, protocols)
|
||||
|
||||
proc init*(p: typedesc[PeerInfo],
|
||||
peerId: string,
|
||||
addrs: seq[MultiAddress] = @[],
|
||||
protocols: seq[string] = @[]): PeerInfo {.inline.} =
|
||||
proc init*(p: typedesc[PeerInfo], key: PublicKey,
|
||||
addrs: openarray[MultiAddress] = [],
|
||||
protocols: openarray[string] = []): PeerInfo {.inline.} =
|
||||
result = PeerInfo(keyType: HasPublic, peerId: PeerID.init(key),
|
||||
key: some(key))
|
||||
result.postInit(addrs, protocols)
|
||||
|
||||
PeerInfo(keyType: HasPublic,
|
||||
peerId: PeerID.init(peerId),
|
||||
addrs: addrs,
|
||||
protocols: protocols)
|
||||
proc close*(p: PeerInfo) {.inline.} =
|
||||
p.lifefut.complete()
|
||||
|
||||
proc init*(p: typedesc[PeerInfo],
|
||||
key: PublicKey,
|
||||
addrs: seq[MultiAddress] = @[],
|
||||
protocols: seq[string] = @[]): PeerInfo {.inline.} =
|
||||
proc join*(p: PeerInfo): Future[void] {.inline.} =
|
||||
var retFuture = newFuture[void]()
|
||||
proc continuation(udata: pointer) {.gcsafe.} =
|
||||
if not(retFuture.finished()):
|
||||
retFuture.complete()
|
||||
proc cancellation(udata: pointer) {.gcsafe.} =
|
||||
p.lifefut.removeCallback(continuation)
|
||||
if p.lifefut.finished:
|
||||
retFuture.complete()
|
||||
else:
|
||||
p.lifefut.addCallback(continuation)
|
||||
retFuture.cancelCallback = cancellation
|
||||
return retFuture
|
||||
|
||||
PeerInfo(keyType: HasPublic,
|
||||
peerId: PeerID.init(key),
|
||||
key: some(key),
|
||||
addrs: addrs,
|
||||
protocols: protocols)
|
||||
proc isClosed*(p: PeerInfo): bool {.inline.} =
|
||||
result = p.lifefut.finished()
|
||||
|
||||
proc lifeFuture*(p: PeerInfo): Future[void] {.inline.} =
|
||||
result = p.lifefut
|
||||
|
||||
proc publicKey*(p: PeerInfo): Option[PublicKey] {.inline.} =
|
||||
if p.keyType == HasPublic:
|
||||
|
@ -87,7 +105,7 @@ proc publicKey*(p: PeerInfo): Option[PublicKey] {.inline.} =
|
|||
result = some(p.privateKey.getKey())
|
||||
|
||||
proc id*(p: PeerInfo): string {.inline.} =
|
||||
p.peerId.pretty
|
||||
result = p.peerId.pretty()
|
||||
|
||||
proc `$`*(p: PeerInfo): string =
|
||||
result.add("PeerID: ")
|
||||
|
|
|
@ -18,7 +18,7 @@ proc newStandardSwitch*(privKey = none(PrivateKey),
|
|||
|
||||
let
|
||||
seckey = privKey.get(otherwise = PrivateKey.random(ECDSA))
|
||||
peerInfo = PeerInfo.init(seckey, @[address])
|
||||
peerInfo = PeerInfo.init(seckey, [address])
|
||||
mplexProvider = newMuxerProvider(createMplex, MplexCodec)
|
||||
transports = @[Transport(newTransport(TcpTransport))]
|
||||
muxers = {MplexCodec: mplexProvider}.toTable
|
||||
|
|
|
@ -144,6 +144,10 @@ proc cleanupConn(s: Switch, conn: Connection) {.async, gcsafe.} =
|
|||
await s.connections[id].close()
|
||||
s.connections.del(id)
|
||||
|
||||
# TODO: Investigate cleanupConn() always called twice for one peer.
|
||||
if not(conn.peerInfo.isClosed()):
|
||||
conn.peerInfo.close()
|
||||
|
||||
proc disconnect*(s: Switch, peer: PeerInfo) {.async, gcsafe.} =
|
||||
let conn = s.connections.getOrDefault(peer.id)
|
||||
if not isNil(conn):
|
||||
|
|
|
@ -19,9 +19,9 @@ suite "Identify":
|
|||
let ma: MultiAddress = Multiaddress.init("/ip4/0.0.0.0/tcp/0")
|
||||
let remoteSecKey = PrivateKey.random(RSA)
|
||||
let remotePeerInfo = PeerInfo.init(remoteSecKey,
|
||||
@[ma],
|
||||
@["/test/proto1/1.0.0",
|
||||
"/test/proto2/1.0.0"])
|
||||
[ma],
|
||||
["/test/proto1/1.0.0",
|
||||
"/test/proto2/1.0.0"])
|
||||
var serverFut: Future[void]
|
||||
let identifyProto1 = newIdentify(remotePeerInfo)
|
||||
let msListen = newMultistream()
|
||||
|
@ -37,7 +37,7 @@ suite "Identify":
|
|||
let transport2: TcpTransport = newTransport(TcpTransport)
|
||||
let conn = await transport2.dial(transport1.ma)
|
||||
|
||||
var peerInfo = PeerInfo.init(PrivateKey.random(RSA), @[ma])
|
||||
var peerInfo = PeerInfo.init(PrivateKey.random(RSA), [ma])
|
||||
let identifyProto2 = newIdentify(peerInfo)
|
||||
discard await msDial.select(conn, IdentifyCodec)
|
||||
let id = await identifyProto2.identify(conn, remotePeerInfo)
|
||||
|
@ -59,7 +59,7 @@ suite "Identify":
|
|||
test "handle failed identify":
|
||||
proc testHandleError() {.async.} =
|
||||
let ma: MultiAddress = Multiaddress.init("/ip4/0.0.0.0/tcp/0")
|
||||
var remotePeerInfo = PeerInfo.init(PrivateKey.random(RSA), @[ma])
|
||||
var remotePeerInfo = PeerInfo.init(PrivateKey.random(RSA), [ma])
|
||||
let identifyProto1 = newIdentify(remotePeerInfo)
|
||||
let msListen = newMultistream()
|
||||
|
||||
|
@ -74,7 +74,7 @@ suite "Identify":
|
|||
let transport2: TcpTransport = newTransport(TcpTransport)
|
||||
let conn = await transport2.dial(transport1.ma)
|
||||
|
||||
var localPeerInfo = PeerInfo.init(PrivateKey.random(RSA), @[ma])
|
||||
var localPeerInfo = PeerInfo.init(PrivateKey.random(RSA), [ma])
|
||||
let identifyProto2 = newIdentify(localPeerInfo)
|
||||
discard await msDial.select(conn, IdentifyCodec)
|
||||
discard await identifyProto2.identify(conn, PeerInfo.init(PrivateKey.random(RSA)))
|
||||
|
|
|
@ -72,7 +72,7 @@ proc createNode*(privKey: Option[PrivateKey] = none(PrivateKey),
|
|||
if privKey.isNone:
|
||||
seckey = some(PrivateKey.random(RSA))
|
||||
|
||||
var peerInfo = NativePeerInfo.init(seckey.get(), @[Multiaddress.init(address)])
|
||||
var peerInfo = NativePeerInfo.init(seckey.get(), [Multiaddress.init(address)])
|
||||
proc createMplex(conn: Connection): Muxer = newMplex(conn)
|
||||
let mplexProvider = newMuxerProvider(createMplex, MplexCodec)
|
||||
let transports = @[Transport(newTransport(TcpTransport))]
|
||||
|
|
|
@ -1,5 +1,6 @@
|
|||
|
||||
import unittest, options
|
||||
import chronos
|
||||
import ../libp2p/crypto/crypto,
|
||||
../libp2p/peerinfo,
|
||||
../libp2p/peer
|
||||
|
@ -51,3 +52,16 @@ suite "PeerInfo":
|
|||
test "Should return some if pubkey is present in id":
|
||||
let peerInfo = PeerInfo.init(PeerID.init(PrivateKey.random(Ed25519)))
|
||||
check peerInfo.publicKey.isSome
|
||||
|
||||
test "join() and isClosed() test":
|
||||
proc testJoin(): Future[bool] {.async, gcsafe.} =
|
||||
let peerInfo = PeerInfo.init(PeerID.init(PrivateKey.random(Ed25519)))
|
||||
check peerInfo.isClosed() == false
|
||||
var joinFut = peerInfo.join()
|
||||
check joinFut.finished() == false
|
||||
peerInfo.close()
|
||||
await wait(joinFut, 100.milliseconds)
|
||||
check peerInfo.isClosed() == true
|
||||
check (joinFut.finished() == true) and (joinFut.cancelled() == false)
|
||||
result = true
|
||||
check waitFor(testJoin()) == true
|
||||
|
|
Loading…
Reference in New Issue