Report changes in supported protocols back to ConnectionHandler

thomaseizinger · thomaseizinger · commit f2801fb86a4c · 2023-03-24T15:48:47.000+01:00
diff --git a/protocols/dcutr/src/handler/direct.rs b/protocols/dcutr/src/handler/direct.rs
@@ -92,7 +92,8 @@ impl ConnectionHandler for Handler {
             | ConnectionEvent::FullyNegotiatedOutbound(_)
             | ConnectionEvent::DialUpgradeError(_)
             | ConnectionEvent::ListenUpgradeError(_)
-            | ConnectionEvent::AddressChange(_) => {}
+            | ConnectionEvent::AddressChange(_)
+            | ConnectionEvent::ProtocolsChange(_) => {}
         }
     }
 }
diff --git a/protocols/dcutr/src/handler/relayed.rs b/protocols/dcutr/src/handler/relayed.rs
@@ -419,7 +419,7 @@ impl ConnectionHandler for Handler {
             ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {
                 self.on_dial_upgrade_error(dial_upgrade_error)
             }
-            ConnectionEvent::AddressChange(_) => {}
+            ConnectionEvent::AddressChange(_) | ConnectionEvent::ProtocolsChange(_) => {}
         }
     }
 }
diff --git a/protocols/gossipsub/src/handler.rs b/protocols/gossipsub/src/handler.rs
@@ -576,7 +576,9 @@ impl ConnectionHandler for Handler {
                 warn!("Dial upgrade error {:?}", e);
                 self.upgrade_errors.push_back(e);
             }
-            ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) => {}
+            ConnectionEvent::AddressChange(_)
+            | ConnectionEvent::ListenUpgradeError(_)
+            | ConnectionEvent::ProtocolsChange(_) => {}
         }
     }
 }
diff --git a/protocols/identify/src/handler.rs b/protocols/identify/src/handler.rs
@@ -344,7 +344,9 @@ impl ConnectionHandler for Handler {
             ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {
                 self.on_dial_upgrade_error(dial_upgrade_error)
             }
-            ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) => {}
+            ConnectionEvent::AddressChange(_)
+            | ConnectionEvent::ListenUpgradeError(_)
+            | ConnectionEvent::ProtocolsChange(_) => {}
         }
     }
 }
diff --git a/protocols/kad/src/handler.rs b/protocols/kad/src/handler.rs
@@ -786,7 +786,9 @@ where
             ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {
                 self.on_dial_upgrade_error(dial_upgrade_error)
             }
-            ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) => {}
+            ConnectionEvent::AddressChange(_)
+            | ConnectionEvent::ListenUpgradeError(_)
+            | ConnectionEvent::ProtocolsChange(_) => {}
         }
     }
 }
diff --git a/protocols/perf/src/client/handler.rs b/protocols/perf/src/client/handler.rs
@@ -127,7 +127,7 @@ impl ConnectionHandler for Handler {
                     .boxed(),
             ),
 
-            ConnectionEvent::AddressChange(_) => {}
+            ConnectionEvent::AddressChange(_) | ConnectionEvent::ProtocolsChange(_) => {}
             ConnectionEvent::DialUpgradeError(DialUpgradeError {
                 info: Command { id, .. },
                 error,
diff --git a/protocols/perf/src/server/handler.rs b/protocols/perf/src/server/handler.rs
@@ -104,7 +104,7 @@ impl ConnectionHandler for Handler {
             ConnectionEvent::DialUpgradeError(DialUpgradeError { info, .. }) => {
                 void::unreachable(info)
             }
-            ConnectionEvent::AddressChange(_) => {}
+            ConnectionEvent::AddressChange(_) | ConnectionEvent::ProtocolsChange(_) => {}
             ConnectionEvent::ListenUpgradeError(ListenUpgradeError { info: (), error }) => {
                 match error {
                     ConnectionHandlerUpgrErr::Timeout => {}
diff --git a/protocols/ping/src/handler.rs b/protocols/ping/src/handler.rs
@@ -411,7 +411,9 @@ impl ConnectionHandler for Handler {
             ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {
                 self.on_dial_upgrade_error(dial_upgrade_error)
             }
-            ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) => {}
+            ConnectionEvent::AddressChange(_)
+            | ConnectionEvent::ListenUpgradeError(_)
+            | ConnectionEvent::ProtocolsChange(_) => {}
         }
     }
 }
diff --git a/protocols/relay/src/behaviour/handler.rs b/protocols/relay/src/behaviour/handler.rs
@@ -954,7 +954,7 @@ impl ConnectionHandler for Handler {
             ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {
                 self.on_dial_upgrade_error(dial_upgrade_error)
             }
-            ConnectionEvent::AddressChange(_) => {}
+            ConnectionEvent::AddressChange(_) | ConnectionEvent::ProtocolsChange(_) => {}
         }
     }
 }
