From 4edd2c9f8aa16ff5a2b67fc42b7052841ac007b9 Mon Sep 17 00:00:00 2001
From: Dmitriy Ryajov <dryajov@gmail.com>
Date: Mon, 2 Sep 2019 14:46:51 -0600
Subject: [PATCH] mplex move everything to one file

---
 libp2p/mplex.nim         | 145 +++++++++++++++++++++++++++++++++++++++
 libp2p/mplex/channel.nim |  58 ----------------
 libp2p/mplex/mplex.nim   |  31 ---------
 libp2p/mplex/types.nim   |  19 -----
 4 files changed, 145 insertions(+), 108 deletions(-)
 create mode 100644 libp2p/mplex.nim
 delete mode 100644 libp2p/mplex/channel.nim
 delete mode 100644 libp2p/mplex/mplex.nim
 delete mode 100644 libp2p/mplex/types.nim

diff --git a/libp2p/mplex.nim b/libp2p/mplex.nim
new file mode 100644
index 000000000..1ecb9cd21
--- /dev/null
+++ b/libp2p/mplex.nim
@@ -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
diff --git a/libp2p/mplex/channel.nim b/libp2p/mplex/channel.nim
deleted file mode 100644
index 96fa77a6b..000000000
--- a/libp2p/mplex/channel.nim
+++ /dev/null
@@ -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
diff --git a/libp2p/mplex/mplex.nim b/libp2p/mplex/mplex.nim
deleted file mode 100644
index 6e5c0685b..000000000
--- a/libp2p/mplex/mplex.nim
+++ /dev/null
@@ -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
diff --git a/libp2p/mplex/types.nim b/libp2p/mplex/types.nim
deleted file mode 100644
index 7cec08f0b..000000000
--- a/libp2p/mplex/types.nim
+++ /dev/null
@@ -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]