fix compilation issues and tests

libp2p/multiaddress.nim

@@ -408,7 +408,10 @@ const
   UDP_IP* = mapAnd(IP, mapEq("udp"))
   UDP* = mapOr(UDP_DNS, UDP_IP)
   UTP* = mapAnd(UDP, mapEq("utp"))
-  QUIC* = mapAnd(UDP, mapEq("quic"))
+  QUIC* = mapEq("quic")
+  QUIC_V1_IP* = mapAnd(UDP_IP, mapEq("quic-v1"))
+  QUIC_V1_DNS* = mapAnd(UDP_DNS, mapEq("quic-v1"))
+  QUIC_V1* = mapOr(QUIC_V1_DNS, QUIC_V1_IP)
   UNIX* = mapEq("unix")
   WS_DNS* = mapAnd(TCP_DNS, mapEq("ws"))
   WS_IP* = mapAnd(TCP_IP, mapEq("ws"))

libp2p/transports/quictransport.nim

@@ -28,12 +28,13 @@ type
   QuicConnection = quic.Connection
 
 # Stream
-type
-  QuicStream* = ref object of P2PConnection
-    stream: Stream
-    cached: seq[byte]
+type QuicStream* = ref object of P2PConnection
+  stream: Stream
+  cached: seq[byte]
 
-proc new(_: type QuicStream, stream: Stream, oaddr: Opt[MultiAddress], peerId: PeerId): QuicStream =
+proc new(
+    _: type QuicStream, stream: Stream, oaddr: Opt[MultiAddress], peerId: PeerId
+): QuicStream =
   let quicstream = QuicStream(stream: stream, observedAddr: oaddr, peerId: peerId)
   procCall P2PConnection(quicstream).initStream()
   quicstream
@@ -43,53 +44,73 @@ template mapExceptions(body: untyped) =
     body
   except QuicError:
     raise newLPStreamEOFError()
+  except CatchableError:
+    raise newLPStreamEOFError()
 
-method readOnce*(stream: QuicStream,
-                 pbytes: pointer,
-                 nbytes: int): Future[int] {.async.} =
-  if stream.cached.len == 0:
-    stream.cached = await mapExceptions(stream.stream.read())
-  result = min(nbytes, stream.cached.len)
-  copyMem(pbytes, addr stream.cached[0], result)
-  stream.cached = stream.cached[result..^1]
+method readOnce*(
+    stream: QuicStream, pbytes: pointer, nbytes: int
+): Future[int] {.async: (raises: [CancelledError, LPStreamError]).} =
+  try:
+    if stream.cached.len == 0:
+      stream.cached = await stream.stream.read()
+    result = min(nbytes, stream.cached.len)
+    copyMem(pbytes, addr stream.cached[0], result)
+    stream.cached = stream.cached[result ..^ 1]
+  except CatchableError as exc:
+    raise newLPStreamEOFError()
 
 {.push warning[LockLevel]: off.}
-method write*(stream: QuicStream, bytes: seq[byte]) {.async.} =
+method write*(
+    stream: QuicStream, bytes: seq[byte]
+) {.async: (raises: [CancelledError, LPStreamError]).} =
   mapExceptions(await stream.stream.write(bytes))
+
 {.pop.}
 
-method closeImpl*(stream: QuicStream) {.async.} =
-  await stream.stream.close()
+method closeImpl*(stream: QuicStream) {.async: (raises: []).} =
+  try:
+    await stream.stream.close()
+  except CatchableError as exc:
+    discard
   await procCall P2PConnection(stream).closeImpl()
 
 # Session
-type
-  QuicSession* = ref object of P2PConnection
-    connection: QuicConnection
+type QuicSession* = ref object of P2PConnection
+  connection: QuicConnection
 
-method close*(session: QuicSession) {.async.} =
+method close*(session: QuicSession) {.async, base.} =
   await session.connection.close()
   await procCall P2PConnection(session).close()
 