diff --git a/protocols/relay/src/priv_client/handler.rs b/protocols/relay/src/priv_client/handler.rs
@@ -613,7 +613,7 @@ impl ConnectionHandler for Handler {
             ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {
                 self.on_dial_upgrade_error(dial_upgrade_error)
             }
-            ConnectionEvent::AddressChange(_) => {}
+            ConnectionEvent::AddressChange(_) | ConnectionEvent::ProtocolsChange(_) => {}
         }
     }
 }
diff --git a/protocols/rendezvous/src/substream_handler.rs b/protocols/rendezvous/src/substream_handler.rs
@@ -396,7 +396,8 @@ where
             // TODO: Handle upgrade errors properly
             ConnectionEvent::AddressChange(_)
             | ConnectionEvent::ListenUpgradeError(_)
-            | ConnectionEvent::DialUpgradeError(_) => {}
+            | ConnectionEvent::DialUpgradeError(_)
+            | ConnectionEvent::ProtocolsChange(_) => {}
         }
     }
 
diff --git a/protocols/request-response/src/handler.rs b/protocols/request-response/src/handler.rs
@@ -425,7 +425,7 @@ where
             ConnectionEvent::ListenUpgradeError(listen_upgrade_error) => {
                 self.on_listen_upgrade_error(listen_upgrade_error)
             }
-            ConnectionEvent::AddressChange(_) => {}
+            ConnectionEvent::AddressChange(_) | ConnectionEvent::ProtocolsChange(_) => {}
         }
     }
 }
diff --git a/swarm/src/behaviour/toggle.rs b/swarm/src/behaviour/toggle.rs
@@ -368,6 +368,9 @@ where
             ConnectionEvent::ListenUpgradeError(listen_upgrade_error) => {
                 self.on_listen_upgrade_error(listen_upgrade_error)
             }
+            ConnectionEvent::ProtocolsChange(_) => {
+                todo!()
+            }
         }
     }
 }
diff --git a/swarm/src/connection.rs b/swarm/src/connection.rs
@@ -29,9 +29,9 @@ pub use error::{
 
 use crate::handler::{
     AddressChange, ConnectionEvent, ConnectionHandler, DialUpgradeError, FullyNegotiatedInbound,
-    FullyNegotiatedOutbound, ListenUpgradeError,
+    FullyNegotiatedOutbound, ListenUpgradeError, ProtocolsChange,
 };
-use crate::upgrade::{InboundUpgradeSend, OutboundUpgradeSend, SendWrapper};
+use crate::upgrade::{InboundUpgradeSend, OutboundUpgradeSend, SendWrapper, UpgradeInfoSend};
 use crate::{ConnectionHandlerEvent, ConnectionHandlerUpgrErr, KeepAlive, SubstreamProtocol};
 use futures::stream::FuturesUnordered;
 use futures::FutureExt;
@@ -43,7 +43,7 @@ use libp2p_core::multiaddr::Multiaddr;
 use libp2p_core::muxing::{StreamMuxerBox, StreamMuxerEvent, StreamMuxerExt, SubstreamBox};
 use libp2p_core::upgrade::{InboundUpgradeApply, OutboundUpgradeApply};
 use libp2p_core::Endpoint;
-use libp2p_core::{upgrade, UpgradeError};
+use libp2p_core::{upgrade, ProtocolName as _, UpgradeError};
 use libp2p_identity::PeerId;
 use std::future::Future;
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -145,6 +145,8 @@ where
     requested_substreams: FuturesUnordered<
         SubstreamRequested<THandler::OutboundOpenInfo, THandler::OutboundProtocol>,
     >,
+
+    supported_protocols: Vec<String>,
 }
 
 impl<THandler> fmt::Debug for Connection<THandler>
