split mplex

libp2p/muxers/mplex/channel.nim

@@ -0,0 +1,34 @@
+## 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 ../../stream/bufferstream
+import types
+
+type
+  Channel* = ref object of BufferStream
+    id*: int
+    initiator*: bool
+    isReset*: bool
+    closedLocal*: bool
+    closedRemote*: bool
+    handlerFuture*: Future[void]
+
+proc newChannel*(id: int,
+                 initiator: bool,
+                 handler: WriteHandler,
+                 size: int = MaxMsgSize): Channel = 
+  new result
+  result.id = id
+  result.initiator = initiator
+  result.initBufferStream(handler, size)
+
+proc closed*(s: Channel): bool = s.closedLocal and s.closedRemote
+proc close*(s: Channel) {.async.} = discard
+proc reset*(s: Channel) {.async.} = discard

libp2p/muxers/mplex/coder.nim

@@ -0,0 +1,43 @@
+## 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, ../../varint, 
+       ../../vbuffer, mplex, types,
+       ../../stream/lpstream
+
+proc readHeader*(conn: Connection): Future[(uint, MessageType)] {.async, 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:
+        return (header shr 3, MessageType(header and 0x7))
+    if res != VarintStatus.Success:
+      buffer.setLen(0)
+      return
+  except LPStreamIncompleteError:
+    buffer.setLen(0)
+
+proc writeHeader*(conn: Connection,
+                  id: int,
+                  msgType: MessageType, 
+                  size: int) {.async, gcsafe.} =
+  ## write lenght prefixed
+  var buf = initVBuffer()
+  buf.writeVarint(LPSomeUVarint(id.uint shl 3 or msgType.uint))
+  if size > 0:
+    buf.writeVarint(LPSomeUVarint(size.uint))
+  buf.finish()
+  result = conn.write(buf.buffer)

libp2p/muxers/mplex/mplex.nim

@@ -9,27 +9,13 @@
 
 import tables, sequtils
 import chronos
-import ../varint, ../connection, 
+import ../../varint, ../../connection, 
-       ../vbuffer, ../protocol,
+       ../../vbuffer, ../../protocol,
-       ../stream/bufferstream, ../stream/lpstream, 
+       ../../stream/bufferstream, 
-       muxer
+       ../../stream/lpstream, ../muxer,
+       coder, types, channel
 
-const MaxMsgSize* = 1 shl 20 # 1mb
+type
-const MaxChannels* = 1000
-const MplexCodec* = "/mplex/6.7.0"
-
-type 
-  MplexUnknownMsgError* = object of CatchableError
-  MessageType* {.pure.} = enum
-    New,
-    MsgIn,
-    MsgOut,
-    CloseIn,
-    CloseOut,
-    ResetIn,
-    ResetOut
-
-  StreamHandler = proc(conn: Connection): Future[void] {.gcsafe.}
   Mplex* = ref object of Muxer
     remote*: Table[int, Channel]
     local*: Table[int, Channel]
@@ -37,71 +23,9 @@ type
     maxChannels*: uint
     streamHandler*: StreamHandler
 
-  Channel* = ref object of BufferStream
+proc newMplexUnknownMsgError(): ref MplexUnknownMsgError =
-    id*: int
-    initiator*: bool
-    isReset*: bool
-    closedLocal*: bool
-    closedRemote*: bool
-    mplex*: Mplex
-    handlerFuture*: Future[void]
-
-proc newMplexUnknownMsgError*(): ref MplexUnknownMsgError =
   result = newException(MplexUnknownMsgError, "Unknown mplex message type")
 
-##########################################
-##           Read/Write Helpers
-##########################################
-
-proc readHeader*(conn: Connection): Future[(uint, MessageType)] {.async, 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:
-        return (header shr 3, MessageType(header and 0x7))
-    if res != VarintStatus.Success:
-      buffer.setLen(0)
-      return
-  except LPStreamIncompleteError:
-    buffer.setLen(0)
-
-proc writeHeader*(conn: Connection,
-                  id: int,
-                  msgType: MessageType, 
-                  size: int) {.async, gcsafe.} =
-  ## write lenght prefixed
-  var buf = initVBuffer()
-  buf.writeVarint(LPSomeUVarint(id.uint shl 3 or msgType.uint))
-  if size > 0:
-    buf.writeVarint(LPSomeUVarint(size.uint))
-  buf.finish()
-  result = conn.write(buf.buffer)
-
-##########################################
-##               Channel
-##########################################
-
-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.initBufferStream(handler, size)
-
-proc closed*(s: Channel): bool = s.closedLocal and s.closedRemote
-proc close*(s: Channel) {.async.} = discard
-proc reset*(s: Channel) {.async.} = discard
-
 ##########################################
 ##               Mplex
 ##########################################
@@ -123,7 +47,7 @@ proc newStreamInternal*(m: Mplex,
     await m.connection.writeHeader(id, msgType, data.len) # write header
     await m.connection.write(data) # write data
 
-  result = newChannel(m, id, initiator, writeHandler)
+  result = newChannel(id, initiator, writeHandler)
   m.getChannelList(initiator)[id] = result
 
 proc newStreamInternal*(m: Mplex): Future[Channel] {.gcsafe.} = 

libp2p/muxers/mplex/types.nim

@@ -0,0 +1,28 @@
+## 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
+
+const MaxMsgSize* = 1 shl 20 # 1mb
+const MaxChannels* = 1000
+const MplexCodec* = "/mplex/6.7.0"
+
+type
+  MplexUnknownMsgError* = object of CatchableError
+  MessageType* {.pure.} = enum
+    New,
+    MsgIn,
+    MsgOut,
+    CloseIn,
+    CloseOut,
+    ResetIn,
+    ResetOut
+
+  StreamHandler* = proc(conn: Connection): Future[void] {.gcsafe.}

tests/testmplex.nim

@@ -1,8 +1,14 @@
 import unittest, sequtils, sugar
 import chronos, nimcrypto/utils
-import ../libp2p/muxers/mplex, ../libp2p/connection,
+import ../libp2p/connection, 
-       ../libp2p/stream/lpstream, ../libp2p/tcptransport,
+       ../libp2p/stream/lpstream, 
-       ../libp2p/transport, ../libp2p/multiaddress
+       ../libp2p/tcptransport, 
+       ../libp2p/transport, 
+       ../libp2p/multiaddress,
+       ../libp2p/muxers/mplex/mplex, 
+       ../libp2p/muxers/mplex/coder, 
+       ../libp2p/muxers/mplex/types,
+       ../libp2p/muxers/mplex/channel
 
 type
   TestEncodeStream = ref object of LPStream