-proc getStream*(session: QuicSession,
-                direction = Direction.In): Future[QuicStream] {.async.} =
+proc getStream*(
+    session: QuicSession, direction = Direction.In
+): Future[QuicStream] {.async.} =
   var stream: Stream
-  case direction:
-    of Direction.In:
-      stream = await session.connection.incomingStream()
-    of Direction.Out:
-      stream = await session.connection.openStream()
-      await stream.write(@[]) # QUIC streams do not exist until data is sent
+  case direction
+  of Direction.In:
+    stream = await session.connection.incomingStream()
+  of Direction.Out:
+    stream = await session.connection.openStream()
+    await stream.write(@[]) # QUIC streams do not exist until data is sent
   return QuicStream.new(stream, session.observedAddr, session.peerId)
 
-# Muxer
-type
-  QuicMuxer = ref object of Muxer
-    quicSession: QuicSession
-    handleFut: Future[void]
+method getWrapped*(self: QuicSession): P2PConnection =
+  nil
 
-method newStream*(m: QuicMuxer, name: string = "", lazy: bool = false): Future[P2PConnection] {.async, gcsafe.} =
-  return await m.quicSession.getStream(Direction.Out)
+# Muxer
+type QuicMuxer = ref object of Muxer
+  quicSession: QuicSession
+  handleFut: Future[void]
+
+method newStream*(
+    m: QuicMuxer, name: string = "", lazy: bool = false
+): Future[P2PConnection] {.
+    async: (raises: [CancelledError, LPStreamError, MuxerError])
+.} =
+  try:
+    return await m.quicSession.getStream(Direction.Out)
+  except CatchableError as exc:
+    raise newException(MuxerError, exc.msg, exc)
 
 proc handleStream(m: QuicMuxer, chann: QuicStream) {.async.} =
   ## call the muxer stream handler for this channel
@@ -102,112 +123,44 @@ proc handleStream(m: QuicMuxer, chann: QuicStream) {.async.} =
     trace "Exception in mplex stream handler", msg = exc.msg
     await chann.close()
 
-
-method handle*(m: QuicMuxer): Future[void] {.async, gcsafe.} =
-  while not m.quicSession.atEof:
-    let incomingStream = await m.quicSession.getStream(Direction.In)
-    asyncSpawn m.handleStream(incomingStream)
-
-method close*(m: QuicMuxer) {.async, gcsafe.} =
-  await m.quicSession.close()
-  m.handleFut.cancel()
-
-# Upgrader
-type
-  QuicUpgrade = ref object of Upgrade
-
-proc identify(
-  self: QuicUpgrade,
-  conn: QuicSession
-  ) {.async, gcsafe.} =
-  # new stream for identify
-  let muxer = QuicMuxer(quicSession: conn, connection: conn)
-  muxer.streamHandler = proc(conn: P2PConnection) {.async, gcsafe.} =
-    trace "Starting stream handler"
-    try:
-      await self.ms.handle(conn) # handle incoming connection
-    except CancelledError as exc:
-      raise exc
-    except CatchableError as exc:
-      trace "exception in stream handler", conn, msg = exc.msg
-    finally:
-      await conn.closeWithEOF()
-    trace "Stream handler done", conn
-
-  self.connManager.storeConn(conn)
-  # store it in muxed connections if we have a peer for it
-  muxer.handleFut = muxer.handle()
-  self.connManager.storeMuxer(muxer, muxer.handleFut)
-
-  var stream = await conn.getStream(Direction.Out)
-  if stream == nil:
-    return
-
+method handle*(m: QuicMuxer): Future[void] {.async: (raises: []).} =
   try:
