reworked protocol
This commit is contained in:
parent
22dd8c0f6b
commit
022a248ca3
|
@ -86,6 +86,13 @@ proc decodeMsg*(buf: seq[byte]): IdentifyInfo =
|
|||
discard pb.getString(6, agentVersion)
|
||||
result.agentVersion = agentVersion
|
||||
|
||||
method init*(p: Identify) =
|
||||
proc handle(conn: Connection, proto: string) {.async, gcsafe.} =
|
||||
var pb = encodeMsg(p.peerInfo, await conn.getObservedAddrs())
|
||||
await conn.writeLp(pb.buffer)
|
||||
|
||||
p.handler = handle
|
||||
|
||||
proc identify*(p: Identify, conn: Connection): Future[IdentifyInfo] {.async.} =
|
||||
var message = await conn.readLp()
|
||||
if len(message) == 0:
|
||||
|
@ -97,7 +104,3 @@ proc push*(p: Identify, conn: Connection) {.async.} =
|
|||
var pb = encodeMsg(p.peerInfo, await conn.getObservedAddrs())
|
||||
let length = pb.getLen()
|
||||
await conn.writeLp(pb.buffer)
|
||||
|
||||
proc handle*(p: Identify, conn: Connection, proto: string) {.async, gcsafe.} =
|
||||
var pb = encodeMsg(p.peerInfo, await conn.getObservedAddrs())
|
||||
await conn.writeLp(pb.buffer)
|
||||
|
|
|
@ -24,7 +24,6 @@ type
|
|||
HandlerHolder* = object
|
||||
proto: string
|
||||
protocol: LPProtocol
|
||||
handler: LPProtoHandler
|
||||
match: Matcher
|
||||
|
||||
MultisteamSelect* = ref object of RootObj
|
||||
|
@ -102,17 +101,15 @@ proc handle*(m: MultisteamSelect, conn: Connection) {.async, gcsafe.} =
|
|||
for h in m.handlers:
|
||||
if (not isNil(h.match) and h.match(ms)) or ms == h.proto:
|
||||
await conn.writeLp(h.proto & "\n")
|
||||
await h.handler(h.protocol, conn, ms)
|
||||
await h.protocol.handler(conn, ms)
|
||||
return
|
||||
await conn.write(m.na)
|
||||
|
||||
proc addHandler*[T: LPProtocol](m: MultisteamSelect,
|
||||
proto: string,
|
||||
protocol: T,
|
||||
handler: LPProtoHandler,
|
||||
matcher: Matcher = nil) =
|
||||
## register a handler for the protocol
|
||||
m.handlers.add(HandlerHolder(proto: proto,
|
||||
handler: handler,
|
||||
protocol: protocol,
|
||||
match: matcher))
|
||||
|
|
|
@ -12,12 +12,14 @@ import connection, transport, stream,
|
|||
peerinfo, multiaddress
|
||||
|
||||
type
|
||||
LPProtoHandler* = proc (protocol: LPProtocol,
|
||||
conn: Connection,
|
||||
proto: string): Future[void] {.gcsafe.}
|
||||
LPProtoHandler* = proc (conn: Connection,
|
||||
proto: string):
|
||||
Future[void] {.gcsafe.}
|
||||
|
||||
LPProtocol* = ref object of RootObj
|
||||
peerInfo*: PeerInfo
|
||||
codec*: string
|
||||
handler*: LPProtoHandler
|
||||
|
||||
proc newProtocol*(p: typedesc[LPProtocol],
|
||||
peerInfo: PeerInfo): p =
|
||||
|
|
|
@ -8,10 +8,9 @@ import ../libp2p/identify, ../libp2p/multiaddress,
|
|||
../libp2p/crypto/crypto
|
||||
|
||||
suite "Identify":
|
||||
test "handle identify message6":
|
||||
test "handle identify message":
|
||||
proc testHandle(): Future[bool] {.async.} =
|
||||
let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53350")
|
||||
|
||||
let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53360")
|
||||
|
||||
let remoteSeckey = PrivateKey.random(RSA)
|
||||
proc receiver() {.async.} =
|
||||
|
@ -19,14 +18,10 @@ suite "Identify":
|
|||
peerInfo.peerId = PeerID.init(remoteSeckey)
|
||||
peerInfo.addrs.add(ma)
|
||||
|
||||
let identify = newProtocol(Identify, peerInfo)
|
||||
let identifyProto = newProtocol(Identify, peerInfo)
|
||||
let msListen = newMultistream()
|
||||
|
||||
proc handle(p: LPProtocol, conn: Connection, proto: string) {.async, gcsafe.} =
|
||||
await identify.handle(conn, proto)
|
||||
|
||||
msListen.addHandler(IdentifyCodec, identify, handle)
|
||||
|
||||
msListen.addHandler(IdentifyCodec, identifyProto)
|
||||
proc connHandler(conn: Connection): Future[void] {.async, gcsafe.} =
|
||||
await msListen.handle(conn)
|
||||
|
||||
|
@ -43,10 +38,10 @@ suite "Identify":
|
|||
peerInfo.peerId = PeerID.init(seckey)
|
||||
peerInfo.addrs.add(ma)
|
||||
|
||||
let identify = newProtocol(Identify, peerInfo)
|
||||
let identifyProto = newProtocol(Identify, peerInfo)
|
||||
let res = await msDial.select(conn, IdentifyCodec)
|
||||
|
||||
let id = await identify.identify(conn)
|
||||
let id = await identifyProto.identify(conn)
|
||||
await conn.close()
|
||||
|
||||
check id.pubKey == remoteSeckey.getKey()
|
||||
|
|
|
@ -1,8 +1,11 @@
|
|||
import unittest, strutils, sequtils, sugar
|
||||
import chronos
|
||||
import ../libp2p/connection, ../libp2p/multistream,
|
||||
../libp2p/stream, ../libp2p/connection, ../libp2p/multiaddress,
|
||||
../libp2p/transport, ../libp2p/tcptransport, ../libp2p/protocol
|
||||
../libp2p/stream, ../libp2p/connection,
|
||||
../libp2p/multiaddress, ../libp2p/transport,
|
||||
../libp2p/tcptransport, ../libp2p/protocol,
|
||||
../libp2p/crypto/crypto, ../libp2p/peerinfo,
|
||||
../libp2p/peer
|
||||
|
||||
## Mock stream for select test
|
||||
type
|
||||
|
@ -140,13 +143,17 @@ suite "Multistream select":
|
|||
let ms = newMultistream()
|
||||
let conn = newConnection(newTestSelectStream())
|
||||
|
||||
var protocol: LPProtocol
|
||||
proc testHandler(protocol: LPProtocol,
|
||||
conn: Connection,
|
||||
proto: string): Future[void] {.async, gcsafe.} =
|
||||
let seckey = PrivateKey.random(RSA)
|
||||
var peerInfo: PeerInfo
|
||||
peerInfo.peerId = PeerID.init(seckey)
|
||||
var protocol: LPProtocol = newProtocol(LPProtocol, peerInfo)
|
||||
proc testHandler(conn: Connection,
|
||||
proto: string):
|
||||
Future[void] {.async, gcsafe.} =
|
||||
check proto == "/test/proto/1.0.0"
|
||||
|
||||
ms.addHandler("/test/proto/1.0.0", protocol, testHandler)
|
||||
protocol.handler = testHandler
|
||||
ms.addHandler("/test/proto/1.0.0", protocol)
|
||||
await ms.handle(conn)
|
||||
result = true
|
||||
|
||||
|
@ -165,12 +172,16 @@ suite "Multistream select":
|
|||
check strProto == "\x26/test/proto1/1.0.0\n/test/proto2/1.0.0\n"
|
||||
await conn.close()
|
||||
|
||||
var protocol: LPProtocol
|
||||
proc testHandler(protocol: LPProtocol,
|
||||
conn: Connection,
|
||||
let seckey = PrivateKey.random(RSA)
|
||||
var peerInfo: PeerInfo
|
||||
peerInfo.peerId = PeerID.init(seckey)
|
||||
var protocol: LPProtocol = newProtocol(LPProtocol, peerInfo)
|
||||
proc testHandler(conn: Connection,
|
||||
proto: string): Future[void] {.async, gcsafe.} = discard
|
||||
ms.addHandler("/test/proto1/1.0.0", protocol, testHandler)
|
||||
ms.addHandler("/test/proto2/1.0.0", protocol, testHandler)
|
||||
|
||||
protocol.handler = testHandler
|
||||
ms.addHandler("/test/proto1/1.0.0", protocol)
|
||||
ms.addHandler("/test/proto2/1.0.0", protocol)
|
||||
await ms.handle(conn)
|
||||
result = true
|
||||
|
||||
|
@ -188,11 +199,14 @@ suite "Multistream select":
|
|||
check cast[string](msg) == "\x3na\n"
|
||||
await conn.close()
|
||||
|
||||
var protocol: LPProtocol
|
||||
proc testHandler(protocol: LPProtocol,
|
||||
conn: Connection,
|
||||
let seckey = PrivateKey.random(RSA)
|
||||
var peerInfo: PeerInfo
|
||||
peerInfo.peerId = PeerID.init(seckey)
|
||||
var protocol: LPProtocol = newProtocol(LPProtocol, peerInfo)
|
||||
proc testHandler(conn: Connection,
|
||||
proto: string): Future[void] {.async, gcsafe.} = discard
|
||||
ms.addHandler("/unabvailable/proto/1.0.0", protocol, testHandler)
|
||||
protocol.handler = testHandler
|
||||
ms.addHandler("/unabvailable/proto/1.0.0", protocol)
|
||||
|
||||
await ms.handle(conn)
|
||||
result = true
|
||||
|
@ -203,16 +217,20 @@ suite "Multistream select":
|
|||
test "e2e - handle":
|
||||
proc endToEnd(): Future[bool] {.async.} =
|
||||
let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53350")
|
||||
var protocol: LPProtocol
|
||||
proc testHandler(protocol: LPProtocol,
|
||||
conn: Connection,
|
||||
|
||||
let seckey = PrivateKey.random(RSA)
|
||||
var peerInfo: PeerInfo
|
||||
peerInfo.peerId = PeerID.init(seckey)
|
||||
var protocol: LPProtocol = newProtocol(LPProtocol, peerInfo)
|
||||
proc testHandler(conn: Connection,
|
||||
proto: string): Future[void] {.async, gcsafe.} =
|
||||
check proto == "/test/proto/1.0.0"
|
||||
await conn.writeLp("Hello!")
|
||||
await conn.close()
|
||||
|
||||
protocol.handler = testHandler
|
||||
let msListen = newMultistream()
|
||||
msListen.addHandler("/test/proto/1.0.0", protocol, testHandler)
|
||||
msListen.addHandler("/test/proto/1.0.0", protocol)
|
||||
|
||||
proc connHandler(conn: Connection): Future[void] {.async, gcsafe.} =
|
||||
await msListen.handle(conn)
|
||||
|
@ -239,12 +257,15 @@ suite "Multistream select":
|
|||
let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53351")
|
||||
|
||||
let msListen = newMultistream()
|
||||
var protocol: LPProtocol
|
||||
proc testHandler(protocol: LPProtocol,
|
||||
conn: Connection,
|
||||
let seckey = PrivateKey.random(RSA)
|
||||
var peerInfo: PeerInfo
|
||||
peerInfo.peerId = PeerID.init(seckey)
|
||||
var protocol: LPProtocol = newProtocol(LPProtocol, peerInfo)
|
||||
proc testHandler(conn: Connection,
|
||||
proto: string): Future[void] {.async.} = discard
|
||||
msListen.addHandler("/test/proto1/1.0.0", protocol, testHandler)
|
||||
msListen.addHandler("/test/proto2/1.0.0", protocol, testHandler)
|
||||
protocol.handler = testHandler
|
||||
msListen.addHandler("/test/proto1/1.0.0", protocol)
|
||||
msListen.addHandler("/test/proto2/1.0.0", protocol)
|
||||
|
||||
let transport1: TcpTransport = newTransport(TcpTransport)
|
||||
proc connHandler(conn: Connection): Future[void] {.async, gcsafe.} =
|
||||
|
|
Loading…
Reference in New Issue