Rename API and to read 0 test in remote forward test

src/native_mux_impl/session.rs

@@ -67,13 +67,17 @@ impl Session {
         Ok(())
     }
 
-    pub(crate) async fn cancel_port_forward(
+    pub(crate) async fn close_port_forward(
         &self,
         forward_type: crate::ForwardType,
         listen_socket: crate::Socket<'_>,
         connect_socket: crate::Socket<'_>,
     ) -> Result<(), Error> {
-        unimplemented!("Port forwarding cancellation is not implemented yet")
+        Connection::connect(&self.ctl).await?.close_port_forward(
+            forward_type.into(),
+            &listen_socket.into(),
+            &connect_socket.into(),
+        )
     }
 
     async fn close_impl(&self) -> Result<(), Error> {

src/process_impl/session.rs

@@ -139,7 +139,7 @@ impl Session {
         }
     }
 
-    pub(crate) async fn cancel_port_forward(
+    pub(crate) async fn close_port_forward(
         &self,
         forward_type: ForwardType,
         listen_socket: Socket<'_>,

src/session.rs

@@ -487,19 +487,17 @@ impl Session {
         })
     }
 
-    /// Cancel a previously established local/remote port forwarding.
+    /// Close a previously established local/remote port forwarding.
     ///
     /// The same set of arguments should be passed as when the port forwarding was requested.
-    ///
-    /// Currently, cancelling port forwarding is only supported for the process mux impl.
-    pub async fn cancel_port_forward(
+    pub async fn close_port_forward(
         &self,
         forward_type: impl Into<ForwardType>,
         listen_socket: impl Into<Socket<'_>>,
         connect_socket: impl Into<Socket<'_>>,
     ) -> Result<(), Error> {
         delegate!(&self.0, imp, {
-            imp.cancel_port_forward(
+            imp.close_port_forward(
                 forward_type.into(),
                 listen_socket.into(),
                 connect_socket.into(),

tests/openssh.rs

@@ -851,17 +851,18 @@ async fn remote_socket_forward() {
 
         assert_eq!(DATA, &buffer);
 
+        eprintln!("Canceling port forward");
+        session
+            .close_port_forward(ForwardType::Remote, (loopback(), *port), &*unix_socket)
+            .await
+            .unwrap();
+
+        eprintln!("Trying to connect again");
+        assert_eq!(output.read(&mut buffer).await.unwrap(), 0);
+
         drop(output);
         drop(output_listener);
 
-        if !cfg!(feature = "native-mux") {
-            eprintln!("Canceling port forward");
-            session
-                .cancel_port_forward(ForwardType::Remote, (loopback(), *port), &*unix_socket)
-                .await
-                .unwrap();
-        }
-
         eprintln!("Waiting for session to end");
         let output = child.wait_with_output().await.unwrap();
         eprintln!("remote_socket_forward: {:#?}", output);
@@ -910,17 +911,15 @@ async fn local_socket_forward() {
 
         drop(output);
 
-        if !cfg!(feature = "native-mux") {
-            eprintln!("Canceling port forward");
-            session
-                .cancel_port_forward(ForwardType::Local, &*unix_socket, (loopback(), port))
-                .await
-                .unwrap();
+        eprintln!("Closing port forward");
+        session
+            .close_port_forward(ForwardType::Local, &*unix_socket, (loopback(), port))
+            .await
+            .unwrap();
 
-            eprintln!("Trying to connect again");
-            let e = UnixStream::connect(&unix_socket).await.unwrap_err();
-            assert_eq!(e.kind(), io::ErrorKind::ConnectionRefused);
-        }
+        eprintln!("Trying to connect again");
+        let e = UnixStream::connect(&unix_socket).await.unwrap_err();
+        assert_eq!(e.kind(), io::ErrorKind::ConnectionRefused);
 
         eprintln!("Waiting for session to end");
         let output = child.wait_with_output().await.unwrap();