mplex move everything to one file
This commit is contained in:
Dmitriy Ryajov
parent
ad1eaffdd6
commit
4edd2c9f8a
|
|
@ -0,0 +1,145 @@
|
|||
## 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
|
||||
|
|
@ -1,58 +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 ../connection, ../stream
|
||||
import mplex
|
||||
|
||||
type
|
||||
Channel = ref object of LPStream
|
||||
id*: int
|
||||
initiator*: bool
|
||||
reset*: bool
|
||||
closedLocal*: bool
|
||||
closedRemote*: bool
|
||||
buffer*: seq[byte]
|
||||
|
||||
proc newChannel*(mplex: Mplex, id: int, initiator: bool): Channel =
|
||||
new result
|
||||
result.id = id
|
||||
result.initiator = initiator
|
||||
|
||||
proc closed*(s: Channel): bool = s.closedLocal and s.closedRemote
|
||||
proc close*(s: Channel) = discard
|
||||
|
||||
method read*(s: Channel, n = -1): Future[seq[byte]] {.async, gcsafe.} =
|
||||
discard
|
||||
|
||||
method readExactly*(s: Channel, pbytes: pointer, nbytes: int): Future[void] {.async, gcsafe.} =
|
||||
discard
|
||||
|
||||
method readLine*(s: Channel, limit = 0, sep = "\r\n"): Future[string] {.async, gcsafe.} =
|
||||
discard
|
||||
|
||||
method readOnce*(s: Channel, pbytes: pointer, nbytes: int): Future[int] {.async, gcsafe.} =
|
||||
discard
|
||||
|
||||
method readUntil*(s: Channel,
|
||||
pbytes: pointer, nbytes: int,
|
||||
sep: seq[byte]): Future[int] {.async, gcsafe.} =
|
||||
discard
|
||||
|
||||
method write*(s: Channel, pbytes: pointer, nbytes: int) {.async, gcsafe.} =
|
||||
discard
|
||||
|
||||
method write*(s: Channel, msg: string, msglen = -1) {.async, gcsafe.} =
|
||||
discard
|
||||
|
||||
method write*(s: Channel, msg: seq[byte], msglen = -1) {.async, gcsafe.} =
|
||||
discard
|
||||
|
||||
method close*(s: Channel) {.async, gcsafe.} =
|
||||
discard
|
||||
|
|
@ -1,31 +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 tables
|
||||
import chronos
|
||||
import ../connection, ../stream
|
||||
import types
|
||||
import channel
|
||||
|
||||
type
|
||||
ChannelHandler* = proc(conn: Connection) {.gcsafe.}
|
||||
|
||||
Mplex* = ref object of RootObj
|
||||
connections*: seq[Connection]
|
||||
channels*: TableRef[uint, Connection]
|
||||
handler*: ChannelHandler
|
||||
|
||||
proc newMplex*(handler: ChannelHandler): Mplex =
|
||||
new result
|
||||
result.channels = newTable[uint, Connection]()
|
||||
result.handler = handler
|
||||
|
||||
proc newStream*(m: Mplex, conn: Connection): Connection {.gcsafe.} = discard
|
||||
proc handle*(m: Mplex, conn: Connection) {.gcsafe.} = discard
|
||||
proc send*(m: Mplex, msg: Message) {.async, gcsafe.} = discard
|
||||
|
|
@ -1,19 +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.
|
||||
|
||||
const MaxMsgSize* = 1 shl 20 # 1mb
|
||||
|
||||
type
|
||||
MessageType* = enum
|
||||
New, InMsg, OutMsg, InClose, OutClose, InReset, OutReset
|
||||
|
||||
Message* = ref object of RootObj
|
||||
id*: int
|
||||
msgaType*: MessageType
|
||||
data*: seq[byte]
|
||||
Loading…
Reference in New Issue