feat(peerEvents): add a peerEvent Identified (#843)

Co-authored-by: diegomrsantos <diego@status.im>

libp2p/connmanager.nim

@@ -48,10 +48,11 @@ type
   PeerEventKind* {.pure.} = enum
     Left
     Joined
+    Identified
 
   PeerEvent* = object
     case kind*: PeerEventKind
-    of PeerEventKind.Joined:
+    of PeerEventKind.Joined, PeerEventKind.Identified:
       initiator*: bool
     else:
       discard

libp2p/dialer.nim

@@ -195,6 +195,10 @@ proc internalConnect(
     try:
       self.connManager.storeMuxer(muxed)
       await self.peerStore.identify(muxed)
+      await self.connManager.triggerPeerEvents(
+        muxed.connection.peerId,
+        PeerEvent(kind: PeerEventKind.Identified, initiator: true),
+      )
     except CatchableError as exc:
       trace "Failed to finish outgoung upgrade", err = exc.msg
       await muxed.close()

libp2p/services/autorelayservice.nim

@@ -66,8 +66,8 @@ method setup*(self: AutoRelayService, switch: Switch): Future[bool] {.async.} =
 
   let hasBeenSetUp = await procCall Service(self).setup(switch)
   if hasBeenSetUp:
-    proc handlePeerJoined(peerId: PeerId, event: PeerEvent) {.async.} =
-      trace "Peer Joined", peerId
+    proc handlePeerIdentified(peerId: PeerId, event: PeerEvent) {.async.} =
+      trace "Peer Identified", peerId
       if self.relayPeers.len < self.maxNumRelays:
         self.peerAvailable.fire()
 
@@ -76,7 +76,7 @@ method setup*(self: AutoRelayService, switch: Switch): Future[bool] {.async.} =
       self.relayPeers.withValue(peerId, future):
         future[].cancel()
 
-    switch.addPeerEventHandler(handlePeerJoined, Joined)
+    switch.addPeerEventHandler(handlePeerIdentified, Identified)
     switch.addPeerEventHandler(handlePeerLeft, Left)
     switch.peerInfo.addressMappers.add(self.addressMapper)
     await self.run(switch)
@@ -119,7 +119,6 @@ proc innerRun(self: AutoRelayService, switch: Switch) {.async.} =
       await one(toSeq(self.relayPeers.values())) or self.peerAvailable.wait()
     else:
       await self.peerAvailable.wait()
-      await sleepAsync(200.millis)
 
 method run*(self: AutoRelayService, switch: Switch) {.async.} =
   if self.running:

libp2p/switch.nim

@@ -204,6 +204,9 @@ proc upgrader(switch: Switch, trans: Transport, conn: Connection) {.async.} =
   let muxed = await trans.upgrade(conn, Opt.none(PeerId))
   switch.connManager.storeMuxer(muxed)
   await switch.peerStore.identify(muxed)
+  await switch.connManager.triggerPeerEvents(
+    muxed.connection.peerId, PeerEvent(kind: PeerEventKind.Identified, initiator: false)
+  )
   trace "Connection upgrade succeeded"
 
 proc upgradeMonitor(
@@ -295,9 +298,6 @@ proc stop*(s: Switch) {.async, public.} =
     except CatchableError as exc:
       warn "error cleaning up transports", msg = exc.msg
 
-  for service in s.services:
-    discard await service.stop(s)
-
   await s.ms.stop()
 
   trace "Switch stopped"