Add tests for Channel read/write.

proto/src/core/channel.rs

@@ -174,6 +174,50 @@ mod tests {
         let _ = env_logger::builder().is_test(true).try_init();
     }
 
+    #[tokio::test]
+    async fn read_write() {
+        init();
+
+        let listener = TcpListener::bind("localhost:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+
+        let listener_task = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.unwrap();
+            let mut channel = Channel::<u32, u32>::new(stream);
+
+            assert_eq!(channel.read().await.unwrap(), 1);
+            channel.write(&2).await.unwrap();
+        });
+
+        let stream = TcpStream::connect(address).await.unwrap();
+        let mut channel = Channel::<u32, u32>::new(stream);
+
+        channel.write(&1).await.unwrap();
+        assert_eq!(channel.read().await.unwrap(), 2);
+
+        listener_task.await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn read_eof() {
+        init();
+
+        let listener = TcpListener::bind("localhost:0").await.unwrap();
+        let address = listener.local_addr().unwrap();
+
+        let listener_task = tokio::spawn(async move {
+            // Accept the stream and immediately drop/close it.
+            listener.accept().await.unwrap();
+        });
+
+        let stream = TcpStream::connect(address).await.unwrap();
+        let mut channel = Channel::<u32, u32>::new(stream);
+
+        assert!(channel.read().await.unwrap_err().is_unexpected_eof());
+
+        listener_task.await.unwrap();
+    }
+
     #[tokio::test]
     async fn open_close() {
         init();