net: fix a bug in channel and clean up

src/net/acceptor.rs

@@ -128,9 +128,10 @@ impl Acceptor {
     /// Run the accept loop in a new thread and error if a connection problem
     /// occurs.
     fn accept(self: Arc<Self>, listener: Box<dyn TransportListener>, executor: Arc<Executor<'_>>) {
+        let self2 = self.clone();
         self.task.clone().start(
             self.clone().run_accept_loop(listener),
-            |result| self.handle_stop(result),
+            |result| self2.handle_stop(result),
             Error::ServiceStopped,
             executor,
         );

src/net/channel.rs

@@ -102,7 +102,6 @@ impl Channel {
         let self2 = self.clone();
         self.receive_task.clone().start(
             self.clone().main_receive_loop(),
-            // Ignore stop handler
             |result| self2.handle_stop(result),
             Error::ServiceStopped,
             executor,
@@ -263,7 +262,7 @@ impl Channel {
                 Ok(packet) => packet,
                 Err(err) => {
                     if Self::is_eof_error(err.clone()) {
-                        info!("Channel {:?} disconnected", self.address());
+                        info!("Inbound connection {} disconnected", self.address());
                     } else {
                         error!("Read error on channel: {}", err);
                     }

src/net/message.rs

@@ -1,7 +1,7 @@
 use std::io;
 
 use futures::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
-use log::{debug, error};
+use log::debug;
 use url::Url;
 
 use crate::{
@@ -170,10 +170,7 @@ pub async fn read_packet<R: AsyncRead + Unpin + Sized>(stream: &mut R) -> Result
     let mut magic = [0u8; 4];
     debug!(target: "net", "reading magic...");
 
-    if let Err(err) = stream.read_exact(&mut magic).await {
-        error!("Failed to read the magic: {}", err);
-        return Err(Error::ConnectFailed)
-    }
+    stream.read_exact(&mut magic).await?;
 
     debug!(target: "net", "read magic {:?}", magic);
     if magic != MAGIC_BYTES {