nim-libp2p/libp2p/muxers/mplex/coder.nim

76 lines
2.2 KiB
Nim
Raw Permalink Normal View History

2019-09-04 03:08:51 +00:00
## Nim-LibP2P
2019-09-24 17:48:23 +00:00
## Copyright (c) 2019 Status Research & Development GmbH
2019-09-04 03:08:51 +00:00
## 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
2020-06-03 02:21:11 +00:00
import nimcrypto/utils, chronicles, stew/byteutils
2019-09-09 17:33:32 +00:00
import types,
../../stream/connection,
2020-05-08 20:58:23 +00:00
../../utility,
2019-09-12 17:07:34 +00:00
../../varint,
../../vbuffer
2019-09-04 03:08:51 +00:00
2019-09-10 02:15:52 +00:00
logScope:
topics = "mplexcoder"
2019-09-10 02:15:52 +00:00
2019-09-06 21:27:55 +00:00
type
Msg* = tuple
2020-03-23 17:14:06 +00:00
id: uint64
2019-09-07 23:34:11 +00:00
msgType: MessageType
data: seq[byte]
2019-09-06 21:27:55 +00:00
InvalidMplexMsgType = object of CatchableError
proc newInvalidMplexMsgType*(): ref InvalidMplexMsgType =
newException(InvalidMplexMsgType, "invalid message type")
2020-02-12 14:37:22 +00:00
proc readMsg*(conn: Connection): Future[Msg] {.async, gcsafe.} =
2020-05-08 20:58:23 +00:00
let header = await conn.readVarint()
trace "read header varint", varint = header
2019-09-07 23:34:11 +00:00
2020-05-08 20:58:23 +00:00
let data = await conn.readLp(MaxMsgSize)
trace "read data", dataLen = data.len, data = shortLog(data)
2019-09-07 23:34:11 +00:00
let msgType = header and 0x7
if msgType.int > ord(MessageType.ResetOut):
raise newInvalidMplexMsgType()
2020-05-08 20:58:23 +00:00
result = (header shr 3, MessageType(msgType), data)
proc writeMsg*(conn: Connection,
2020-03-23 17:14:06 +00:00
id: uint64,
msgType: MessageType,
2019-09-07 23:34:11 +00:00
data: seq[byte] = @[]) {.async, gcsafe.} =
trace "sending data over mplex", oid = $conn.oid,
id,
msgType,
data = data.len
2020-05-23 16:50:05 +00:00
var
left = data.len
offset = 0
while left > 0 or data.len == 0:
let
chunkSize = if left > MaxMsgSize: MaxMsgSize - 64 else: left
2020-05-29 16:25:25 +00:00
## write length prefixed
2020-05-23 16:50:05 +00:00
var buf = initVBuffer()
buf.writePBVarint(id shl 3 or ord(msgType).uint64)
buf.writeSeq(data.toOpenArray(offset, offset + chunkSize - 1))
2020-05-23 16:50:05 +00:00
buf.finish()
left = left - chunkSize
offset = offset + chunkSize
await conn.write(buf.buffer)
2020-05-23 16:50:05 +00:00
if data.len == 0:
return
2019-09-07 23:34:11 +00:00
proc writeMsg*(conn: Connection,
2020-03-23 17:14:06 +00:00
id: uint64,
msgType: MessageType,
data: string): Future[void] =
conn.writeMsg(id, msgType, data.toBytes())