Fix and refactoring of some procedures which are able to return nil as result (#97)

* Fix do not return nil as result.

* Fix mplex test to properly raise.

libp2p/muxers/mplex/lpchannel.nim

@@ -96,7 +96,10 @@ method closed*(s: LPChannel): bool =
 
 proc pushTo*(s: LPChannel, data: seq[byte]): Future[void] =
   if s.closedRemote or s.isReset:
-    raise newLPStreamEOFError()
+    var retFuture = newFuture[void]("LPChannel.pushTo")
+    retFuture.fail(newLPStreamEOFError())
+    return retFuture
+
   trace "pushing data to channel", data = data.toHex(),
                                    id = s.id,
                                    initiator = s.initiator
@@ -105,7 +108,9 @@ proc pushTo*(s: LPChannel, data: seq[byte]): Future[void] =
 
 method read*(s: LPChannel, n = -1): Future[seq[byte]] =
   if s.closed or s.isReset:
-    raise newLPStreamEOFError()
+    var retFuture = newFuture[seq[byte]]("LPChannel.read")
+    retFuture.fail(newLPStreamEOFError())
+    return retFuture
 
   result = procCall read(BufferStream(s), n)
 
@@ -114,7 +119,10 @@ method readExactly*(s: LPChannel,
                     nbytes: int):
                     Future[void] =
   if s.closed or s.isReset:
-    raise newLPStreamEOFError()
+    var retFuture = newFuture[void]("LPChannel.readExactly")
+    retFuture.fail(newLPStreamEOFError())
+    return retFuture
+
   result = procCall readExactly(BufferStream(s), pbytes, nbytes)
 
 method readLine*(s: LPChannel,
@@ -122,7 +130,10 @@ method readLine*(s: LPChannel,
                  sep = "\r\n"):
                  Future[string] =
   if s.closed or s.isReset:
-    raise newLPStreamEOFError()
+    var retFuture = newFuture[string]("LPChannel.readLine")
+    retFuture.fail(newLPStreamEOFError())
+    return retFuture
+
   result = procCall readLine(BufferStream(s), limit, sep)
 
 method readOnce*(s: LPChannel,
@@ -130,7 +141,10 @@ method readOnce*(s: LPChannel,
                  nbytes: int):
                  Future[int] =
   if s.closed or s.isReset:
-    raise newLPStreamEOFError()
+    var retFuture = newFuture[int]("LPChannel.readOnce")
+    retFuture.fail(newLPStreamEOFError())
+    return retFuture
+
   result = procCall readOnce(BufferStream(s), pbytes, nbytes)
 
 method readUntil*(s: LPChannel,
@@ -138,7 +152,10 @@ method readUntil*(s: LPChannel,
                   sep: seq[byte]):
                   Future[int] =
   if s.closed or s.isReset:
-    raise newLPStreamEOFError()
+    var retFuture = newFuture[int]("LPChannel.readUntil")
+    retFuture.fail(newLPStreamEOFError())
+    return retFuture
+
   result = procCall readOnce(BufferStream(s), pbytes, nbytes)
 
 template writePrefix: untyped =
@@ -147,16 +164,14 @@ template writePrefix: untyped =
   if s.closedLocal or s.isReset:
     raise newLPStreamEOFError()
 
-method write*(s: LPChannel,
+method write*(s: LPChannel, pbytes: pointer, nbytes: int) {.async.} =
-              pbytes: pointer,
-              nbytes: int): Future[void] {.async.} =
   writePrefix()
-  result = procCall write(BufferStream(s), pbytes, nbytes)
+  await procCall write(BufferStream(s), pbytes, nbytes)
 
 method write*(s: LPChannel, msg: string, msglen = -1) {.async.} =
   writePrefix()
-  result = procCall write(BufferStream(s), msg, msglen)
+  await procCall write(BufferStream(s), msg, msglen)
 
 method write*(s: LPChannel, msg: seq[byte], msglen = -1) {.async.} =
   writePrefix()
-  result = procCall write(BufferStream(s), msg, msglen)
+  await procCall write(BufferStream(s), msg, msglen)

libp2p/protocols/pubsub/gossipsub.nim

@@ -59,7 +59,7 @@ proc addInterval(every: Duration, cb: CallbackFunc,
     udata: pointer = nil): Future[void] =
   ## Arrange the callback ``cb`` to be called on every ``Duration`` window
 
-  var retFuture = newFuture[void]("chronos.addInterval(Duration)")
+  var retFuture = newFuture[void]("gossipsub.addInterval(Duration)")
   proc interval(arg: pointer = nil) {.gcsafe.}
   proc scheduleNext() =
     if not retFuture.finished():

libp2p/protocols/secure/secure.nim

@@ -14,7 +14,8 @@ import ../protocol,
 type
   Secure* = ref object of LPProtocol # base type for secure managers
 
-method secure*(p: Secure, conn: Connection): Future[Connection]
+method secure*(p: Secure, conn: Connection): Future[Connection] {.base.} =
-  {.base, async, gcsafe.} =
   ## default implementation matches plaintext
-  result = conn
+  var retFuture = newFuture[Connection]("secure.secure")
+  retFuture.complete(conn)
+  return retFuture

libp2p/stream/bufferstream.nim

@@ -271,9 +271,13 @@ method write*(s: BufferStream,
   ##
   ## 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)
-  if not isNil(s.writeHandler):
   result = s.writeHandler(buf)
 
 method write*(s: BufferStream,
@@ -287,9 +291,13 @@ method write*(s: BufferStream,
   ## 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)
-  if not isNil(s.writeHandler):
   result = s.writeHandler(cast[seq[byte]](buf))
 
 method write*(s: BufferStream,
@@ -304,9 +312,13 @@ method write*(s: BufferStream,
   ## If ``msglen > len(sbytes)`` only ``len(sbytes)`` bytes will be written to
   ## stream.
   ##
+  if isNil(s.writeHandler):
+    var retFuture = newFuture[void]("BufferStream.write(seq)")
+    retFuture.fail(newNotWritableError())
+    return retFuture
+
   var buf: seq[byte]
   shallowCopy(buf, if msglen > 0: msg[0..<msglen] else: msg)
-  if not isNil(s.writeHandler):
   result = s.writeHandler(buf)
 
 proc pipe*(s: BufferStream,

libp2p/stream/lpstream.nim

@@ -51,12 +51,11 @@ proc newLPStreamEOFError*(): ref Exception {.inline.} =
 method closed*(s: LPStream): bool {.base, inline.} =
   s.isClosed
 
-method read*(s: LPStream, n = -1): Future[seq[byte]]
+method read*(s: LPStream, n = -1): Future[seq[byte]] {.base, async.} =
-  {.base, async.} =
   doAssert(false, "not implemented!")
 
-method readExactly*(s: LPStream, pbytes: pointer, nbytes: int): Future[void]
+method readExactly*(s: LPStream, pbytes: pointer,
-  {.base, async.} =
+                    nbytes: int): Future[void] {.base, async.} =
   doAssert(false, "not implemented!")
 
 method readLine*(s: LPStream, limit = 0, sep = "\r\n"): Future[string]

libp2p/switch.nim

@@ -325,24 +325,31 @@ proc subscribeToPeer(s: Switch, peerInfo: PeerInfo) {.async, gcsafe.} =
       warn "unable to initiate pubsub", exc = exc.msg
       s.dialedPubSubPeers.excl(peerInfo.id)
 
-proc subscribe*(s: Switch, topic: string, handler: TopicHandler): Future[void] {.gcsafe.} =
+proc subscribe*(s: Switch, topic: string,
+                handler: TopicHandler): Future[void] =
   ## subscribe to a pubsub topic
   if s.pubSub.isNone:
-    raise newNoPubSubException()
+    var retFuture = newFuture[void]("Switch.subscribe")
+    retFuture.fail(newNoPubSubException())
+    return retFuture
 
   result = s.pubSub.get().subscribe(topic, handler)
 
-proc unsubscribe*(s: Switch, topics: seq[TopicPair]): Future[void] {.gcsafe.} =
+proc unsubscribe*(s: Switch, topics: seq[TopicPair]): Future[void] =
   ## unsubscribe from topics
   if s.pubSub.isNone:
-    raise newNoPubSubException()
+    var retFuture = newFuture[void]("Switch.unsubscribe")
+    retFuture.fail(newNoPubSubException())
+    return retFuture
 
   result = s.pubSub.get().unsubscribe(topics)
 
-proc publish*(s: Switch, topic: string, data: seq[byte]): Future[void] {.gcsafe.} =
+proc publish*(s: Switch, topic: string, data: seq[byte]): Future[void] =
   # pubslish to pubsub topic
   if s.pubSub.isNone:
-    raise newNoPubSubException()
+    var retFuture = newFuture[void]("Switch.publish")
+    retFuture.fail(newNoPubSubException())
+    return retFuture
 
   result = s.pubSub.get().publish(topic, data)
 

tests/testmplex.nim

@@ -381,7 +381,8 @@ suite "Mplex":
       proc writeHandler(data: seq[byte]) {.async, gcsafe.} = discard
       let chann = newChannel(1, newConnection(newBufferStream(writeHandler)), true)
       await chann.reset()
-      asyncDiscard chann.read()
+      var data = await chann.read()
+      doAssert(len(data) == 1)
 
     expect LPStreamEOFError:
       waitFor(testResetRead())