-    await self.identify(stream)
-  finally:
-    await stream.closeWithEOF()
-
-method upgradeIncoming*(
-  self: QuicUpgrade,
-  conn: P2PConnection): Future[void] {.async.} =
-  let qs = QuicSession(conn)
-  #TODO home made shortcut to get the Peer's id
-  # in the future, Quic encryption should be used
-  # instead
-  let stream = await qs.getStream(Direction.Out)
-  await stream.writeLp(self.identity.peerInfo.peerId.getBytes())
-  assert qs.peerId.init(await stream.readLp(1024))
-  await stream.close()
+    while not m.quicSession.atEof:
+      let incomingStream = await m.quicSession.getStream(Direction.In)
+      asyncSpawn m.handleStream(incomingStream)
+  except CatchableError as exc:
+    trace "Exception in mplex handler", msg = exc.msg
 
+method close*(m: QuicMuxer) {.async: (raises: []).} =
   try:
-    await self.identify(qs)
+    await m.quicSession.close()
+    m.handleFut.cancel()
   except CatchableError as exc:
-    info "Failed to upgrade incoming connection", msg=exc.msg
-
-method upgradeOutgoing*(
-  self: QuicUpgrade,
-  conn: P2PConnection): Future[P2PConnection] {.async.} =
-  let qs = QuicSession(conn)
-  #TODO home made shortcut to get the Peer's id
-  let stream = await qs.getStream(Direction.In)
-  await stream.writeLp(self.identity.peerInfo.peerId.getBytes())
-  assert qs.peerId.init(await stream.readLp(1024))
-  await stream.close()
-
-  await self.identify(qs)
-  return conn
+    discard
 
 # Transport
-type
-  QuicTransport* = ref object of Transport
-    listener: Listener
-    connections: seq[P2PConnection]
+type QuicUpgrade = ref object of Upgrade
+
+type QuicTransport* = ref object of Transport
+  listener: Listener
+  connections: seq[P2PConnection]
 
 func new*(_: type QuicTransport, u: Upgrade): QuicTransport =
-  QuicTransport(
-    upgrader: QuicUpgrade(
-                ms: u.ms,
-                identity: u.identity,
-                connManager: u.connManager
-                )
-  )
+  QuicTransport(upgrader: QuicUpgrade(ms: u.ms))
 
 method handles*(transport: QuicTransport, address: MultiAddress): bool =
   if not procCall Transport(transport).handles(address):
     return false
-  QUIC.match(address)
+  QUIC_V1.match(address)
 
 method start*(transport: QuicTransport, addrs: seq[MultiAddress]) {.async.} =
   doAssert transport.listener.isNil, "start() already called"
   #TODO handle multiple addr
   transport.listener = listen(initTAddress(addrs[0]).tryGet)
   await procCall Transport(transport).start(addrs)
   transport.addrs[0] =
-    MultiAddress.init(
-      transport.listener.localAddress(),
-      IPPROTO_UDP
-    ).tryGet() & MultiAddress.init("/quic").get()
+    MultiAddress.init(transport.listener.localAddress(), IPPROTO_UDP).tryGet() &
+    MultiAddress.init("/quic-v1").get()
   transport.running = true
 
 method stop*(transport: QuicTransport) {.async.} =
@@ -220,12 +173,13 @@ method stop*(transport: QuicTransport) {.async.} =
     transport.listener = nil
 
 proc wrapConnection(
-    transport: QuicTransport,
-    connection: QuicConnection): P2PConnection {.raises: [Defect, TransportOsError, LPError].} =
+    transport: QuicTransport, connection: QuicConnection
+): P2PConnection {.raises: [Defect, TransportOsError, LPError].} =
   let
     remoteAddr = connection.remoteAddress()
     observedAddr =
-      MultiAddress.init(remoteAddr, IPPROTO_UDP).get() & MultiAddress.init("/quic").get()
+      MultiAddress.init(remoteAddr, IPPROTO_UDP).get() &
+      MultiAddress.init("/quic-v1").get()
     conres = QuicSession(connection: connection, observedAddr: Opt.some(observedAddr))
   conres.initStream()
 
