test for openned/closed resource
This commit is contained in:
parent
6e0eb93d4f
commit
a4277cf39a
|
@ -2,11 +2,12 @@
|
||||||
|
|
||||||
import unittest, tables
|
import unittest, tables
|
||||||
import chronos
|
import chronos
|
||||||
import chronicles
|
import stew/byteutils
|
||||||
import nimcrypto/sysrand
|
import nimcrypto/sysrand
|
||||||
import ../libp2p/[errors,
|
import ../libp2p/[errors,
|
||||||
switch,
|
switch,
|
||||||
multistream,
|
multistream,
|
||||||
|
standard_setup,
|
||||||
stream/bufferstream,
|
stream/bufferstream,
|
||||||
protocols/identify,
|
protocols/identify,
|
||||||
connection,
|
connection,
|
||||||
|
@ -30,25 +31,6 @@ const
|
||||||
type
|
type
|
||||||
TestProto = ref object of LPProtocol
|
TestProto = ref object of LPProtocol
|
||||||
|
|
||||||
proc createSwitch(ma: MultiAddress): (Switch, PeerInfo) =
|
|
||||||
var peerInfo: PeerInfo = PeerInfo.init(PrivateKey.random(ECDSA).tryGet())
|
|
||||||
peerInfo.addrs.add(ma)
|
|
||||||
let identify = newIdentify(peerInfo)
|
|
||||||
|
|
||||||
proc createMplex(conn: Connection): Muxer =
|
|
||||||
result = newMplex(conn)
|
|
||||||
|
|
||||||
let mplexProvider = newMuxerProvider(createMplex, MplexCodec)
|
|
||||||
let transports = @[Transport(TcpTransport.init())]
|
|
||||||
let muxers = [(MplexCodec, mplexProvider)].toTable()
|
|
||||||
let secureManagers = [Secure(newSecio(peerInfo.privateKey))]
|
|
||||||
let switch = newSwitch(peerInfo,
|
|
||||||
transports,
|
|
||||||
identify,
|
|
||||||
muxers,
|
|
||||||
secureManagers)
|
|
||||||
result = (switch, peerInfo)
|
|
||||||
|
|
||||||
suite "Switch":
|
suite "Switch":
|
||||||
teardown:
|
teardown:
|
||||||
for tracker in testTrackers():
|
for tracker in testTrackers():
|
||||||
|
@ -57,42 +39,53 @@ suite "Switch":
|
||||||
|
|
||||||
test "e2e use switch dial proto string":
|
test "e2e use switch dial proto string":
|
||||||
proc testSwitch() {.async, gcsafe.} =
|
proc testSwitch() {.async, gcsafe.} =
|
||||||
let ma1: MultiAddress = Multiaddress.init("/ip4/0.0.0.0/tcp/0").tryGet()
|
|
||||||
let ma2: MultiAddress = Multiaddress.init("/ip4/0.0.0.0/tcp/0").tryGet()
|
|
||||||
|
|
||||||
var peerInfo1, peerInfo2: PeerInfo
|
|
||||||
var switch1, switch2: Switch
|
|
||||||
var awaiters: seq[Future[void]]
|
|
||||||
|
|
||||||
(switch1, peerInfo1) = createSwitch(ma1)
|
|
||||||
|
|
||||||
let done = newFuture[void]()
|
let done = newFuture[void]()
|
||||||
|
|
||||||
proc handle(conn: Connection, proto: string) {.async, gcsafe.} =
|
proc handle(conn: Connection, proto: string) {.async, gcsafe.} =
|
||||||
let msg = cast[string](await conn.readLp(1024))
|
try:
|
||||||
check "Hello!" == msg
|
let msg = string.fromBytes(await conn.readLp(1024))
|
||||||
await conn.writeLp("Hello!")
|
check "Hello!" == msg
|
||||||
await conn.close()
|
await conn.writeLp("Hello!")
|
||||||
done.complete()
|
finally:
|
||||||
|
await conn.close()
|
||||||
|
done.complete()
|
||||||
|
|
||||||
let testProto = new TestProto
|
let testProto = new TestProto
|
||||||
testProto.codec = TestCodec
|
testProto.codec = TestCodec
|
||||||
testProto.handler = handle
|
testProto.handler = handle
|
||||||
|
|
||||||
|
let switch1 = newStandardSwitch()
|
||||||
switch1.mount(testProto)
|
switch1.mount(testProto)
|
||||||
|
|
||||||
(switch2, peerInfo2) = createSwitch(ma2)
|
let switch2 = newStandardSwitch()
|
||||||
|
var awaiters: seq[Future[void]]
|
||||||
awaiters.add(await switch1.start())
|
awaiters.add(await switch1.start())
|
||||||
awaiters.add(await switch2.start())
|
awaiters.add(await switch2.start())
|
||||||
|
|
||||||
let conn = await switch2.dial(switch1.peerInfo, TestCodec)
|
let conn = await switch2.dial(switch1.peerInfo, TestCodec)
|
||||||
|
|
||||||
await conn.writeLp("Hello!")
|
await conn.writeLp("Hello!")
|
||||||
let msg = cast[string](await conn.readLp(1024))
|
let msg = string.fromBytes(await conn.readLp(1024))
|
||||||
check "Hello!" == msg
|
check "Hello!" == msg
|
||||||
|
await conn.close()
|
||||||
|
|
||||||
|
await sleepAsync(2.seconds) # wait a little for cleanup to happen
|
||||||
|
var bufferTracker = getTracker(BufferStreamTrackerName)
|
||||||
|
# echo bufferTracker.dump()
|
||||||
|
|
||||||
|
# plus 4 for the pubsub streams
|
||||||
|
check (BufferStreamTracker(bufferTracker).opened ==
|
||||||
|
(BufferStreamTracker(bufferTracker).closed + 4.uint64))
|
||||||
|
|
||||||
|
var connTracker = getTracker(ConnectionTrackerName)
|
||||||
|
# echo connTracker.dump()
|
||||||
|
|
||||||
|
# plus 8 is for the secured connection and the socket
|
||||||
|
# and the pubsub streams that won't clean up until
|
||||||
|
# `disconnect()` or `stop()`
|
||||||
|
check (ConnectionTracker(connTracker).opened ==
|
||||||
|
(ConnectionTracker(connTracker).closed + 8.uint64))
|
||||||
|
|
||||||
await all(
|
await all(
|
||||||
done.wait(5.seconds) #[if OK won't happen!!]#,
|
done.wait(5.seconds), #[if OK won't happen!!]#
|
||||||
conn.close(),
|
|
||||||
switch1.stop(),
|
switch1.stop(),
|
||||||
switch2.stop(),
|
switch2.stop(),
|
||||||
)
|
)
|
||||||
|
@ -102,37 +95,37 @@ suite "Switch":
|
||||||
|
|
||||||
waitFor(testSwitch())
|
waitFor(testSwitch())
|
||||||
|
|
||||||
test "e2e use switch no proto string":
|
test "e2e use connect then dial":
|
||||||
proc testSwitch(): Future[bool] {.async, gcsafe.} =
|
proc testSwitch(): Future[bool] {.async, gcsafe.} =
|
||||||
let ma1: MultiAddress = Multiaddress.init("/ip4/0.0.0.0/tcp/0")
|
|
||||||
let ma2: MultiAddress = Multiaddress.init("/ip4/0.0.0.0/tcp/0")
|
|
||||||
|
|
||||||
var peerInfo1, peerInfo2: PeerInfo
|
|
||||||
var switch1, switch2: Switch
|
|
||||||
var awaiters: seq[Future[void]]
|
var awaiters: seq[Future[void]]
|
||||||
|
|
||||||
(switch1, peerInfo1) = createSwitch(ma1)
|
|
||||||
|
|
||||||
proc handle(conn: Connection, proto: string) {.async, gcsafe.} =
|
proc handle(conn: Connection, proto: string) {.async, gcsafe.} =
|
||||||
let msg = cast[string](await conn.readLp(1024))
|
try:
|
||||||
check "Hello!" == msg
|
let msg = string.fromBytes(await conn.readLp(1024))
|
||||||
await conn.writeLp("Hello!")
|
check "Hello!" == msg
|
||||||
await conn.close()
|
finally:
|
||||||
|
await conn.writeLp("Hello!")
|
||||||
|
await conn.close()
|
||||||
|
|
||||||
let testProto = new TestProto
|
let testProto = new TestProto
|
||||||
testProto.codec = TestCodec
|
testProto.codec = TestCodec
|
||||||
testProto.handler = handle
|
testProto.handler = handle
|
||||||
|
|
||||||
|
let switch1 = newStandardSwitch()
|
||||||
switch1.mount(testProto)
|
switch1.mount(testProto)
|
||||||
|
|
||||||
(switch2, peerInfo2) = createSwitch(ma2)
|
let switch2 = newStandardSwitch()
|
||||||
awaiters.add(await switch1.start())
|
awaiters.add(await switch1.start())
|
||||||
awaiters.add(await switch2.start())
|
awaiters.add(await switch2.start())
|
||||||
|
|
||||||
await switch2.connect(switch1.peerInfo)
|
await switch2.connect(switch1.peerInfo)
|
||||||
|
check switch1.peerInfo.id in switch2.connections
|
||||||
|
|
||||||
let conn = await switch2.dial(switch1.peerInfo, TestCodec)
|
let conn = await switch2.dial(switch1.peerInfo, TestCodec)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
await conn.writeLp("Hello!")
|
await conn.writeLp("Hello!")
|
||||||
let msg = cast[string](await conn.readLp(1024))
|
let msg = string.fromBytes(await conn.readLp(1024))
|
||||||
check "Hello!" == msg
|
check "Hello!" == msg
|
||||||
result = true
|
result = true
|
||||||
except LPStreamError:
|
except LPStreamError:
|
||||||
|
@ -148,6 +141,36 @@ suite "Switch":
|
||||||
check:
|
check:
|
||||||
waitFor(testSwitch()) == true
|
waitFor(testSwitch()) == true
|
||||||
|
|
||||||
|
test "e2e should not leak on peer disconnect":
|
||||||
|
proc testSwitch() {.async, gcsafe.} =
|
||||||
|
var awaiters: seq[Future[void]]
|
||||||
|
|
||||||
|
let switch1 = newStandardSwitch()
|
||||||
|
let switch2 = newStandardSwitch()
|
||||||
|
awaiters.add(await switch1.start())
|
||||||
|
awaiters.add(await switch2.start())
|
||||||
|
await switch2.connect(switch1.peerInfo)
|
||||||
|
|
||||||
|
await sleepAsync(100.millis)
|
||||||
|
await switch2.disconnect(switch1.peerInfo)
|
||||||
|
|
||||||
|
await sleepAsync(2.seconds)
|
||||||
|
var bufferTracker = getTracker(BufferStreamTrackerName)
|
||||||
|
# echo bufferTracker.dump()
|
||||||
|
check bufferTracker.isLeaked() == false
|
||||||
|
|
||||||
|
var connTracker = getTracker(ConnectionTrackerName)
|
||||||
|
# echo connTracker.dump()
|
||||||
|
check connTracker.isLeaked() == false
|
||||||
|
|
||||||
|
await all(
|
||||||
|
switch1.stop(),
|
||||||
|
switch2.stop()
|
||||||
|
)
|
||||||
|
await all(awaiters)
|
||||||
|
|
||||||
|
waitFor(testSwitch())
|
||||||
|
|
||||||
# test "e2e: handle read + secio fragmented":
|
# test "e2e: handle read + secio fragmented":
|
||||||
# proc testListenerDialer(): Future[bool] {.async.} =
|
# proc testListenerDialer(): Future[bool] {.async.} =
|
||||||
# let
|
# let
|
||||||
|
|
Loading…
Reference in New Issue