@@ -182,6 +184,7 @@ where
             substream_upgrade_protocol_override,
             max_negotiating_inbound_streams,
             requested_substreams: Default::default(),
+            supported_protocols: vec![],
         }
     }
 
@@ -211,6 +214,7 @@ where
             shutdown,
             max_negotiating_inbound_streams,
             substream_upgrade_protocol_override,
+            supported_protocols,
         } = self.get_mut();
 
         loop {
@@ -355,6 +359,23 @@ where
                     Poll::Ready(substream) => {
                         let protocol = handler.listen_protocol();
 
+                        let mut new_protocols = protocol
+                            .upgrade()
+                            .protocol_info()
+                            .filter_map(|i| String::from_utf8(i.protocol_name().to_vec()).ok())
+                            .collect::<Vec<_>>();
+
+                        new_protocols.sort();
+
+                        if supported_protocols != &new_protocols {
+                            handler.on_connection_event(ConnectionEvent::ProtocolsChange(
+                                ProtocolsChange {
+                                    protocols: &new_protocols,
+                                },
+                            ));
+                            *supported_protocols = new_protocols;
+                        }
+
                         negotiating_in.push(SubstreamUpgrade::new_inbound(substream, protocol));
 
                         continue; // Go back to the top, handler can potentially make progress again.
@@ -610,9 +631,10 @@ enum Shutdown {
 mod tests {
     use super::*;
     use crate::keep_alive;
+    use futures::future;
     use futures::AsyncRead;
     use futures::AsyncWrite;
-    use libp2p_core::upgrade::DeniedUpgrade;
+    use libp2p_core::upgrade::{DeniedUpgrade, InboundUpgrade, OutboundUpgrade, UpgradeInfo};
     use libp2p_core::StreamMuxer;
     use quickcheck::*;
     use std::sync::{Arc, Weak};
@@ -673,6 +695,33 @@ mod tests {
         ))
     }
 
+    #[test]
+    fn propagates_changes_to_supported_inbound_protocols() {
+        let mut connection = Connection::new(
+            StreamMuxerBox::new(DummyStreamMuxer {
+                counter: Arc::new(()),
+            }),
+            ConfigurableProtocolConnectionHandler::default(),
+            None,
+            2,
+        );
+        connection.handler.active_protocols = vec!["/foo"];
+
+        // DummyStreamMuxer will yield a new stream
+        let _ = Pin::new(&mut connection)
+            .poll(&mut Context::from_waker(futures::task::noop_waker_ref()));
+        assert_eq!(connection.handler.reported_protocols, vec!["/foo"]);
+
+        connection.handler.active_protocols = vec!["/foo", "/bar"];
+        connection.negotiating_in.clear(); // Hack to request more substreams from the muxer.
+
+        // DummyStreamMuxer will yield a new stream
+        let _ = Pin::new(&mut connection)
+            .poll(&mut Context::from_waker(futures::task::noop_waker_ref()));
+
+        assert_eq!(connection.handler.reported_protocols, vec!["/bar", "/foo"])
+    }
+
     struct DummyStreamMuxer {
         counter: Arc<()>,
     }
@@ -790,6 +839,12 @@ mod tests {
         }
     }
 
+    #[derive(Default)]
+    struct ConfigurableProtocolConnectionHandler {
+        active_protocols: Vec<&'static str>,
+        reported_protocols: Vec<String>,
+    }
+
     impl ConnectionHandler for MockConnectionHandler {
         type InEvent = Void;
         type OutEvent = Void;
@@ -826,7 +881,9 @@ mod tests {
                 ConnectionEvent::DialUpgradeError(DialUpgradeError { error, .. }) => {
                     self.error = Some(error)
                 }
-                ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) => {}
+                ConnectionEvent::AddressChange(_)
+                | ConnectionEvent::ListenUpgradeError(_)
+                | ConnectionEvent::ProtocolsChange(_) => {}
             }
         }
 