@@ -234,6 +188,7 @@ proc wrapConnection(
     await conres.join()
     transport.connections.keepItIf(it != conres)
     trace "Cleaned up client"
+
   asyncSpawn onClose()
   return conres
 
@@ -242,8 +197,33 @@ method accept*(transport: QuicTransport): Future[P2PConnection] {.async.} =
   let connection = await transport.listener.accept()
   return transport.wrapConnection(connection)
 
-method dial*(transport: QuicTransport,
-             hostname: string,
-             address: MultiAddress): Future[P2PConnection] {.async.} =
+method dial*(
+    transport: QuicTransport,
+    hostname: string,
+    address: MultiAddress,
+    peerId: Opt[PeerId] = Opt.none(PeerId),
+): Future[P2PConnection] {.async, gcsafe.} =
   let connection = await dial(initTAddress(address).tryGet)
   return transport.wrapConnection(connection)
+
+method upgrade*(
+    self: QuicTransport, conn: P2PConnection, peerId: Opt[PeerId]
+): Future[Muxer] {.async: (raises: [CancelledError, LPError]).} =
+  let qs = QuicSession(conn)
+  if peerId.isSome:
+    qs.peerId = peerId.get()
+
+  let muxer = QuicMuxer(quicSession: qs, connection: conn)
+  muxer.streamHandler = proc(conn: P2PConnection) {.async: (raises: []).} =
+    trace "Starting stream handler"
+    try:
+      await self.upgrader.ms.handle(conn) # handle incoming connection
+    except CancelledError as exc:
+      return
+    except CatchableError as exc:
+      trace "exception in stream handler", conn, msg = exc.msg
+    finally:
+      await conn.closeWithEOF()
+    trace "Stream handler done", conn
+  muxer.handleFut = muxer.handle()
+  return muxer

libp2p/wire.nim

@@ -20,19 +20,15 @@ when defined(windows): import winlean else: import posix
 const
   RTRANSPMA* = mapOr(TCP, WebSockets, UNIX)
 
-  TRANSPMA* = mapOr(
-    RTRANSPMA,
-    QUIC,
-    UDP
-  )
+  TRANSPMA* = mapOr(RTRANSPMA, QUIC, UDP)
 
 proc initTAddress*(ma: MultiAddress): MaResult[TransportAddress] =
   ## Initialize ``TransportAddress`` with MultiAddress ``ma``.
   ##
   ## MultiAddress must be wire address, e.g. ``{IP4, IP6, UNIX}/{TCP, UDP}``.
   ##
 
-  if mapOr(TCP_IP, WebSockets_IP, UNIX, UDP_IP).match(ma):
+  if mapOr(TCP_IP, WebSockets_IP, UNIX, UDP_IP, QUIC_V1_IP).match(ma):
     var pbuf: array[2, byte]
     let code = (?(?ma[0]).protoCode())
     if code == multiCodec("unix"):

tests/testquic.nim

@@ -2,19 +2,22 @@
 
 import sequtils
 import chronos, stew/byteutils
-import ../libp2p/[stream/connection,
-                  transports/transport,
-                  transports/quictransport,
-                  upgrademngrs/upgrade,
-                  multiaddress,
-                  errors,
-                  wire]
+import
+  ../libp2p/[
+    stream/connection,
+    transports/transport,
+    transports/quictransport,
+    upgrademngrs/upgrade,
+    multiaddress,
+    errors,
+    wire,
+  ]
 
 import ./helpers, ./commontransport
 
 suite "Quic transport":
   asyncTest "can handle local address":
-    let ma = @[MultiAddress.init("/ip4/127.0.0.1/udp/0/quic").tryGet()]
+    let ma = @[MultiAddress.init("/ip4/127.0.0.1/udp/0/quic-v1").tryGet()]
     let transport1 = QuicTransport.new()
     await transport1.start(ma)
     check transport1.handles(transport1.addrs[0])