Add Discv5Event::SessionEstablished to surface ENRs (sigp#123)

* Add 'Discv5Event::SessionEstablished' to surface ENRs

* Replace Enr with NodeContact in SessionEstablished event

* Replace NodeContact with (Enr, SocketAddr) in SessionEstablished

Author: jacobkaufmann

examples/find_nodes.rs

@@ -193,6 +193,7 @@ async fn main() {
                     Discv5Event::Discovered(enr) => info!("Enr discovered {}", enr),
                     Discv5Event::EnrAdded { enr, replaced: _ } => info!("Enr added {}", enr),
                     Discv5Event::NodeInserted { node_id, replaced: _ } => info!("Node inserted {}", node_id),
+                    Discv5Event::SessionEstablished(enr, _) => info!("Session established {}", enr),
                     Discv5Event::SocketUpdated(addr) => info!("Socket updated {}", addr),
                     Discv5Event::TalkRequest(_) => info!("Talk request received"),
                 };

src/discv5.rs

@@ -61,6 +61,8 @@ pub enum Discv5Event {
         node_id: NodeId,
         replaced: Option<NodeId>,
     },
+    /// A new session has been established with a node.
+    SessionEstablished(Enr, SocketAddr),
     /// Our local ENR IP address has been updated.
     SocketUpdated(SocketAddr),
     /// A node has initiated a talk request.

src/handler/mod.rs

@@ -108,8 +108,9 @@ pub enum HandlerOut {
     /// A session has been established with a node.
     ///
     /// A session is only considered established once we have received a signed ENR from the
-    /// node and received messages from it's `SocketAddr` matching it's ENR fields.
-    Established(Enr, ConnectionDirection),
+    /// node and either the observed `SocketAddr` matches the one declared in the ENR or the
+    /// ENR declares no `SocketAddr`.
+    Established(Enr, SocketAddr, ConnectionDirection),
 
     /// A Request has been received from a node on the network.
     Request(NodeAddress, Box<Request>),
@@ -691,7 +692,11 @@ impl Handler {
 
                 // Notify the application that the session has been established
                 self.service_send
-                    .send(HandlerOut::Established(enr, connection_direction))
+                    .send(HandlerOut::Established(
+                        enr,
+                        node_address.socket_addr,
+                        connection_direction,
+                    ))
                     .await
                     .unwrap_or_else(|e| warn!("Error with sending channel: {}", e));
             }
@@ -778,7 +783,11 @@ impl Handler {
                         // This occurs when a node established a connection with us.
                         if let Err(e) = self
                             .service_send
-                            .send(HandlerOut::Established(enr, ConnectionDirection::Incoming))
+                            .send(HandlerOut::Established(
+                                enr,
+                                node_address.socket_addr,
+                                ConnectionDirection::Incoming,
+                            ))
                             .await
                         {
                             warn!("Failed to inform of established session {}", e)
@@ -943,6 +952,7 @@ impl Handler {
                                                 .service_send
                                                 .send(HandlerOut::Established(
                                                     enr,
+                                                    node_address.socket_addr,
                                                     ConnectionDirection::Outgoing,
                                                 ))
                                                 .await

src/handler/tests.rs

@@ -169,7 +169,7 @@ async fn multiple_messages() {
     let sender = async move {
         loop {
             match sender_handler_recv.recv().await {
-                Some(HandlerOut::Established(_, _)) => {
+                Some(HandlerOut::Established(_, _, _)) => {
                     // now the session is established, send the rest of the messages
                     for _ in 0..messages_to_send - 1 {
                         let _ = sender_handler.send(HandlerIn::Request(

src/service.rs

@@ -341,8 +341,9 @@ impl Service {
                 }
                 Some(event) = self.handler_recv.recv() => {
                     match event {
-                        HandlerOut::Established(enr, direction) => {
-                            self.inject_session_established(enr,direction);
+                        HandlerOut::Established(enr, socket_addr, direction) => {
+                            self.send_event(Discv5Event::SessionEstablished(enr.clone(), socket_addr));
+                            self.inject_session_established(enr, direction);
                         }
                         HandlerOut::Request(node_address, request) => {
                                 self.handle_rpc_request(node_address, *request);