@@ -860,6 +917,97 @@ mod tests {
             Poll::Pending
         }
     }
+
+    impl ConnectionHandler for ConfigurableProtocolConnectionHandler {
+        type InEvent = Void;
+        type OutEvent = Void;
+        type Error = Void;
+        type InboundProtocol = ManyProtocolsUpgrade;
+        type OutboundProtocol = DeniedUpgrade;
+        type InboundOpenInfo = ();
+        type OutboundOpenInfo = ();
+
+        fn listen_protocol(
+            &self,
+        ) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
+            SubstreamProtocol::new(
+                ManyProtocolsUpgrade {
+                    protocols: self.active_protocols.clone(),
+                },
+                (),
+            )
+        }
+
+        fn on_connection_event(
+            &mut self,
+            event: ConnectionEvent<
+                Self::InboundProtocol,
+                Self::OutboundProtocol,
+                Self::InboundOpenInfo,
+                Self::OutboundOpenInfo,
+            >,
+        ) {
+            if let ConnectionEvent::ProtocolsChange(ProtocolsChange { protocols }) = event {
+                self.reported_protocols = protocols
+                    .to_vec();
+            }
+        }
+
+        fn on_behaviour_event(&mut self, event: Self::InEvent) {
+            void::unreachable(event)
+        }
+
+        fn connection_keep_alive(&self) -> KeepAlive {
+            KeepAlive::Yes
+        }
+
+        fn poll(
+            &mut self,
+            _: &mut Context<'_>,
+        ) -> Poll<
+            ConnectionHandlerEvent<
+                Self::OutboundProtocol,
+                Self::OutboundOpenInfo,
+                Self::OutEvent,
+                Self::Error,
+            >,
+        > {
+            Poll::Pending
+        }
+    }
+
+    struct ManyProtocolsUpgrade {
+        protocols: Vec<&'static str>,
+    }
+
+    impl UpgradeInfo for ManyProtocolsUpgrade {
+        type Info = &'static str;
+        type InfoIter = std::vec::IntoIter<Self::Info>;
+
+        fn protocol_info(&self) -> Self::InfoIter {
+            self.protocols.clone().into_iter()
+        }
+    }
+
+    impl<C> InboundUpgrade<C> for ManyProtocolsUpgrade {
+        type Output = C;
+        type Error = Void;
+        type Future = future::Ready<Result<Self::Output, Self::Error>>;
+
+        fn upgrade_inbound(self, stream: C, _: Self::Info) -> Self::Future {
+            future::ready(Ok(stream))
+        }
+    }
+
+    impl<C> OutboundUpgrade<C> for ManyProtocolsUpgrade {
+        type Output = C;
+        type Error = Void;
+        type Future = future::Ready<Result<Self::Output, Self::Error>>;
+
+        fn upgrade_outbound(self, stream: C, _: Self::Info) -> Self::Future {
+            future::ready(Ok(stream))
+        }
+    }
 }
 
 /// The endpoint roles associated with a pending peer-to-peer connection.
diff --git a/swarm/src/dummy.rs b/swarm/src/dummy.rs
@@ -139,7 +139,9 @@ impl crate::handler::ConnectionHandler for ConnectionHandler {
                     unreachable!("Denied upgrade does not support any protocols")
                 }
             },
-            ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) => {}
+            ConnectionEvent::AddressChange(_)
+            | ConnectionEvent::ListenUpgradeError(_)
+            | ConnectionEvent::ProtocolsChange(_) => {}
         }
     }
 }
