Merge branch 'master' into gcsafe-callback-compat

libp2p/muxers/mplex/mplex.nim

@@ -247,7 +247,7 @@ method close*(m: Mplex) {.async: (raises: []).} =
 
   trace "Closed mplex", m
 
-method getStreams*(m: Mplex): seq[Connection] =
+method getStreams*(m: Mplex): seq[Connection] {.gcsafe.} =
   for c in m.channels[false].values:
     result.add(c)
   for c in m.channels[true].values:

libp2p/muxers/muxer.nim

@@ -67,5 +67,5 @@ proc new*(
   let muxerProvider = T(newMuxer: creator, codec: codec)
   muxerProvider
 
-method getStreams*(m: Muxer): seq[Connection] {.base.} =
+method getStreams*(m: Muxer): seq[Connection] {.base, gcsafe.} =
   raiseAssert("Not implemented!")

libp2p/muxers/yamux/yamux.nim

@@ -205,6 +205,7 @@ proc remoteClosed(channel: YamuxChannel) {.async: (raises: []).} =
   if not channel.closedRemotely.isSet():
     channel.closedRemotely.fire()
     await channel.actuallyClose()
+  channel.isClosedRemotely = true
 
 method closeImpl*(channel: YamuxChannel) {.async: (raises: []).} =
   if not channel.closedLocally:
@@ -632,7 +633,7 @@ method handle*(m: Yamux) {.async: (raises: []).} =
     await m.close()
   trace "Stopped yamux handler"
 
-method getStreams*(m: Yamux): seq[Connection] =
+method getStreams*(m: Yamux): seq[Connection] {.gcsafe.} =
   for c in m.channels.values:
     result.add(c)
 

libp2p/protocols/pubsub/floodsub.nim

@@ -194,7 +194,7 @@ method init*(f: FloodSub) =
 
 method publish*(
     f: FloodSub, topic: string, data: seq[byte]
-): Future[int] {.async: (raises: [LPError]).} =
+): Future[int] {.async: (raises: []).} =
   # base returns always 0
   discard await procCall PubSub(f).publish(topic, data)
 

libp2p/protocols/pubsub/gossipsub.nim

@@ -705,7 +705,7 @@ method onTopicSubscription*(g: GossipSub, topic: string, subscribed: bool) =
 
 method publish*(
     g: GossipSub, topic: string, data: seq[byte]
-): Future[int] {.async: (raises: [LPError]).} =
+): Future[int] {.async: (raises: []).} =
   logScope:
     topic
 

libp2p/protocols/pubsub/pubsub.nim

@@ -559,7 +559,7 @@ proc subscribe*(p: PubSub, topic: string, handler: TopicHandler) {.public.} =
 
 method publish*(
     p: PubSub, topic: string, data: seq[byte]
-): Future[int] {.base, async: (raises: [LPError]), public.} =
+): Future[int] {.base, async: (raises: []), public.} =
   ## publish to a ``topic``
   ##
   ## The return value is the number of neighbours that we attempted to send the

libp2p/protocols/pubsub/rpc/message.nim

@@ -65,7 +65,10 @@ proc init*(
     topic: string,
     seqno: Option[uint64],
     sign: bool = true,
-): Message {.gcsafe, raises: [LPError].} =
+): Message {.gcsafe, raises: [].} =
+  if sign and peer.isNone():
+    doAssert(false, "Cannot sign message without peer info")
+
   var msg = Message(data: data, topic: topic)
 
   # order matters, we want to include seqno in the signature
@@ -81,9 +84,6 @@ proc init*(
         .expect("Invalid private key!")
         .getBytes()
         .expect("Couldn't get public key bytes!")
-  else:
-    if sign:
-      raise (ref LPError)(msg: "Cannot sign message without peer info")
 
   msg
 

libp2p/stream/lpstream.nim

@@ -43,6 +43,7 @@ type
     oid*: Oid
     dir*: Direction
     closedWithEOF: bool # prevent concurrent calls
+    isClosedRemotely*: bool
 
   LPStreamError* = object of LPError
   LPStreamIncompleteError* = object of LPStreamError