nim-libp2p-experimental/libp2p/muxers/mplex/channel.nim

145 lines
4.5 KiB
Nim
Raw Normal View History

2019-09-04 03:08:51 +00:00
## 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.
2019-09-06 06:51:19 +00:00
import chronos, strformat
2019-09-04 06:40:11 +00:00
import ../../stream/bufferstream,
../../stream/lpstream,
2019-09-06 06:51:19 +00:00
../../connection,
nimcrypto/utils,
types, coder
2019-09-04 03:08:51 +00:00
2019-09-06 21:27:55 +00:00
const DefaultChannelSize* = DefaultBufferSize * 64 # 64kb
2019-09-04 03:08:51 +00:00
type
Channel* = ref object of BufferStream
id*: uint
2019-09-06 21:27:55 +00:00
name*: string
2019-09-04 06:40:11 +00:00
conn*: Connection
2019-09-04 03:08:51 +00:00
initiator*: bool
isReset*: bool
closedLocal*: bool
closedRemote*: bool
handlerFuture*: Future[void]
2019-09-04 06:40:11 +00:00
msgCode*: MessageType
closeCode*: MessageType
resetCode*: MessageType
2019-09-07 23:32:32 +00:00
asyncLock: AsyncLock
2019-09-04 03:08:51 +00:00
proc newChannel*(id: uint,
2019-09-04 06:40:11 +00:00
conn: Connection,
2019-09-04 03:08:51 +00:00
initiator: bool,
2019-09-06 21:27:55 +00:00
name: string = "",
size: int = DefaultChannelSize): Channel =
2019-09-04 03:08:51 +00:00
new result
result.id = id
2019-09-06 21:27:55 +00:00
result.name = name
2019-09-04 06:40:11 +00:00
result.conn = conn
2019-09-04 03:08:51 +00:00
result.initiator = initiator
2019-09-04 06:40:11 +00:00
result.msgCode = if initiator: MessageType.MsgOut else: MessageType.MsgIn
result.closeCode = if initiator: MessageType.CloseOut else: MessageType.CloseIn
result.resetCode = if initiator: MessageType.ResetOut else: MessageType.ResetIn
2019-09-07 23:32:32 +00:00
result.asyncLock = newAsyncLock()
2019-09-04 03:08:51 +00:00
2019-09-04 06:40:11 +00:00
let chan = result
proc writeHandler(data: seq[byte]): Future[void] {.async, gcsafe.} =
2019-09-07 23:32:32 +00:00
# writes should happen in sequence
await chan.asyncLock.acquire()
await conn.writeMsg(chan.id, chan.msgCode, data) # write header
2019-09-07 23:32:32 +00:00
chan.asyncLock.release()
2019-09-04 06:40:11 +00:00
result.initBufferStream(writeHandler, size)
proc closeMessage(s: Channel) {.async, gcsafe.} =
await s.conn.writeMsg(s.id, s.closeCode) # write header
2019-09-04 06:40:11 +00:00
proc closed*(s: Channel): bool =
2019-09-07 23:32:32 +00:00
s.closedLocal and s.closedLocal
2019-09-04 06:40:11 +00:00
2019-09-04 16:40:05 +00:00
proc closedByRemote*(s: Channel) {.async.} =
2019-09-04 06:40:11 +00:00
s.closedRemote = true
2019-09-07 23:32:32 +00:00
proc cleanUp*(s: Channel): Future[void] =
result = procCall close(BufferStream(s))
2019-09-04 06:40:11 +00:00
method close*(s: Channel) {.async, gcsafe.} =
s.closedLocal = true
await s.closeMessage()
proc resetMessage(s: Channel) {.async, gcsafe.} =
await s.conn.writeMsg(s.id, s.resetCode)
2019-09-04 06:40:11 +00:00
2019-09-04 16:40:05 +00:00
proc resetByRemote*(s: Channel) {.async, gcsafe.} =
await allFutures(s.close(), s.closedByRemote())
2019-09-04 06:40:11 +00:00
s.isReset = true
proc reset*(s: Channel) {.async.} =
2019-09-04 16:40:05 +00:00
await allFutures(s.resetMessage(), s.resetByRemote())
2019-09-04 06:40:11 +00:00
proc isReadEof(s: Channel): bool =
2019-09-04 06:51:16 +00:00
bool((s.closedRemote or s.closedLocal) and s.len() < 1)
2019-09-04 06:40:11 +00:00
2019-09-04 06:51:16 +00:00
proc pushTo*(s: Channel, data: seq[byte]): Future[void] {.gcsafe.} =
2019-09-04 06:40:11 +00:00
if s.closedRemote:
raise newLPStreamClosedError()
2019-09-04 06:51:16 +00:00
result = procCall pushTo(BufferStream(s), data)
2019-09-04 06:40:11 +00:00
method read*(s: Channel, n = -1): Future[seq[byte]] {.gcsafe.} =
if s.isReadEof():
raise newLPStreamClosedError()
result = procCall read(BufferStream(s), n)
method readExactly*(s: Channel,
pbytes: pointer,
nbytes: int):
Future[void] {.gcsafe.} =
if s.isReadEof():
raise newLPStreamClosedError()
result = procCall readExactly(BufferStream(s), pbytes, nbytes)
method readLine*(s: Channel,
limit = 0,
sep = "\r\n"):
Future[string] {.gcsafe.} =
if s.isReadEof():
raise newLPStreamClosedError()
result = procCall readLine(BufferStream(s), limit, sep)
method readOnce*(s: Channel,
pbytes: pointer,
nbytes: int):
Future[int] {.gcsafe.} =
if s.isReadEof():
raise newLPStreamClosedError()
result = procCall readOnce(BufferStream(s), pbytes, nbytes)
method readUntil*(s: Channel,
pbytes: pointer, nbytes: int,
sep: seq[byte]):
Future[int] {.gcsafe.} =
if s.isReadEof():
raise newLPStreamClosedError()
result = procCall readOnce(BufferStream(s), pbytes, nbytes)
method write*(s: Channel,
pbytes: pointer,
nbytes: int): Future[void] {.gcsafe.} =
if s.closedLocal:
raise newLPStreamClosedError()
result = procCall write(BufferStream(s), pbytes, nbytes)
method write*(s: Channel, msg: string, msglen = -1) {.async, gcsafe.} =
if s.closedLocal:
raise newLPStreamClosedError()
result = procCall write(BufferStream(s), msg, msglen)
method write*(s: Channel, msg: seq[byte], msglen = -1) {.async, gcsafe.} =
if s.closedLocal:
raise newLPStreamClosedError()
result = procCall write(BufferStream(s), msg, msglen)