diff --git a/swarm/src/handler.rs b/swarm/src/handler.rs
@@ -210,6 +210,8 @@ pub enum ConnectionEvent<'a, IP: InboundUpgradeSend, OP: OutboundUpgradeSend, IO
     DialUpgradeError(DialUpgradeError<OOI, OP>),
     /// Informs the handler that upgrading an inbound substream to the given protocol has failed.
     ListenUpgradeError(ListenUpgradeError<IOI, IP>),
+    /// The [`ConnectionHandler`] now supports a different set of protocols.
+    ProtocolsChange(ProtocolsChange<'a>),
 }
 
 /// [`ConnectionEvent`] variant that informs the handler about
@@ -239,6 +241,12 @@ pub struct AddressChange<'a> {
     pub new_address: &'a Multiaddr,
 }
 
+/// [`ConnectionEvent`] variant that informs the handler about a change in the address of the remote.
+#[derive(Clone, Copy)]
+pub struct ProtocolsChange<'a> {
+    pub protocols: &'a [String],
+}
+
 /// [`ConnectionEvent`] variant that informs the handler
 /// that upgrading an outbound substream to the given protocol has failed.
 pub struct DialUpgradeError<OOI, OP: OutboundUpgradeSend> {
diff --git a/swarm/src/handler/either.rs b/swarm/src/handler/either.rs
diff --git a/swarm/src/handler/multi.rs b/swarm/src/handler/multi.rs
diff --git a/swarm/src/handler/one_shot.rs b/swarm/src/handler/one_shot.rs
diff --git a/swarm/src/handler/pending.rs b/swarm/src/handler/pending.rs
diff --git a/swarm/src/handler/select.rs b/swarm/src/handler/select.rs
diff --git a/swarm/src/keep_alive.rs b/swarm/src/keep_alive.rs

Original file line number	Diff line number	Diff line change
`@@ -92,7 +92,8 @@ impl ConnectionHandler for Handler {`
`92`	`92`	`\| ConnectionEvent::FullyNegotiatedOutbound(_)`
`93`	`93`	`\| ConnectionEvent::DialUpgradeError(_)`
`94`	`94`	`\| ConnectionEvent::ListenUpgradeError(_)`
`95`		`- \| ConnectionEvent::AddressChange(_) => {}`
	`95`	`+ \| ConnectionEvent::AddressChange(_)`
	`96`	`+ \| ConnectionEvent::ProtocolsChange(_) => {}`
`96`	`97`	`}`
`97`	`98`	`}`
`98`	`99`	`}`
Original file line number	Diff line number	Diff line change
`@@ -419,7 +419,7 @@ impl ConnectionHandler for Handler {`
`419`	`419`	`ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {`
`420`	`420`	`self.on_dial_upgrade_error(dial_upgrade_error)`
`421`	`421`	`}`
`422`		`- ConnectionEvent::AddressChange(_) => {}`
	`422`	`+ ConnectionEvent::AddressChange(_) \| ConnectionEvent::ProtocolsChange(_) => {}`
`423`	`423`	`}`
`424`	`424`	`}`
`425`	`425`	`}`
Original file line number	Diff line number	Diff line change
`@@ -576,7 +576,9 @@ impl ConnectionHandler for Handler {`
`576`	`576`	`warn!("Dial upgrade error {:?}", e);`
`577`	`577`	`self.upgrade_errors.push_back(e);`
`578`	`578`	`}`
`579`		`- ConnectionEvent::AddressChange(_) \| ConnectionEvent::ListenUpgradeError(_) => {}`
	`579`	`+ ConnectionEvent::AddressChange(_)`
	`580`	`+ \| ConnectionEvent::ListenUpgradeError(_)`
	`581`	`+ \| ConnectionEvent::ProtocolsChange(_) => {}`
`580`	`582`	`}`
`581`	`583`	`}`
`582`	`584`	`}`
Original file line number	Diff line number	Diff line change
`@@ -344,7 +344,9 @@ impl ConnectionHandler for Handler {`
`344`	`344`	`ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {`
`345`	`345`	`self.on_dial_upgrade_error(dial_upgrade_error)`
`346`	`346`	`}`
`347`		`- ConnectionEvent::AddressChange(_) \| ConnectionEvent::ListenUpgradeError(_) => {}`
	`347`	`+ ConnectionEvent::AddressChange(_)`
	`348`	`+ \| ConnectionEvent::ListenUpgradeError(_)`
	`349`	`+ \| ConnectionEvent::ProtocolsChange(_) => {}`
`348`	`350`	`}`
`349`	`351`	`}`
`350`	`352`	`}`
Original file line number	Diff line number	Diff line change
`@@ -786,7 +786,9 @@ where`
`786`	`786`	`ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {`
`787`	`787`	`self.on_dial_upgrade_error(dial_upgrade_error)`
`788`	`788`	`}`
`789`		`- ConnectionEvent::AddressChange(_) \| ConnectionEvent::ListenUpgradeError(_) => {}`
	`789`	`+ ConnectionEvent::AddressChange(_)`
	`790`	`+ \| ConnectionEvent::ListenUpgradeError(_)`
	`791`	`+ \| ConnectionEvent::ProtocolsChange(_) => {}`
`790`	`792`	`}`
`791`	`793`	`}`
`792`	`794`	`}`
Original file line number	Diff line number	Diff line change
`@@ -104,7 +104,7 @@ impl ConnectionHandler for Handler {`
`104`	`104`	`ConnectionEvent::DialUpgradeError(DialUpgradeError { info, .. }) => {`
`105`	`105`	`void::unreachable(info)`
`106`	`106`	`}`
`107`		`- ConnectionEvent::AddressChange(_) => {}`
	`107`	`+ ConnectionEvent::AddressChange(_) \| ConnectionEvent::ProtocolsChange(_) => {}`
`108`	`108`	`ConnectionEvent::ListenUpgradeError(ListenUpgradeError { info: (), error }) => {`
`109`	`109`	`match error {`
`110`	`110`	`ConnectionHandlerUpgrErr::Timeout => {}`
Original file line number	Diff line number	Diff line change
`@@ -411,7 +411,9 @@ impl ConnectionHandler for Handler {`
`411`	`411`	`ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {`
`412`	`412`	`self.on_dial_upgrade_error(dial_upgrade_error)`
`413`	`413`	`}`
`414`		`- ConnectionEvent::AddressChange(_) \| ConnectionEvent::ListenUpgradeError(_) => {}`
	`414`	`+ ConnectionEvent::AddressChange(_)`
	`415`	`+ \| ConnectionEvent::ListenUpgradeError(_)`
	`416`	`+ \| ConnectionEvent::ProtocolsChange(_) => {}`
`415`	`417`	`}`
`416`	`418`	`}`
`417`	`419`	`}`
Original file line number	Diff line number	Diff line change
`@@ -954,7 +954,7 @@ impl ConnectionHandler for Handler {`
`954`	`954`	`ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {`
`955`	`955`	`self.on_dial_upgrade_error(dial_upgrade_error)`
`956`	`956`	`}`
`957`		`- ConnectionEvent::AddressChange(_) => {}`
	`957`	`+ ConnectionEvent::AddressChange(_) \| ConnectionEvent::ProtocolsChange(_) => {}`
`958`	`958`	`}`
`959`	`959`	`}`
`960`	`960`	`}`
Original file line number	Diff line number	Diff line change
`@@ -613,7 +613,7 @@ impl ConnectionHandler for Handler {`
`613`	`613`	`ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {`
`614`	`614`	`self.on_dial_upgrade_error(dial_upgrade_error)`
`615`	`615`	`}`
`616`		`- ConnectionEvent::AddressChange(_) => {}`
	`616`	`+ ConnectionEvent::AddressChange(_) \| ConnectionEvent::ProtocolsChange(_) => {}`
`617`	`617`	`}`
`618`	`618`	`}`
`619`	`619`	`}`