wip: mplex tests
This commit is contained in:
Dmitriy Ryajov
parent
b26d1ac23a
commit
96cd7bcf50
145
libp2p/mplex.nim
145
libp2p/mplex.nim
|
|
@ -1,145 +0,0 @@
|
||||||
## Nim-LibP2P
|
|
||||||
## Copyright (c) 2018 Status Research & Development GmbH
|
|
||||||
## Licensed under either of
|
|
||||||
## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
|
|
||||||
## * MIT license ([LICENSE-MIT](LICENSE-MIT))
|
|
||||||
## at your option.
|
|
||||||
## This file may not be copied, modified, or distributed except according to
|
|
||||||
## those terms.
|
|
||||||
|
|
||||||
import chronos
|
|
||||||
import ../varint, ../connection,
|
|
||||||
../vbuffer, ../protocol
|
|
||||||
|
|
||||||
const DefaultReadSize: uint = 1024
|
|
||||||
|
|
||||||
const MaxMsgSize* = 1 shl 20 # 1mb
|
|
||||||
const MaxChannels* = 1000
|
|
||||||
|
|
||||||
type
|
|
||||||
MplexUnknownMsgError* = object of CatchableError
|
|
||||||
|
|
||||||
MessageType* {.pure.} = enum
|
|
||||||
New,
|
|
||||||
MsgIn,
|
|
||||||
MsgOut,
|
|
||||||
CloseIn,
|
|
||||||
CloseOut,
|
|
||||||
ResetIn,
|
|
||||||
ResetOut
|
|
||||||
|
|
||||||
ChannelHandler* = proc(conn: Connection) {.gcsafe.}
|
|
||||||
|
|
||||||
Mplex* = ref object of LPProtocol
|
|
||||||
remote*: seq[Connection]
|
|
||||||
local*: seq[Connection]
|
|
||||||
channelHandler*: ChannelHandler
|
|
||||||
currentId*: uint
|
|
||||||
|
|
||||||
Channel* = ref object of BufferStream
|
|
||||||
id*: int
|
|
||||||
initiator*: bool
|
|
||||||
reset*: bool
|
|
||||||
closedLocal*: bool
|
|
||||||
closedRemote*: bool
|
|
||||||
mplex*: Mplex
|
|
||||||
|
|
||||||
proc newMplexUnknownMsgError*(): ref MplexUnknownMsgError =
|
|
||||||
result = newException(MplexUnknownMsgError, "Unknown mplex message type")
|
|
||||||
|
|
||||||
proc readLp*(conn: Connection): Future[tuple[id: uint, msgType: MessageType]] {.gcsafe.} =
|
|
||||||
var
|
|
||||||
header: uint
|
|
||||||
length: int
|
|
||||||
res: VarintStatus
|
|
||||||
var buffer = newSeq[byte](10)
|
|
||||||
try:
|
|
||||||
for i in 0..<len(buffer):
|
|
||||||
await conn.readExactly(addr buffer[i], 1)
|
|
||||||
res = LP.getUVarint(buffer.toOpenArray(0, i), length, header)
|
|
||||||
if res == VarintStatus.Success:
|
|
||||||
break
|
|
||||||
if res != VarintStatus.Success:
|
|
||||||
buffer.setLen(0)
|
|
||||||
except TransportIncompleteError:
|
|
||||||
buffer.setLen(0)
|
|
||||||
|
|
||||||
result.id = header shl 3
|
|
||||||
result.msgType = MessageType(header and 0x7)
|
|
||||||
|
|
||||||
proc writeLp*(conn: Connection, id: uint, msgType: MessageType) {.async, gcsafe.} =
|
|
||||||
## write lenght prefixed
|
|
||||||
var buf = initVBuffer()
|
|
||||||
buf.writeVarint(LPSomeUVarint(id shl 3 or uint(msgType)))
|
|
||||||
buf.finish()
|
|
||||||
result = conn.write(buf.buffer)
|
|
||||||
|
|
||||||
proc newChannel*(mplex: Mplex,
|
|
||||||
id: int,
|
|
||||||
initiator: bool,
|
|
||||||
handler: WriteHandler,
|
|
||||||
size: int = MaxMsgSize): Channel =
|
|
||||||
new result
|
|
||||||
result.id = id
|
|
||||||
result.mplex = mplex
|
|
||||||
result.initiator = initiator
|
|
||||||
result.writeHandler = handler
|
|
||||||
result.maxSize = size
|
|
||||||
|
|
||||||
proc closed*(s: Channel): bool = s.closedLocal and s.closedRemote
|
|
||||||
proc closeRemote*(s: Channel) = discard
|
|
||||||
proc close*(s: Channel) = discard
|
|
||||||
|
|
||||||
proc newStream*(m: Mplex,
|
|
||||||
conn: Connection,
|
|
||||||
initiator: bool = true):
|
|
||||||
Connection {.gcsafe.}
|
|
||||||
|
|
||||||
proc newMplex*(conn: Connection,
|
|
||||||
handler: ChannelHandler,
|
|
||||||
maxChanns: uint = MaxChannels): Mplex =
|
|
||||||
new result
|
|
||||||
result.channelHandler = handler
|
|
||||||
|
|
||||||
proc processNewStream(m: Mplex, conn: Connection) {.async, gcsafe.} =
|
|
||||||
discard
|
|
||||||
|
|
||||||
proc procesMessage(m: Mplex, conn: Connection, initiator: bool) {.async, gcsafe.} =
|
|
||||||
discard
|
|
||||||
|
|
||||||
proc processClose(m: Mplex, conn: Connection, initiator: bool) {.async, gcsafe.} =
|
|
||||||
discard
|
|
||||||
|
|
||||||
proc processReset(m: Mplex, conn: Connection, initiator: bool) {.async, gcsafe.} =
|
|
||||||
discard
|
|
||||||
|
|
||||||
proc newStream*(m: Mplex,
|
|
||||||
conn: Connection,
|
|
||||||
initiator: bool = true):
|
|
||||||
Connection {.gcsafe.} =
|
|
||||||
## create new channel/stream
|
|
||||||
let id = m.currentId
|
|
||||||
inc(m.currentId)
|
|
||||||
proc writeHandler(data: seq[byte]): Future[void] {.gcsafe.} =
|
|
||||||
let msgType = if initiator: MessageType.MsgIn else: MessageType.MsgOut
|
|
||||||
await conn.writeLp(id, msgType) # write header
|
|
||||||
await conn.writeLp(data) # write data
|
|
||||||
|
|
||||||
let channel = newChannel(m, id, initiator, writeHandler)
|
|
||||||
result = newConnection(channel)
|
|
||||||
|
|
||||||
method init(m: Mplex) =
|
|
||||||
proc handle(conn: Connection, proto: string) {.async, closure, gcsafe.} =
|
|
||||||
let (id, msgType) = await conn.readLp()
|
|
||||||
case msgType:
|
|
||||||
of MessageType.New:
|
|
||||||
await m.processNewStream(conn)
|
|
||||||
of MessageType.MsgIn, MessageType.MsgOut:
|
|
||||||
await m.procesMessage(conn, bool(ord(msgType) and 1))
|
|
||||||
of MessageType.CloseIn, MessageType.CloseOut:
|
|
||||||
await m.processClose(conn, bool(ord(msgType) and 1))
|
|
||||||
of MessageType.ResetIn, MessageType.ResetOut:
|
|
||||||
await m.processReset(conn, bool(ord(msgType) and 1))
|
|
||||||
else: raise newMplexUnknownMsgError()
|
|
||||||
|
|
||||||
m.handler = handle
|
|
||||||
|
|
@ -0,0 +1,162 @@
|
||||||
|
## Nim-LibP2P
|
||||||
|
## Copyright (c) 2018 Status Research & Development GmbH
|
||||||
|
## Licensed under either of
|
||||||
|
## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
|
||||||
|
## * MIT license ([LICENSE-MIT](LICENSE-MIT))
|
||||||
|
## at your option.
|
||||||
|
## This file may not be copied, modified, or distributed except according to
|
||||||
|
## those terms.
|
||||||
|
|
||||||
|
import sequtils
|
||||||
|
import chronos
|
||||||
|
import ../varint, ../connection,
|
||||||
|
../vbuffer, ../protocol,
|
||||||
|
../stream/bufferstream,
|
||||||
|
muxer
|
||||||
|
|
||||||
|
const MaxMsgSize* = 1 shl 20 # 1mb
|
||||||
|
const MaxChannels* = 1000
|
||||||
|
const MplexCodec* = "/mplex/6.7.0"
|
||||||
|
|
||||||
|
type
|
||||||
|
MplexUnknownMsgError* = object of CatchableError
|
||||||
|
MessageType* {.pure.} = enum
|
||||||
|
New,
|
||||||
|
MsgIn,
|
||||||
|
MsgOut,
|
||||||
|
CloseIn,
|
||||||
|
CloseOut,
|
||||||
|
ResetIn,
|
||||||
|
ResetOut
|
||||||
|
|
||||||
|
StreamHandler = proc(conn: Connection): Future[void] {.gcsafe.}
|
||||||
|
Mplex* = ref object of Muxer
|
||||||
|
remote*: seq[Channel]
|
||||||
|
local*: seq[Channel]
|
||||||
|
currentId*: int
|
||||||
|
maxChannels*: uint
|
||||||
|
streamHandler*: StreamHandler
|
||||||
|
|
||||||
|
Channel* = ref object of BufferStream
|
||||||
|
id*: int
|
||||||
|
initiator*: bool
|
||||||
|
isReset*: bool
|
||||||
|
closedLocal*: bool
|
||||||
|
closedRemote*: bool
|
||||||
|
mplex*: Mplex
|
||||||
|
|
||||||
|
proc newMplexUnknownMsgError*(): ref MplexUnknownMsgError =
|
||||||
|
result = newException(MplexUnknownMsgError, "Unknown mplex message type")
|
||||||
|
|
||||||
|
##########################################
|
||||||
|
## Read/Write Helpers
|
||||||
|
##########################################
|
||||||
|
|
||||||
|
proc readHeader*(conn: Connection): Future[(uint, MessageType)] {.async, gcsafe.} =
|
||||||
|
var
|
||||||
|
header: uint
|
||||||
|
length: int
|
||||||
|
res: VarintStatus
|
||||||
|
var buffer = newSeq[byte](10)
|
||||||
|
try:
|
||||||
|
for i in 0..<len(buffer):
|
||||||
|
await conn.readExactly(addr buffer[i], 1)
|
||||||
|
res = LP.getUVarint(buffer.toOpenArray(0, i), length, header)
|
||||||
|
if res == VarintStatus.Success:
|
||||||
|
return (header shr 3, MessageType(header and 0x7))
|
||||||
|
if res != VarintStatus.Success:
|
||||||
|
buffer.setLen(0)
|
||||||
|
return
|
||||||
|
except TransportIncompleteError:
|
||||||
|
buffer.setLen(0)
|
||||||
|
|
||||||
|
proc writeHeader*(conn: Connection,
|
||||||
|
id: int,
|
||||||
|
msgType: MessageType,
|
||||||
|
size: int) {.async, gcsafe.} =
|
||||||
|
## write lenght prefixed
|
||||||
|
var buf = initVBuffer()
|
||||||
|
buf.writeVarint(LPSomeUVarint(id.uint shl 3 or msgType.uint))
|
||||||
|
buf.writeVarint(LPSomeUVarint(size.uint))
|
||||||
|
buf.finish()
|
||||||
|
result = conn.write(buf.buffer)
|
||||||
|
|
||||||
|
##########################################
|
||||||
|
## Channel
|
||||||
|
##########################################
|
||||||
|
|
||||||
|
proc newChannel*(mplex: Mplex,
|
||||||
|
id: int,
|
||||||
|
initiator: bool,
|
||||||
|
handler: WriteHandler,
|
||||||
|
size: int = MaxMsgSize): Channel =
|
||||||
|
new result
|
||||||
|
result.id = id
|
||||||
|
result.mplex = mplex
|
||||||
|
result.initiator = initiator
|
||||||
|
result.writeHandler = handler
|
||||||
|
result.maxSize = size
|
||||||
|
|
||||||
|
proc closed*(s: Channel): bool = s.closedLocal and s.closedRemote
|
||||||
|
proc close*(s: Channel) {.async.} = discard
|
||||||
|
proc reset*(s: Channel) {.async.} = discard
|
||||||
|
|
||||||
|
##########################################
|
||||||
|
##
|
||||||
|
## Mplex
|
||||||
|
##
|
||||||
|
##########################################
|
||||||
|
|
||||||
|
proc getChannelList(m: Mplex, initiator: bool): var seq[Channel] =
|
||||||
|
if initiator:
|
||||||
|
result = m.remote
|
||||||
|
else:
|
||||||
|
result = m.local
|
||||||
|
|
||||||
|
proc newStream*(m: Mplex,
|
||||||
|
chanId: int = -1,
|
||||||
|
initiator: bool = true):
|
||||||
|
Future[Connection] {.async, gcsafe.} =
|
||||||
|
## create new channel/stream
|
||||||
|
defer: inc(m.currentId)
|
||||||
|
let id = if chanId > -1: chanId else: m.currentId
|
||||||
|
proc writeHandler(data: seq[byte]): Future[void] {.async, gcsafe.} =
|
||||||
|
let msgType = if initiator: MessageType.MsgIn else: MessageType.MsgOut
|
||||||
|
await m.connection.writeHeader(id, msgType, data.len) # write header
|
||||||
|
await m.connection.write(data) # write data
|
||||||
|
|
||||||
|
let channel = newChannel(m, id, initiator, writeHandler)
|
||||||
|
m.getChannelList(initiator)[id] = channel
|
||||||
|
result = newConnection(channel)
|
||||||
|
|
||||||
|
proc handle*(m: Mplex) {.async, gcsafe.} =
|
||||||
|
while not m.connection.closed:
|
||||||
|
let (id, msgType) = await m.connection.readHeader()
|
||||||
|
let initiator = bool(ord(msgType) and 1)
|
||||||
|
case msgType:
|
||||||
|
of MessageType.New:
|
||||||
|
await m.streamHandler(await m.newStream(id.int, false))
|
||||||
|
of MessageType.MsgIn, MessageType.MsgOut:
|
||||||
|
await m.getChannelList(initiator)[id.int].pushTo(await m.connection.readLp())
|
||||||
|
of MessageType.CloseIn, MessageType.CloseOut:
|
||||||
|
await m.getChannelList(initiator)[id.int].close()
|
||||||
|
of MessageType.ResetIn, MessageType.ResetOut:
|
||||||
|
await m.getChannelList(initiator)[id.int].reset()
|
||||||
|
else: raise newMplexUnknownMsgError()
|
||||||
|
|
||||||
|
proc newMplex*(conn: Connection,
|
||||||
|
streamHandler: StreamHandler,
|
||||||
|
maxChanns: uint = MaxChannels): Mplex =
|
||||||
|
new result
|
||||||
|
result.connection = conn
|
||||||
|
result.maxChannels = maxChanns
|
||||||
|
result.streamHandler = streamHandler
|
||||||
|
|
||||||
|
method newStream*(m: Mplex): Future[Connection] {.gcsafe.} =
|
||||||
|
result = m.newStream(true)
|
||||||
|
|
||||||
|
method close(m: Mplex) {.async, gcsafe.} =
|
||||||
|
let futs = @[allFutures(m.remote.mapIt(it.close())),
|
||||||
|
allFutures(m.local.mapIt(it.close()))]
|
||||||
|
await allFutures(futs)
|
||||||
|
await m.connection.close()
|
||||||
|
|
@ -0,0 +1,34 @@
|
||||||
|
## Nim-LibP2P
|
||||||
|
## Copyright (c) 2018 Status Research & Development GmbH
|
||||||
|
## Licensed under either of
|
||||||
|
## * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
|
||||||
|
## * MIT license ([LICENSE-MIT](LICENSE-MIT))
|
||||||
|
## at your option.
|
||||||
|
## This file may not be copied, modified, or distributed except according to
|
||||||
|
## those terms.
|
||||||
|
|
||||||
|
import chronos
|
||||||
|
import ../protocol, ../connection
|
||||||
|
|
||||||
|
type
|
||||||
|
Muxer* = ref object of RootObj
|
||||||
|
connection*: Connection
|
||||||
|
|
||||||
|
MuxerCreator* = proc(conn: Connection): Muxer {.gcsafe, closure.}
|
||||||
|
# this wraps a creator proc that knows how to make muxers
|
||||||
|
MuxerProvider* = ref object of LPProtocol
|
||||||
|
newMuxer*: MuxerCreator
|
||||||
|
|
||||||
|
proc newMuxerProvider*(creator: MuxerCreator, codec: string): MuxerProvider {.gcsafe.} =
|
||||||
|
new result
|
||||||
|
result.newMuxer = creator
|
||||||
|
result.codec = codec
|
||||||
|
|
||||||
|
method init(c: MuxerProvider) =
|
||||||
|
proc handler(conn: Connection, proto: string) {.async, gcsafe, closure.} =
|
||||||
|
let muxer = c.newMuxer(conn)
|
||||||
|
|
||||||
|
c.handler = handler
|
||||||
|
|
||||||
|
method newStream*(m: Muxer): Future[Connection] {.base, async, gcsafe.} = discard
|
||||||
|
method close*(m: Muxer) {.base, async, gcsafe.} = discard
|
||||||
|
|
@ -0,0 +1,90 @@
|
||||||
|
import unittest, sequtils, sugar
|
||||||
|
import chronos, nimcrypto/utils
|
||||||
|
import ../libp2p/muxers/mplex, ../libp2p/connection,
|
||||||
|
../libp2p/stream/lpstream, ../libp2p/tcptransport,
|
||||||
|
../libp2p/transport, ../libp2p/multiaddress
|
||||||
|
|
||||||
|
type
|
||||||
|
TestEncodeStream = ref object of LPStream
|
||||||
|
handler*: proc(data: seq[byte])
|
||||||
|
|
||||||
|
method write*(s: TestEncodeStream,
|
||||||
|
msg: seq[byte],
|
||||||
|
msglen = -1):
|
||||||
|
Future[void] {.gcsafe.} =
|
||||||
|
s.handler(msg)
|
||||||
|
|
||||||
|
proc newTestEncodeStream(handler: proc(data: seq[byte])): TestEncodeStream =
|
||||||
|
new result
|
||||||
|
result.handler = handler
|
||||||
|
|
||||||
|
type
|
||||||
|
TestDecodeStream = ref object of LPStream
|
||||||
|
handler*: proc(data: seq[byte])
|
||||||
|
step*: int
|
||||||
|
msg*: seq[byte]
|
||||||
|
|
||||||
|
method readExactly*(s: TestDecodeStream,
|
||||||
|
pbytes: pointer,
|
||||||
|
nbytes: int): Future[void] {.async, gcsafe.} =
|
||||||
|
let buff: seq[byte] = s.msg
|
||||||
|
copyMem(pbytes, unsafeAddr buff[s.step], nbytes)
|
||||||
|
s.step += nbytes
|
||||||
|
|
||||||
|
proc newTestDecodeStream(): TestDecodeStream =
|
||||||
|
new result
|
||||||
|
result.step = 0
|
||||||
|
result.msg = fromHex("8801023137")
|
||||||
|
|
||||||
|
suite "Mplex":
|
||||||
|
# test "encode header":
|
||||||
|
# proc testEncodeHeader(): Future[bool] {.async.} =
|
||||||
|
# proc encHandler(msg: seq[byte]) =
|
||||||
|
# check msg == fromHex("880102")
|
||||||
|
|
||||||
|
# let conn = newConnection(newTestEncodeStream(encHandler))
|
||||||
|
# await conn.writeHeader(uint(17), MessageType.New, 2)
|
||||||
|
# result = true
|
||||||
|
|
||||||
|
# check:
|
||||||
|
# waitFor(testEncodeHeader()) == true
|
||||||
|
|
||||||
|
# test "decode header":
|
||||||
|
# proc testDecodeHeader(): Future[bool] {.async.} =
|
||||||
|
# let conn = newConnection(newTestDecodeStream())
|
||||||
|
# let (id, msgType) = await conn.readHeader()
|
||||||
|
|
||||||
|
# check id == 17
|
||||||
|
# check msgType == MessageType.New
|
||||||
|
# let data = await conn.readLp()
|
||||||
|
# check cast[string](data) == "17"
|
||||||
|
# result = true
|
||||||
|
|
||||||
|
# check:
|
||||||
|
# waitFor(testDecodeHeader()) == true
|
||||||
|
|
||||||
|
test "e2e - new stream":
|
||||||
|
proc testNewStream(): Future[bool] {.async.} =
|
||||||
|
let ma: MultiAddress = Multiaddress.init("/ip4/127.0.0.1/tcp/53351")
|
||||||
|
|
||||||
|
proc connHandler(conn: Connection) {.async, gcsafe.} =
|
||||||
|
proc handleListen(stream: Connection) {.async, gcsafe.} =
|
||||||
|
await stream.writeLp("Hello from stream!")
|
||||||
|
|
||||||
|
let mplexListen = newMplex(conn, handleListen)
|
||||||
|
await mplexListen.handle()
|
||||||
|
|
||||||
|
let transport1: TcpTransport = newTransport(TcpTransport)
|
||||||
|
await transport1.listen(ma, connHandler)
|
||||||
|
|
||||||
|
let transport2: TcpTransport = newTransport(TcpTransport)
|
||||||
|
let conn = await transport2.dial(ma)
|
||||||
|
proc handleDial(stream: Connection) {.async, gcsafe.} =
|
||||||
|
let msg = await stream.readLp()
|
||||||
|
|
||||||
|
let mplexDial = newMplex(conn, handleDial)
|
||||||
|
let handleFut = mplexDial.handle()
|
||||||
|
result = true
|
||||||
|
|
||||||
|
check:
|
||||||
|
waitFor(testNewStream()) == true
|
||||||
Loading…
Reference in New Issue