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.
|
|
|
|
|
2019-09-09 17:33:32 +00:00
|
|
|
import chronos, chronicles
|
2019-09-12 17:07:34 +00:00
|
|
|
import types,
|
2019-09-08 07:59:14 +00:00
|
|
|
coder,
|
|
|
|
nimcrypto/utils,
|
2019-09-12 17:07:34 +00:00
|
|
|
../../stream/bufferstream,
|
|
|
|
../../stream/lpstream,
|
2020-03-23 06:03:36 +00:00
|
|
|
../../connection,
|
|
|
|
../../utility
|
2019-09-04 03:08:51 +00:00
|
|
|
|
2019-09-10 02:15:52 +00:00
|
|
|
logScope:
|
2019-09-28 19:54:52 +00:00
|
|
|
topic = "MplexChannel"
|
2019-09-10 02:15:52 +00:00
|
|
|
|
2019-09-04 03:08:51 +00:00
|
|
|
type
|
2019-09-12 17:07:34 +00:00
|
|
|
LPChannel* = ref object of BufferStream
|
2019-09-07 06:20:03 +00:00
|
|
|
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
|
2020-02-11 17:30:36 +00:00
|
|
|
isLazy*: bool
|
|
|
|
isOpen*: bool
|
2019-09-04 03:08:51 +00:00
|
|
|
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-04 03:08:51 +00:00
|
|
|
|
2019-09-07 06:20:03 +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 = "",
|
2020-02-11 17:30:36 +00:00
|
|
|
size: int = DefaultChannelSize,
|
|
|
|
lazy: bool = false): LPChannel =
|
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
|
2020-02-11 17:30:36 +00:00
|
|
|
result.isLazy = lazy
|
2019-09-04 03:08:51 +00:00
|
|
|
|
2019-09-04 06:40:11 +00:00
|
|
|
let chan = result
|
2019-12-10 20:50:35 +00:00
|
|
|
proc writeHandler(data: seq[byte]): Future[void] {.async.} =
|
2019-09-07 23:32:32 +00:00
|
|
|
# writes should happen in sequence
|
2020-03-23 06:03:36 +00:00
|
|
|
trace "sending data ", data = data.shortLog,
|
2019-12-04 04:44:54 +00:00
|
|
|
id = chan.id,
|
|
|
|
initiator = chan.initiator
|
|
|
|
|
2019-09-07 06:20:03 +00:00
|
|
|
await conn.writeMsg(chan.id, chan.msgCode, data) # write header
|
2019-09-04 06:40:11 +00:00
|
|
|
|
|
|
|
result.initBufferStream(writeHandler, size)
|
|
|
|
|
2019-12-04 04:44:54 +00:00
|
|
|
proc closeMessage(s: LPChannel) {.async.} =
|
2020-02-13 15:19:26 +00:00
|
|
|
await s.conn.writeMsg(s.id, s.closeCode) # write header
|
2019-09-04 06:40:11 +00:00
|
|
|
|
2019-12-10 20:50:35 +00:00
|
|
|
proc closedByRemote*(s: LPChannel) {.async.} =
|
2019-09-04 06:40:11 +00:00
|
|
|
s.closedRemote = true
|
|
|
|
|
2019-09-12 17:07:34 +00:00
|
|
|
proc cleanUp*(s: LPChannel): Future[void] =
|
2019-12-10 20:50:35 +00:00
|
|
|
# method which calls the underlying buffer's `close`
|
2019-12-04 04:44:54 +00:00
|
|
|
# method used instead of `close` since it's overloaded to
|
|
|
|
# simulate half-closed streams
|
2019-09-07 23:32:32 +00:00
|
|
|
result = procCall close(BufferStream(s))
|
|
|
|
|
2019-12-04 04:44:54 +00:00
|
|
|
proc open*(s: LPChannel): Future[void] =
|
2020-02-11 17:30:36 +00:00
|
|
|
s.isOpen = true
|
2019-12-04 04:44:54 +00:00
|
|
|
s.conn.writeMsg(s.id, MessageType.New, s.name)
|
|
|
|
|
2019-09-12 17:07:34 +00:00
|
|
|
method close*(s: LPChannel) {.async, gcsafe.} =
|
2019-09-04 06:40:11 +00:00
|
|
|
s.closedLocal = true
|
|
|
|
await s.closeMessage()
|
|
|
|
|
2019-12-04 04:44:54 +00:00
|
|
|
proc resetMessage(s: LPChannel) {.async.} =
|
2019-09-07 06:20:03 +00:00
|
|
|
await s.conn.writeMsg(s.id, s.resetCode)
|
2019-09-04 06:40:11 +00:00
|
|
|
|
2019-12-04 04:44:54 +00:00
|
|
|
proc resetByRemote*(s: LPChannel) {.async.} =
|
2019-09-04 16:40:05 +00:00
|
|
|
await allFutures(s.close(), s.closedByRemote())
|
2019-09-04 06:40:11 +00:00
|
|
|
s.isReset = true
|
|
|
|
|
2019-09-12 17:07:34 +00:00
|
|
|
proc reset*(s: LPChannel) {.async.} =
|
2019-09-04 16:40:05 +00:00
|
|
|
await allFutures(s.resetMessage(), s.resetByRemote())
|
2019-09-04 06:40:11 +00:00
|
|
|
|
2019-12-04 04:44:54 +00:00
|
|
|
method closed*(s: LPChannel): bool =
|
|
|
|
result = s.closedRemote and s.len == 0
|
2019-09-04 06:40:11 +00:00
|
|
|
|
2019-12-04 04:44:54 +00:00
|
|
|
proc pushTo*(s: LPChannel, data: seq[byte]): Future[void] =
|
|
|
|
if s.closedRemote or s.isReset:
|
2020-03-04 19:45:14 +00:00
|
|
|
var retFuture = newFuture[void]("LPChannel.pushTo")
|
|
|
|
retFuture.fail(newLPStreamEOFError())
|
|
|
|
return retFuture
|
|
|
|
|
2020-03-23 06:03:36 +00:00
|
|
|
trace "pushing data to channel", data = data.shortLog,
|
2019-12-10 20:50:35 +00:00
|
|
|
id = s.id,
|
2019-12-04 04:44:54 +00:00
|
|
|
initiator = s.initiator
|
|
|
|
|
2019-09-04 06:51:16 +00:00
|
|
|
result = procCall pushTo(BufferStream(s), data)
|
2019-09-04 06:40:11 +00:00
|
|
|
|
2019-12-04 04:44:54 +00:00
|
|
|
method read*(s: LPChannel, n = -1): Future[seq[byte]] =
|
|
|
|
if s.closed or s.isReset:
|
2020-03-04 19:45:14 +00:00
|
|
|
var retFuture = newFuture[seq[byte]]("LPChannel.read")
|
|
|
|
retFuture.fail(newLPStreamEOFError())
|
|
|
|
return retFuture
|
2019-12-04 04:44:54 +00:00
|
|
|
|
2019-09-04 06:40:11 +00:00
|
|
|
result = procCall read(BufferStream(s), n)
|
|
|
|
|
2019-12-04 04:44:54 +00:00
|
|
|
method readExactly*(s: LPChannel,
|
|
|
|
pbytes: pointer,
|
|
|
|
nbytes: int):
|
|
|
|
Future[void] =
|
|
|
|
if s.closed or s.isReset:
|
2020-03-04 19:45:14 +00:00
|
|
|
var retFuture = newFuture[void]("LPChannel.readExactly")
|
|
|
|
retFuture.fail(newLPStreamEOFError())
|
|
|
|
return retFuture
|
|
|
|
|
2019-09-04 06:40:11 +00:00
|
|
|
result = procCall readExactly(BufferStream(s), pbytes, nbytes)
|
|
|
|
|
2019-09-12 17:07:34 +00:00
|
|
|
method readLine*(s: LPChannel,
|
2019-09-04 06:40:11 +00:00
|
|
|
limit = 0,
|
|
|
|
sep = "\r\n"):
|
2019-12-04 04:44:54 +00:00
|
|
|
Future[string] =
|
|
|
|
if s.closed or s.isReset:
|
2020-03-04 19:45:14 +00:00
|
|
|
var retFuture = newFuture[string]("LPChannel.readLine")
|
|
|
|
retFuture.fail(newLPStreamEOFError())
|
|
|
|
return retFuture
|
|
|
|
|
2019-09-04 06:40:11 +00:00
|
|
|
result = procCall readLine(BufferStream(s), limit, sep)
|
|
|
|
|
2019-12-10 20:50:35 +00:00
|
|
|
method readOnce*(s: LPChannel,
|
|
|
|
pbytes: pointer,
|
|
|
|
nbytes: int):
|
2019-12-04 04:44:54 +00:00
|
|
|
Future[int] =
|
|
|
|
if s.closed or s.isReset:
|
2020-03-04 19:45:14 +00:00
|
|
|
var retFuture = newFuture[int]("LPChannel.readOnce")
|
|
|
|
retFuture.fail(newLPStreamEOFError())
|
|
|
|
return retFuture
|
|
|
|
|
2019-09-04 06:40:11 +00:00
|
|
|
result = procCall readOnce(BufferStream(s), pbytes, nbytes)
|
|
|
|
|
2019-09-12 17:07:34 +00:00
|
|
|
method readUntil*(s: LPChannel,
|
2019-09-04 06:40:11 +00:00
|
|
|
pbytes: pointer, nbytes: int,
|
2019-12-10 20:50:35 +00:00
|
|
|
sep: seq[byte]):
|
2019-12-04 04:44:54 +00:00
|
|
|
Future[int] =
|
|
|
|
if s.closed or s.isReset:
|
2020-03-04 19:45:14 +00:00
|
|
|
var retFuture = newFuture[int]("LPChannel.readUntil")
|
|
|
|
retFuture.fail(newLPStreamEOFError())
|
|
|
|
return retFuture
|
|
|
|
|
2019-09-04 06:40:11 +00:00
|
|
|
result = procCall readOnce(BufferStream(s), pbytes, nbytes)
|
|
|
|
|
2020-02-11 17:30:36 +00:00
|
|
|
template writePrefix: untyped =
|
|
|
|
if s.isLazy and not s.isOpen:
|
|
|
|
await s.open()
|
2019-12-04 04:44:54 +00:00
|
|
|
if s.closedLocal or s.isReset:
|
2019-12-10 20:50:35 +00:00
|
|
|
raise newLPStreamEOFError()
|
2020-02-11 17:30:36 +00:00
|
|
|
|
2020-03-04 19:45:14 +00:00
|
|
|
method write*(s: LPChannel, pbytes: pointer, nbytes: int) {.async.} =
|
2020-02-11 17:30:36 +00:00
|
|
|
writePrefix()
|
2020-03-04 19:45:14 +00:00
|
|
|
await procCall write(BufferStream(s), pbytes, nbytes)
|
2019-09-04 06:40:11 +00:00
|
|
|
|
2019-12-04 04:44:54 +00:00
|
|
|
method write*(s: LPChannel, msg: string, msglen = -1) {.async.} =
|
2020-02-11 17:30:36 +00:00
|
|
|
writePrefix()
|
2020-03-04 19:45:14 +00:00
|
|
|
await procCall write(BufferStream(s), msg, msglen)
|
2019-09-04 06:40:11 +00:00
|
|
|
|
2019-12-04 04:44:54 +00:00
|
|
|
method write*(s: LPChannel, msg: seq[byte], msglen = -1) {.async.} =
|
2020-02-11 17:30:36 +00:00
|
|
|
writePrefix()
|
2020-03-04 19:45:14 +00:00
|
|
|
await procCall write(BufferStream(s), msg, msglen)
|