removals (#159)

* remove unused stream methods
* reimplement some of them with proc's
* remove broken tests
* Error->Defect for defect
* warning fixes

libp2p/connection.nim

@@ -17,6 +17,8 @@ import peerinfo,
        varint,
        vbuffer
 
+export lpstream
+
 logScope:
   topic = "Connection"
 
@@ -103,46 +105,18 @@ proc newConnection*(stream: LPStream): Connection =
   ## create a new Connection for the specified async reader/writer
   result.init(stream)
 
-method read*(s: Connection, n = -1): Future[seq[byte]] {.gcsafe.} =
- s.stream.read(n)
-
 method readExactly*(s: Connection,
                     pbytes: pointer,
                     nbytes: int):
                     Future[void] {.gcsafe.} =
  s.stream.readExactly(pbytes, nbytes)
 
-method readLine*(s: Connection,
-                 limit = 0,
-                 sep = "\r\n"):
-                 Future[string] {.gcsafe.} =
-  s.stream.readLine(limit, sep)
-
 method readOnce*(s: Connection,
                  pbytes: pointer,
                  nbytes: int):
                  Future[int] {.gcsafe.} =
   s.stream.readOnce(pbytes, nbytes)
 
-method readUntil*(s: Connection,
-                  pbytes: pointer,
-                  nbytes: int,
-                  sep: seq[byte]):
-                  Future[int] {.gcsafe.} =
-  s.stream.readUntil(pbytes, nbytes, sep)
-
-method write*(s: Connection,
-              pbytes: pointer,
-              nbytes: int):
-              Future[void] {.gcsafe.} =
-  s.stream.write(pbytes, nbytes)
-
-method write*(s: Connection,
-              msg: string,
-              msglen = -1):
-              Future[void] {.gcsafe.} =
-  s.stream.write(msg, msglen)
-
 method write*(s: Connection,
               msg: seq[byte],
               msglen = -1):

libp2p/crypto/curve25519.nim

@@ -16,7 +16,6 @@
 # RFC @ https://tools.ietf.org/html/rfc7748
 
 import bearssl
-import nimcrypto/sysrand
 
 const
   Curve25519KeySize* = 32
@@ -32,7 +31,7 @@ proc intoCurve25519Key*(s: openarray[byte]): Curve25519Key =
   copyMem(addr result[0], unsafeaddr s[0], Curve25519KeySize)
 
 proc getBytes*(key: Curve25519Key): seq[byte] = @key
-  
+
 const
   ForbiddenCurveValues: array[12, Curve25519Key] = [
                 [0.byte, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
@@ -55,12 +54,12 @@ proc byteswap(buf: var Curve25519Key) {.inline.} =
       x = buf[i]
     buf[i] = buf[31 - i]
     buf[31 - i] = x
-    
+
 proc mul*(_: type[Curve25519], dst: var Curve25519Key, scalar: Curve25519Key, point: Curve25519Key) =
   let defaultBrEc = brEcGetDefault()
-  
+
   # The source point is provided in array G (of size Glen bytes);
-  # the multiplication result is written over it. 
+  # the multiplication result is written over it.
   dst = scalar
 
   # point needs to be big-endian
@@ -82,7 +81,7 @@ proc mulgen*(_: type[Curve25519], dst: var Curve25519Key, point: Curve25519Key)
   var
     rpoint = point
   rpoint.byteswap()
-  
+
   block iterate:
     while true:
       block derive:

libp2p/crypto/hkdf.nim

@@ -9,7 +9,8 @@
 
 # https://tools.ietf.org/html/rfc5869
 
-import math
+{.push raises: [Defect].}
+
 import nimcrypto
 import bearssl
 
@@ -17,16 +18,23 @@ type
   BearHKDFContext {.importc: "br_hkdf_context", header: "bearssl_kdf.h".} = object
   HKDFResult*[len: static int] = array[len, byte]
 
-proc br_hkdf_init(ctx: ptr BearHKDFContext; hashClass: ptr HashClass; salt: pointer; len: csize) {.importc: "br_hkdf_init", header: "bearssl_kdf.h".}
+proc br_hkdf_init(ctx: ptr BearHKDFContext; hashClass: ptr HashClass; salt: pointer; len: csize_t) {.importc: "br_hkdf_init", header: "bearssl_kdf.h", raises: [].}
-proc br_hkdf_inject(ctx: ptr BearHKDFContext; ikm: pointer; len: csize) {.importc: "br_hkdf_inject", header: "bearssl_kdf.h".}
+proc br_hkdf_inject(ctx: ptr BearHKDFContext; ikm: pointer; len: csize_t) {.importc: "br_hkdf_inject", header: "bearssl_kdf.h", raises: [].}
-proc br_hkdf_flip(ctx: ptr BearHKDFContext) {.importc: "br_hkdf_flip", header: "bearssl_kdf.h".}
+proc br_hkdf_flip(ctx: ptr BearHKDFContext) {.importc: "br_hkdf_flip", header: "bearssl_kdf.h", raises: [].}
-proc br_hkdf_produce(ctx: ptr BearHKDFContext; info: pointer; infoLen: csize; output: pointer; outputLen: csize) {.importc: "br_hkdf_produce", header: "bearssl_kdf.h".}
+proc br_hkdf_produce(ctx: ptr BearHKDFContext; info: pointer; infoLen: csize_t; output: pointer; outputLen: csize_t) {.importc: "br_hkdf_produce", header: "bearssl_kdf.h", raises: [].}
 
 proc hkdf*[T: sha256; len: static int](_: type[T]; salt, ikm, info: openarray[byte]; outputs: var openarray[HKDFResult[len]]) =
   var
     ctx: BearHKDFContext
-  br_hkdf_init(addr ctx, addr sha256Vtable, if salt.len > 0: unsafeaddr salt[0] else: nil, salt.len)
+  br_hkdf_init(
-  br_hkdf_inject(addr ctx, if ikm.len > 0: unsafeaddr ikm[0] else: nil, ikm.len)
+    addr ctx, addr sha256Vtable,
+    if salt.len > 0: unsafeaddr salt[0] else: nil, csize_t(salt.len))
+  br_hkdf_inject(
+    addr ctx, if ikm.len > 0: unsafeaddr ikm[0] else: nil, csize_t(ikm.len))
   br_hkdf_flip(addr ctx)
   for i in 0..outputs.high:
-    br_hkdf_produce(addr ctx, if info.len > 0: unsafeaddr info[0] else: nil, info.len, addr outputs[i][0], outputs[i].len)
+    br_hkdf_produce(
+      addr ctx,
+      if info.len > 0: unsafeaddr info[0]
+      else: nil, csize_t(info.len),
+      addr outputs[i][0], csize_t(outputs[i].len))

libp2p/multistream.nim

@@ -7,7 +7,7 @@
 ## This file may not be copied, modified, or distributed except according to
 ## those terms.
 
-import strutils
+import strutils, stew/byteutils
 import chronos, chronicles
 import connection,
        vbuffer,
@@ -36,8 +36,6 @@ type
   MultistreamSelect* = ref object of RootObj
     handlers*: seq[HandlerHolder]
     codec*: string
-    na: string
-    ls: string
 
   MultistreamHandshakeException* = object of CatchableError
 
@@ -48,8 +46,6 @@ proc newMultistreamHandshakeException*(): ref Exception {.inline.} =
 proc newMultistream*(): MultistreamSelect =
   new result
   result.codec = MSCodec
-  result.ls = Ls
-  result.na = Na
 
 proc select*(m: MultistreamSelect,
              conn: Connection,
@@ -71,7 +67,7 @@ proc select*(m: MultistreamSelect,
   if proto.len() == 0: # no protocols, must be a handshake call
     return
 
-  result = cast[string]((await conn.readLp())) # read the first proto
+  result = string.fromBytes(await conn.readLp()) # read the first proto
   trace "reading first requested proto"
   result.removeSuffix("\n")
   if result == proto[0]:
@@ -82,7 +78,7 @@ proc select*(m: MultistreamSelect,
     trace "selecting one of several protos"
     for p in proto[1..<proto.len()]:
       await conn.writeLp((p & "\n")) # select proto
-      result = cast[string]((await conn.readLp())) # read the first proto
+      result = string.fromBytes(await conn.readLp()) # read the first proto
       result.removeSuffix("\n")
       if result == p:
         trace "selected protocol", protocol = result
@@ -105,10 +101,10 @@ proc list*(m: MultistreamSelect,
   if not await m.select(conn):
     return
 
-  await conn.write(m.ls) # send ls
+  await conn.write(Ls) # send ls
 
   var list = newSeq[string]()
-  let ms = cast[string]((await conn.readLp()))
+  let ms = string.fromBytes(await conn.readLp())
   for s in ms.split("\n"):
     if s.len() > 0:
       list.add(s)
@@ -119,17 +115,17 @@ proc handle*(m: MultistreamSelect, conn: Connection) {.async, gcsafe.} =
   trace "handle: starting multistream handling"
   tryAndWarn "multistream handle":
     while not conn.closed:
-      var ms = cast[string]((await conn.readLp()))
+      var ms = string.fromBytes(await conn.readLp())
       ms.removeSuffix("\n")
 
       trace "handle: got request for ", ms
       if ms.len() <= 0:
         trace "handle: invalid proto"
-        await conn.write(m.na)
+        await conn.write(Na)
 
       if m.handlers.len() == 0:
         trace "handle: sending `na` for protocol ", protocol = ms
-        await conn.write(m.na)
+        await conn.write(Na)
         continue
 
       case ms:
@@ -150,7 +146,7 @@ proc handle*(m: MultistreamSelect, conn: Connection) {.async, gcsafe.} =
                 await h.protocol.handler(conn, ms)
                 return
           warn "no handlers for ", protocol = ms
-          await conn.write(m.na)
+          await conn.write(Na)
   trace "leaving multistream loop"
   # we might be tempted to close conn here but that would be a bad idea!
   # we indeed will reuse it later on

libp2p/muxers/mplex/lpchannel.nim

@@ -17,6 +17,8 @@ import types,
        ../../utility,
        ../../errors
 
+export lpstream
+
 logScope:
   topic = "MplexChannel"
 
@@ -136,11 +138,6 @@ template raiseEOF(): untyped =
   if s.closed or s.isReset:
     raise newLPStreamEOFError()
 
-method read*(s: LPChannel, n = -1): Future[seq[byte]] {.async.} =
-  raiseEOF()
-  result = (await procCall(read(BufferStream(s), n)))
-  await s.tryCleanup()
-
 method readExactly*(s: LPChannel,
                     pbytes: pointer,
                     nbytes: int):
@@ -149,14 +146,6 @@ method readExactly*(s: LPChannel,
   await procCall readExactly(BufferStream(s), pbytes, nbytes)
   await s.tryCleanup()
 
-method readLine*(s: LPChannel,
-                 limit = 0,
-                 sep = "\r\n"):
-                 Future[string] {.async.} =
-  raiseEOF()
-  result = await procCall readLine(BufferStream(s), limit, sep)
-  await s.tryCleanup()
-
 method readOnce*(s: LPChannel,
                  pbytes: pointer,
                  nbytes: int):
@@ -165,14 +154,6 @@ method readOnce*(s: LPChannel,
   result = await procCall readOnce(BufferStream(s), pbytes, nbytes)
   await s.tryCleanup()
 
-method readUntil*(s: LPChannel,
-                  pbytes: pointer, nbytes: int,
-                  sep: seq[byte]):
-                  Future[int] {.async.} =
-  raiseEOF()
-  result = await procCall readOnce(BufferStream(s), pbytes, nbytes)
-  await s.tryCleanup()
-
 template writePrefix: untyped =
   if s.closedLocal or s.isReset:
     raise newLPStreamEOFError()
@@ -180,14 +161,6 @@ template writePrefix: untyped =
   if s.isLazy and not s.isOpen:
     await s.open()
 
-method write*(s: LPChannel, pbytes: pointer, nbytes: int) {.async.} =
-  writePrefix()
-  await procCall write(BufferStream(s), pbytes, nbytes)
-
-method write*(s: LPChannel, msg: string, msglen = -1) {.async.} =
-  writePrefix()
-  await procCall write(BufferStream(s), msg, msglen)
-
 method write*(s: LPChannel, msg: seq[byte], msglen = -1) {.async.} =
   writePrefix()
   await procCall write(BufferStream(s), msg, msglen)

libp2p/muxers/mplex/mplex.nim

@@ -25,8 +25,6 @@ import ../muxer,
 logScope:
   topic = "Mplex"
 
-const DefaultRWTimeout = InfiniteDuration
-
 type
   Mplex* = ref object of Muxer
     remote*: Table[uint64, LPChannel]

libp2p/protocols/pubsub/floodsub.nim

@@ -7,7 +7,7 @@
 ## This file may not be copied, modified, or distributed except according to
 ## those terms.
 
-import sequtils, tables, options, sets, strutils
+import sequtils, tables, sets, strutils
 import chronos, chronicles
 import pubsub,
        pubsubpeer,
@@ -15,8 +15,8 @@ import pubsub,
        rpc/[messages, message],
        ../../crypto/crypto,
        ../../connection,
-       ../../peerinfo,
        ../../peer,
+       ../../peerinfo,
        ../../utility,
        ../../errors
 

libp2p/protocols/pubsub/pubsub.nim

@@ -13,8 +13,7 @@ import pubsubpeer,
        rpc/messages,
        ../protocol,
        ../../connection,
-       ../../peerinfo,
+       ../../peerinfo
-       ../../peer
 
 export PubSubPeer
 

libp2p/protocols/pubsub/rpc/protobuf.nim

@@ -10,10 +10,8 @@
 import options
 import chronicles
 import messages,
-       ../../../protobuf/minprotobuf,
+       ../../../utility,
-       ../../../crypto/crypto,
+       ../../../protobuf/minprotobuf
-       ../../../peer,
-       ../../../utility
 
 proc encodeGraft*(graft: ControlGraft, pb: var ProtoBuffer) {.gcsafe.} =
   pb.write(initProtoField(1, graft.topicID))
@@ -37,7 +35,7 @@ proc decodePrune*(pb: var ProtoBuffer): seq[ControlPrune] {.gcsafe.} =
     var topic: string
     if pb.getString(1, topic) < 0:
       break
-   
+
     trace "read topic field from prune msg", topicID = topic
     result.add(ControlPrune(topicID: topic))
 

libp2p/protocols/secure/noise.nim

@@ -520,7 +520,7 @@ method init*(p: Noise) {.gcsafe.} =
   procCall Secure(p).init()
   p.codec = NoiseCodec
 
-method secure*(p: Noise, conn: Connection): Future[Connection] {.async, gcsafe.} =
+proc secure*(p: Noise, conn: Connection): Future[Connection] {.async, gcsafe.} =
   trace "Noise.secure called", initiator=p.outgoing
   try:
     result = await p.handleConn(conn, p.outgoing)

libp2p/protocols/secure/secio.nim

@@ -14,7 +14,6 @@ import secure,
        ../../stream/lpstream,
        ../../crypto/crypto,
        ../../crypto/ecnist,
-       ../../protobuf/minprotobuf,
        ../../peer,
        ../../utility
 export hmac, sha2, sha, hash, rijndael, bcmode

libp2p/protocols/secure/secure.nim

@@ -11,7 +11,6 @@ import options
 import chronos, chronicles
 import ../protocol,
        ../../stream/bufferstream,
-       ../../crypto/crypto,
        ../../connection,
        ../../peerinfo,
        ../../utility

libp2p/stream/bufferstream.nim

@@ -34,6 +34,8 @@ import deques, math, oids
 import chronos, chronicles, metrics
 import ../stream/lpstream
 
+export lpstream
+
 const
   BufferStreamTrackerName* = "libp2p.bufferstream"
   DefaultBufferSize* = 1024
@@ -168,31 +170,6 @@ proc pushTo*(s: BufferStream, data: seq[byte]) {.async.} =
   finally:
     s.lock.release()
 
-method read*(s: BufferStream, n = -1): Future[seq[byte]] {.async.} =
-  ## Read all bytes (n <= 0) or exactly `n` bytes from buffer
-  ##
-  ## This procedure allocates buffer seq[byte] and return it as result.
-  ##
-  when chronicles.enabledLogLevel == LogLevel.TRACE:
-    logScope:
-      stream_oid = $s.oid
-
-  trace "read()", requested_bytes = n
-  var size = if n > 0: n else: s.readBuf.len()
-  var index = 0
-
-  if s.readBuf.len() == 0:
-    await s.requestReadBytes()
-
-  while index < size:
-    while s.readBuf.len() > 0 and index < size:
-      result.add(s.popFirst())
-      inc(index)
-    trace "read()", read_bytes = index
-
-    if index < size:
-      await s.requestReadBytes()
-
 method readExactly*(s: BufferStream,
                     pbytes: pointer,
                     nbytes: int):
@@ -207,53 +184,21 @@ method readExactly*(s: BufferStream,
     logScope:
       stream_oid = $s.oid
 
-  var buff: seq[byte]
+  trace "read()", requested_bytes = nbytes
-  try:
-    buff = await s.read(nbytes)
-  except LPStreamEOFError as exc:
-    trace "Exception occurred", exc = exc.msg
-
-  if nbytes > buff.len():
-    raise newLPStreamIncompleteError()
-
-  copyMem(pbytes, addr buff[0], nbytes)
-
-method readLine*(s: BufferStream,
-                 limit = 0,
-                 sep = "\r\n"):
-                 Future[string] {.async.} =
-  ## Read one line from read-only stream ``rstream``, where ``"line"`` is a
-  ## sequence of bytes ending with ``sep`` (default is ``"\r\n"``).
-  ##
-  ## If EOF is received, and ``sep`` was not found, the method will return the
-  ## partial read bytes.
-  ##
-  ## If the EOF was received and the internal buffer is empty, return an
-  ## empty string.
-  ##
-  ## If ``limit`` more then 0, then result string will be limited to ``limit``
-  ## bytes.
-  ##
-  result = ""
-  var lim = if limit <= 0: -1 else: limit
-  var state = 0
   var index = 0
 
-  index = 0
+  if s.readBuf.len() == 0:
-  while index < s.readBuf.len:
+    await s.requestReadBytes()
-    let ch = char(s.readBuf[index])
+
-    if sep[state] == ch:
+  let output = cast[ptr UncheckedArray[byte]](pbytes)
-      inc(state)
+  while index < nbytes:
-      if state == len(sep):
+    while s.readBuf.len() > 0 and index < nbytes:
-        s.shrink(index + 1)
+      output[index] = s.popFirst()
-        break
+      inc(index)
-    else:
+    trace "readExactly()", read_bytes = index
-      state = 0
+
-      result.add(ch)
+    if index < nbytes:
-      if len(result) == lim:
+      await s.requestReadBytes()
-        s.shrink(index + 1)
-        break
-    inc(index)
 
 method readOnce*(s: BufferStream,
                  pbytes: pointer,
@@ -271,93 +216,6 @@ method readOnce*(s: BufferStream,
   await s.readExactly(pbytes, len)
   result = len
 
-method readUntil*(s: BufferStream,
-                  pbytes: pointer,
-                  nbytes: int,
-                  sep: seq[byte]):
-                  Future[int] {.async.} =
-  ## Read data from the read-only stream ``rstream`` until separator ``sep`` is
-  ## found.
-  ##
-  ## On success, the data and separator will be removed from the internal
-  ## buffer (consumed). Returned data will include the separator at the end.
-  ##
-  ## If EOF is received, and `sep` was not found, procedure will raise
-  ## ``LPStreamIncompleteError``.
-  ##
-  ## If ``nbytes`` bytes has been received and `sep` was not found, procedure
-  ## will raise ``LPStreamLimitError``.
-  ##
-  ## Procedure returns actual number of bytes read.
-  ##
-  var
-    dest = cast[ptr UncheckedArray[byte]](pbytes)
-    state = 0
-    k = 0
-
-  let datalen = s.readBuf.len()
-  if datalen == 0 and s.readBuf.len() == 0:
-    raise newLPStreamIncompleteError()
-
-  var index = 0
-  while index < datalen:
-    let ch = s.readBuf[index]
-    if sep[state] == ch:
-      inc(state)
-    else:
-      state = 0
-    if k < nbytes:
-      dest[k] = ch
-      inc(k)
-    else:
-      raise newLPStreamLimitError()
-    if state == len(sep):
-      break
-    inc(index)
-
-  if state == len(sep):
-    s.shrink(index + 1)
-    result = k
-  else:
-    s.shrink(datalen)
-
-method write*(s: BufferStream,
-              pbytes: pointer,
-              nbytes: int): Future[void] =
-  ## Consume (discard) all bytes (n <= 0) or ``n`` bytes from read-only stream
-  ## ``rstream``.
-  ##
-  ## Return number of bytes actually consumed (discarded).
-  ##
-  if isNil(s.writeHandler):
-    var retFuture = newFuture[void]("BufferStream.write(pointer)")
-    retFuture.fail(newNotWritableError())
-    return retFuture
-
-  var buf: seq[byte] = newSeq[byte](nbytes)
-  copyMem(addr buf[0], pbytes, nbytes)
-  result = s.writeHandler(buf)
-
-method write*(s: BufferStream,
-              msg: string,
-              msglen = -1): Future[void] =
-  ## Write string ``sbytes`` of length ``msglen`` to writer stream ``wstream``.
-  ##
-  ## String ``sbytes`` must not be zero-length.
-  ##
-  ## If ``msglen < 0`` whole string ``sbytes`` will be writen to stream.
-  ## If ``msglen > len(sbytes)`` only ``len(sbytes)`` bytes will be written to
-  ## stream.
-  ##
-  if isNil(s.writeHandler):
-    var retFuture = newFuture[void]("BufferStream.write(string)")
-    retFuture.fail(newNotWritableError())
-    return retFuture
-
-  var buf = ""
-  shallowCopy(buf, if msglen > 0: msg[0..<msglen] else: msg)
-  result = s.writeHandler(cast[seq[byte]](buf))
-
 method write*(s: BufferStream,
               msg: seq[byte],
               msglen = -1): Future[void] =
@@ -375,9 +233,7 @@ method write*(s: BufferStream,
     retFuture.fail(newNotWritableError())
     return retFuture
 
-  var buf: seq[byte]
+  result = s.writeHandler(if msglen >= 0: msg[0..<msglen] else: msg)
-  shallowCopy(buf, if msglen > 0: msg[0..<msglen] else: msg)
-  result = s.writeHandler(buf)
 
 proc pipe*(s: BufferStream,
            target: BufferStream): BufferStream =

libp2p/stream/chronosstream.nim

@@ -36,17 +36,10 @@ template withExceptions(body: untyped) =
   except TransportLimitError:
     raise newLPStreamLimitError()
   except TransportError as exc:
-    raise newLPStreamIncorrectError(exc.msg)
+    raise newLPStreamIncorrectDefect(exc.msg)
   except AsyncStreamIncompleteError:
     raise newLPStreamIncompleteError()
 
-method read*(s: ChronosStream, n = -1): Future[seq[byte]] {.async.} =
-  if s.reader.atEof:
-    raise newLPStreamEOFError()
-
-  withExceptions:
-    result = await s.reader.read(n)
-
 method readExactly*(s: ChronosStream,
                     pbytes: pointer,
                     nbytes: int): Future[void] {.async.} =
@@ -56,13 +49,6 @@ method readExactly*(s: ChronosStream,
   withExceptions:
     await s.reader.readExactly(pbytes, nbytes)
 
-method readLine*(s: ChronosStream, limit = 0, sep = "\r\n"): Future[string] {.async.} =
-  if s.reader.atEof:
-    raise newLPStreamEOFError()
-
-  withExceptions:
-    result = await s.reader.readLine(limit, sep)
-
 method readOnce*(s: ChronosStream, pbytes: pointer, nbytes: int): Future[int] {.async.} =
   if s.reader.atEof:
     raise newLPStreamEOFError()
@@ -70,30 +56,6 @@ method readOnce*(s: ChronosStream, pbytes: pointer, nbytes: int): Future[int] {.
   withExceptions:
     result = await s.reader.readOnce(pbytes, nbytes)
 
-method readUntil*(s: ChronosStream,
-                  pbytes: pointer,
-                  nbytes: int,
-                  sep: seq[byte]): Future[int] {.async.} =
-  if s.reader.atEof:
-    raise newLPStreamEOFError()
-
-  withExceptions:
-    result = await s.reader.readUntil(pbytes, nbytes, sep)
-
-method write*(s: ChronosStream, pbytes: pointer, nbytes: int) {.async.} =
-  if s.writer.atEof:
-    raise newLPStreamEOFError()
-
-  withExceptions:
-    await s.writer.write(pbytes, nbytes)
-
-method write*(s: ChronosStream, msg: string, msglen = -1) {.async.} =
-  if s.writer.atEof:
-    raise newLPStreamEOFError()
-
-  withExceptions:
-    await s.writer.write(msg, msglen)
-
 method write*(s: ChronosStream, msg: seq[byte], msglen = -1) {.async.} =
   if s.writer.atEof:
     raise newLPStreamEOFError()

libp2p/stream/lpstream.nim

@@ -19,7 +19,7 @@ type
 
   LPStreamError* = object of CatchableError
   LPStreamIncompleteError* = object of LPStreamError
-  LPStreamIncorrectError* = object of Defect
+  LPStreamIncorrectDefect* = object of Defect
   LPStreamLimitError* = object of LPStreamError
   LPStreamReadError* = object of LPStreamError
     par*: ref Exception
@@ -45,8 +45,8 @@ proc newLPStreamIncompleteError*(): ref Exception {.inline.} =
 proc newLPStreamLimitError*(): ref Exception {.inline.} =
   result = newException(LPStreamLimitError, "Buffer limit reached")
 
-proc newLPStreamIncorrectError*(m: string): ref Exception {.inline.} =
+proc newLPStreamIncorrectDefect*(m: string): ref Exception {.inline.} =
-  result = newException(LPStreamIncorrectError, m)
+  result = newException(LPStreamIncorrectDefect, m)
 
 proc newLPStreamEOFError*(): ref Exception {.inline.} =
   result = newException(LPStreamEOFError, "Stream EOF!")
@@ -54,24 +54,12 @@ proc newLPStreamEOFError*(): ref Exception {.inline.} =
 method closed*(s: LPStream): bool {.base, inline.} =
   s.isClosed
 
-method read*(s: LPStream,
-             n = -1):
-             Future[seq[byte]] {.base, async.} =
-  doAssert(false, "not implemented!")
-
 method readExactly*(s: LPStream,
                     pbytes: pointer,
                     nbytes: int):
                     Future[void] {.base, async.} =
   doAssert(false, "not implemented!")
 
-method readLine*(s: LPStream,
-                 limit = 0,
-                 sep = "\r\n"):
-                 Future[string]
-  {.base, async.} =
-  doAssert(false, "not implemented!")
-
 method readOnce*(s: LPStream,
                  pbytes: pointer,
                  nbytes: int):
@@ -79,26 +67,56 @@ method readOnce*(s: LPStream,
   {.base, async.} =
   doAssert(false, "not implemented!")
 
-method readUntil*(s: LPStream,
+proc read*(s: LPStream, nbytes: int): Future[seq[byte]] {.async, deprecated: "readExactly".} =
-                  pbytes: pointer,
+  # This function is deprecated - it was broken and used inappropriately as
-                  nbytes: int,
+  # `readExacltly` in tests and code - tests still need refactoring to remove
-                  sep: seq[byte]):
+  # any calls
-                  Future[int]
+  # `read` without nbytes was also incorrectly implemented - it worked more
-  {.base, async.} =
+  # like `readOnce` in that it would not wait for stream to close, in
-  doAssert(false, "not implemented!")
+  # BufferStream in particular - both tests and implementation were broken
+  var ret = newSeq[byte](nbytes)
+  await readExactly(s, addr ret[0], ret.len)
+  return ret
 
-method write*(s: LPStream, pbytes: pointer, nbytes: int)
+proc readLine*(s: LPStream, limit = 0, sep = "\r\n"): Future[string] {.async, deprecated: "todo".} =
-  {.base, async.} =
+  # TODO replace with something that exploits buffering better
-  doAssert(false, "not implemented!")
+  var lim = if limit <= 0: -1 else: limit
+  var state = 0
 
-method write*(s: LPStream, msg: string, msglen = -1)
+  try:
-  {.base, async.} =
+    while true:
-  doAssert(false, "not implemented!")
+      var ch: char
+      await readExactly(s, addr ch, 1)
+
+      if sep[state] == ch:
+        inc(state)
+        if state == len(sep):
+          break
+      else:
+        state = 0
+        if limit > 0:
+          let missing = min(state, lim - len(result) - 1)
+          result.add(sep[0 ..< missing])
+        else:
+          result.add(sep[0 ..< state])
+
+        result.add(ch)
+        if len(result) == lim:
+          break
+  except LPStreamIncompleteError, LPStreamReadError:
+    discard # EOF, in which case we should return whatever we read so far..
 
 method write*(s: LPStream, msg: seq[byte], msglen = -1)
   {.base, async.} =
   doAssert(false, "not implemented!")
 
+proc write*(s: LPStream, pbytes: pointer, nbytes: int): Future[void] {.deprecated: "seq".} =
+  s.write(@(toOpenArray(cast[ptr UncheckedArray[byte]](pbytes), 0, nbytes - 1)))
+
+proc write*(s: LPStream, msg: string, msglen = -1): Future[void] =
+  let nbytes = if msglen >= 0: msglen else: msg.len
+  s.write(@(toOpenArrayByte(msg, 0, nbytes - 1)))
+
 method close*(s: LPStream)
   {.base, async.} =
   doAssert(false, "not implemented!")

libp2p/transports/tcptransport.nim

@@ -7,7 +7,7 @@
 ## This file may not be copied, modified, or distributed except according to
 ## those terms.
 
-import chronos, chronicles, sequtils, sets
+import chronos, chronicles, sequtils
 import transport,
        ../errors,
        ../wire,

tests/pubsub/testpubsub.nim

@@ -1,3 +1,5 @@
+{.used.}
+
 import testgossipinternal,
        testfloodsub,
        testgossipsub,

tests/testbufferstream.nim

@@ -45,22 +45,6 @@ suite "BufferStream":
     check:
       waitFor(testPushTo()) == true
 
-  test "read":
-    proc testRead(): Future[bool] {.async.} =
-      proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
-      let buff = newBufferStream(writeHandler, 10)
-      check buff.len == 0
-
-      await buff.pushTo(cast[seq[byte]](@"12345"))
-      check @"12345" == cast[string](await buff.read())
-
-      result = true
-
-      await buff.close()
-
-    check:
-      waitFor(testRead()) == true
-
   test "read with size":
     proc testRead(): Future[bool] {.async.} =
       proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
@@ -99,32 +83,6 @@ suite "BufferStream":
     check:
       waitFor(testRead()) == true
 
-  test "read all from small buffer":
-    proc testRead(): Future[bool] {.async.} =
-      proc writeHandler(data: seq[byte]) {.async, gcsafe.} =
-        discard
-
-      let buff = newBufferStream(writeHandler, 4)
-      check buff.len == 0
-
-      proc reader() {.async.} =
-        var size = 0
-        while size != 5:
-          var msg = await buff.read()
-          size += msg.len
-        check size == 5
-
-      var fut = reader()
-      await buff.pushTo(cast[seq[byte]](@"12345"))
-      await fut
-
-      result = true
-
-      await buff.close()
-
-    check:
-      waitFor(testRead()) == true
-
   test "readExactly":
     proc testReadExactly(): Future[bool] {.async.} =
       proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
@@ -144,23 +102,6 @@ suite "BufferStream":
     check:
       waitFor(testReadExactly()) == true
 
-  test "readLine":
-    proc testReadLine(): Future[bool] {.async.} =
-      proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
-      let buff = newBufferStream(writeHandler, 16)
-      check buff.len == 0
-
-      await buff.pushTo(cast[seq[byte]](@"12345\n67890"))
-      check buff.len == 11
-      check "12345" == await buff.readLine(0, "\n")
-
-      result = true
-
-      await buff.close()
-
-    check:
-      waitFor(testReadLine()) == true
-
   test "readOnce":
     proc testReadOnce(): Future[bool] {.async.} =
       proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
@@ -182,27 +123,6 @@ suite "BufferStream":
     check:
       waitFor(testReadOnce()) == true
 
-  test "readUntil":
-    proc testReadUntil(): Future[bool] {.async.} =
-      proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
-      let buff = newBufferStream(writeHandler, 10)
-      check buff.len == 0
-
-      var data: seq[byte] = newSeq[byte](3)
-      await buff.pushTo(cast[seq[byte]](@"123$45"))
-      check buff.len == 6
-      let readFut = buff.readUntil(addr data[0], 5, @[byte('$')])
-
-      check (await readFut) == 4
-      check cast[string](data) == @['1', '2', '3']
-
-      result = true
-
-      await buff.close()
-
-    check:
-      waitFor(testReadUntil()) == true
-
   test "write ptr":
     proc testWritePtr(): Future[bool] {.async.} =
       proc writeHandler(data: seq[byte]) {.async, gcsafe.} =

tests/testcrypto.nim

@@ -510,7 +510,7 @@ suite "Key interface test suite":
       public1: Curve25519Key
       public1Test: Curve25519Key = [0x9.byte, 0xd, 0x85, 0xe5, 0x99, 0xea, 0x8e, 0x2b, 0xee, 0xb6, 0x13, 0x4, 0xd3, 0x7b, 0xe1, 0xe, 0xc5, 0xc9, 0x5, 0xf9, 0x92, 0x7d, 0x32, 0xf4, 0x2a, 0x9a, 0xa, 0xfb, 0x3e, 0xb, 0x40, 0x74]
 
-    Curve25519.mul(public1, base, private1) 
+    Curve25519.mul(public1, base, private1)
     check public1.toHex == public1Test.toHex
 
     # RFC vectors
@@ -523,10 +523,9 @@ suite "Key interface test suite":
       private2 = fromHex("5dab087e624a8a4b79e17f8b83800ee66f3bb1292618b6fd1c2f8b27ff88e0eb").intoCurve25519Key
       p1Pub = private1.public()
       p2Pub = private2.public()
-      p2Gen: Curve25519Key
     check p1Pub.toHex  == "8520F0098930A754748B7DDCB43EF75A0DBF3A0D26381AF4EBA4A98EAA9B4E6A"
     check p2Pub.toHex  == "DE9EDB7D7B7DC1B4D35B61C2ECE435373F8343C85B78674DADFC7E146F882B4F"
-    
+
     var
       secret1: Curve25519Key
       secret2: Curve25519Key
@@ -534,7 +533,7 @@ suite "Key interface test suite":
     Curve25519.mul(secret2, p1Pub, private2)
     check secret1.toHex == secret2.toHex
 
-  
+
   test "HKDF 1":
     let
       ikm = fromHex("0x0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b0b")

tests/testidentify.nim

@@ -1,5 +1,5 @@
 import unittest, options
-import chronos, strutils, sequtils
+import chronos, strutils
 import ../libp2p/[protocols/identify,
                   multiaddress,
                   peerinfo,
@@ -8,7 +8,6 @@ import ../libp2p/[protocols/identify,
                   multistream,
                   transports/transport,
                   transports/tcptransport,
-                  protocols/protocol,
                   crypto/crypto]
 
 when defined(nimHasUsed): {.used.}

tests/testmplex.nim

@@ -1,4 +1,4 @@
-import unittest, sequtils, sugar, strformat, options, strformat, random
+import unittest, strformat, strformat, random
 import chronos, nimcrypto/utils, chronicles
 import ../libp2p/[errors,
                   connection,
@@ -6,7 +6,6 @@ import ../libp2p/[errors,
                   stream/bufferstream,
                   transports/tcptransport,
                   transports/transport,
-                  protocols/identify,
                   multiaddress,
                   muxers/mplex/mplex,
                   muxers/mplex/coder,
@@ -470,8 +469,8 @@ suite "Mplex":
       try:
         await chann.pushTo(cast[seq[byte]]("Hello!"))
         await chann.closedByRemote()
-        discard await chann.read() # this should work, since there is data in the buffer
+        discard await chann.read(6) # this should work, since there is data in the buffer
-        discard await chann.read() # this should throw
+        discard await chann.read(6) # this should throw
       finally:
         await chann.cleanUp()
         await conn.close()
@@ -616,7 +615,7 @@ suite "Mplex":
 
       try:
         await chann.reset()
-        var data = await chann.read()
+        var data = await chann.read(1)
         doAssert(len(data) == 1)
       finally:
         await chann.cleanUp()

tests/testmultistream.nim

@@ -1,4 +1,4 @@
-import unittest, strutils, sequtils, strformat, options
+import unittest, strutils, sequtils, strformat, stew/byteutils
 import chronos
 import ../libp2p/errors,
        ../libp2p/connection,
@@ -9,10 +9,7 @@ import ../libp2p/errors,
        ../libp2p/multiaddress,
        ../libp2p/transports/transport,
        ../libp2p/transports/tcptransport,
-       ../libp2p/protocols/protocol,
+       ../libp2p/protocols/protocol
-       ../libp2p/crypto/crypto,
-       ../libp2p/peerinfo,
-       ../libp2p/peer
 
 when defined(nimHasUsed): {.used.}
 
@@ -54,9 +51,6 @@ method readExactly*(s: TestSelectStream,
 method write*(s: TestSelectStream, msg: seq[byte], msglen = -1)
   {.async, gcsafe.} = discard
 
-method write*(s: TestSelectStream, msg: string, msglen = -1)
-  {.async, gcsafe.} = discard
-
 method close(s: TestSelectStream) {.async, gcsafe.} =
   s.isClosed = true
 
@@ -95,15 +89,13 @@ method readExactly*(s: TestLsStream,
       var buf = "ls\n"
       copyMem(pbytes, addr buf[0], buf.len())
     else:
-      copyMem(pbytes, cstring(Na), Na.len())
+      var buf = "na\n"
+      copyMem(pbytes, addr buf[0], buf.len())
 
 method write*(s: TestLsStream, msg: seq[byte], msglen = -1) {.async, gcsafe.} =
   if s.step == 4:
     await s.ls(msg)
 
-method write*(s: TestLsStream, msg: string, msglen = -1)
-  {.async, gcsafe.} = discard
-
 method close(s: TestLsStream) {.async, gcsafe.} =
   s.isClosed = true
 
@@ -147,9 +139,9 @@ method readExactly*(s: TestNaStream,
               cstring("\0x3na\n"),
               "\0x3na\n".len())
 
-method write*(s: TestNaStream, msg: string, msglen = -1) {.async, gcsafe.} =
+method write*(s: TestNaStream, msg: seq[byte], msglen = -1) {.async, gcsafe.} =
   if s.step == 4:
-    await s.na(msg)
+    await s.na(string.fromBytes(msg))
 
 method close(s: TestNaStream) {.async, gcsafe.} =
   s.isClosed = true
@@ -174,7 +166,7 @@ suite "Multistream select":
       if not isNil(tracker):
         # echo tracker.dump()
         check tracker.isLeaked() == false
- 
+
   test "test select custom proto":
     proc testSelect(): Future[bool] {.async.} =
       let ms = newMultistream()
@@ -240,7 +232,7 @@ suite "Multistream select":
       let conn = newConnection(newTestNaStream(testNaHandler))
 
       proc testNaHandler(msg: string): Future[void] {.async, gcsafe.} =
-        check cast[string](msg) == Na
+        check msg == Na
         await conn.close()
 
       var protocol: LPProtocol = new LPProtocol
@@ -294,7 +286,7 @@ suite "Multistream select":
       let hello = cast[string](await conn.readLp())
       result = hello == "Hello!"
       await conn.close()
-     
+
       await transport2.close()
       await transport1.close()
 

tests/testnoise.nim

@@ -7,6 +7,8 @@
 ## This file may not be copied, modified, or distributed except according to
 ## those terms.
 
+{.used.}
+
 import unittest, tables
 import chronos
 import chronicles
@@ -23,7 +25,6 @@ import ../libp2p/[switch,
                   multiaddress,
                   peerinfo,
                   crypto/crypto,
-                  peer,
                   protocols/protocol,
                   muxers/muxer,
                   muxers/mplex/mplex,
@@ -83,7 +84,7 @@ suite "Noise":
       if not isNil(tracker):
         # echo tracker.dump()
         check tracker.isLeaked() == false
- 
+
   test "e2e: handle write + noise":
     proc testListenerDialer(): Future[bool] {.async.} =
       let
@@ -256,7 +257,7 @@ suite "Noise":
   #           error getCurrentExceptionMsg()
   #         finally:
   #           await conn.close()
-     
+
   #       ms.addHandler(proto, noise)
 
   #       let
@@ -311,7 +312,7 @@ suite "Noise":
   #           trace "ms.handle exited"
   #         finally:
   #           await conn.close()
-         
+
   #       let
   #         clientConn = await transport.listen(local, connHandler)
   #       await clientConn

tests/testpeerinfo.nim

@@ -1,3 +1,4 @@
+{.used.}
 
 import unittest, options
 import chronos

tests/testswitch.nim

@@ -13,7 +13,6 @@ import ../libp2p/[errors,
                   multiaddress,
                   peerinfo,
                   crypto/crypto,
-                  peer,
                   protocols/protocol,
                   muxers/muxer,
                   muxers/mplex/mplex,

tests/testtransport.nim

@@ -1,7 +1,6 @@
 import unittest
 import chronos
-import ../libp2p/[errors,
+import ../libp2p/[connection,
-                  connection,
                   transports/transport,
                   transports/tcptransport,
                   multiaddress,
@@ -13,12 +12,6 @@ const
   StreamTransportTrackerName = "stream.transport"
   StreamServerTrackerName = "stream.server"
 
-template ignoreErrors(body: untyped): untyped =
-  try:
-    body
-  except:
-    echo getCurrentExceptionMsg()
-
 suite "TCP transport":
   teardown:
     check: