Merge #752

752: Replace embassy::io with embedded_io. r=Dirbaio a=Dirbaio

TODO:

- [x] Release embedded-io on crates.io
- [x] Remove git dep

Co-authored-by: Dario Nieuwenhuis <[email protected]>

4 files changed, 135 insertions, 81 deletions

diff --git a/embassy-net/src/device.rs b/embassy-net/src/device.rs
index f66ebc193..1f4fa5208 100644
--- a/embassy-net/src/device.rs
+++ b/embassy-net/src/device.rs
@@ -4,7 +4,6 @@ use smoltcp::phy::DeviceCapabilities;
 use smoltcp::time::Instant as SmolInstant;
 
 use crate::packet_pool::PacketBoxExt;
-use crate::Result;
 use crate::{Packet, PacketBox, PacketBuf};
 
 #[derive(PartialEq, Eq, Clone, Copy)]
@@ -78,9 +77,9 @@ pub struct RxToken {
 }
 
 impl smoltcp::phy::RxToken for RxToken {
-    fn consume<R, F>(mut self, _timestamp: SmolInstant, f: F) -> Result<R>
+    fn consume<R, F>(mut self, _timestamp: SmolInstant, f: F) -> smoltcp::Result<R>
     where
-        F: FnOnce(&mut [u8]) -> Result<R>,
+        F: FnOnce(&mut [u8]) -> smoltcp::Result<R>,
     {
         f(&mut self.pkt)
     }
@@ -92,9 +91,9 @@ pub struct TxToken<'a> {
 }
 
 impl<'a> smoltcp::phy::TxToken for TxToken<'a> {
-    fn consume<R, F>(self, _timestamp: SmolInstant, len: usize, f: F) -> Result<R>
+    fn consume<R, F>(self, _timestamp: SmolInstant, len: usize, f: F) -> smoltcp::Result<R>
     where
-        F: FnOnce(&mut [u8]) -> Result<R>,
+        F: FnOnce(&mut [u8]) -> smoltcp::Result<R>,
     {
         let mut buf = self.pkt.slice(0..len);
         let r = f(&mut buf)?;
diff --git a/embassy-net/src/lib.rs b/embassy-net/src/lib.rs
index ffe786b36..ded841909 100644
--- a/embassy-net/src/lib.rs
+++ b/embassy-net/src/lib.rs
@@ -1,5 +1,9 @@
 #![cfg_attr(not(feature = "std"), no_std)]
 #![allow(clippy::new_without_default)]
+#![cfg_attr(
+    feature = "nightly",
+    feature(generic_associated_types, type_alias_impl_trait)
+)]
 
 // This mod MUST go first, so that the others see its macros.
 pub(crate) mod fmt;
@@ -20,9 +24,7 @@ pub use stack::{
 };
 
 #[cfg(feature = "tcp")]
-mod tcp_socket;
-#[cfg(feature = "tcp")]
-pub use tcp_socket::TcpSocket;
+pub mod tcp;
 
 // smoltcp reexports
 pub use smoltcp::phy::{DeviceCapabilities, Medium};
@@ -32,4 +34,3 @@ pub use smoltcp::time::Instant as SmolInstant;
 pub use smoltcp::wire::{EthernetAddress, HardwareAddress};
 pub use smoltcp::wire::{IpAddress, IpCidr, Ipv4Address, Ipv4Cidr};
 pub type Interface = smoltcp::iface::Interface<'static, device::DeviceAdapter>;
-pub use smoltcp::{Error, Result};
diff --git a/embassy-net/src/tcp/io_impl.rs b/embassy-net/src/tcp/io_impl.rs
new file mode 100644
index 000000000..155733497
--- /dev/null
+++ b/embassy-net/src/tcp/io_impl.rs
@@ -0,0 +1,67 @@
+use core::future::Future;
+use core::task::Poll;
+use futures::future::poll_fn;
+
+use super::{Error, TcpSocket};
+
+impl<'d> embedded_io::asynch::Read for TcpSocket<'d> {
+    type ReadFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
+    where
+        Self: 'a;
+
+    fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Self::ReadFuture<'a> {
+        poll_fn(move |cx| {
+            // CAUTION: smoltcp semantics around EOF are different to what you'd expect
+            // from posix-like IO, so we have to tweak things here.
+            self.with(|s, _| match s.recv_slice(buf) {
+                // No data ready
+                Ok(0) => {
+                    s.register_recv_waker(cx.waker());
+                    Poll::Pending
+                }
+                // Data ready!
+                Ok(n) => Poll::Ready(Ok(n)),
+                // EOF
+                Err(smoltcp::Error::Finished) => Poll::Ready(Ok(0)),
+                // Connection reset. TODO: this can also be timeouts etc, investigate.
+                Err(smoltcp::Error::Illegal) => Poll::Ready(Err(Error::ConnectionReset)),
+                // smoltcp returns no errors other than the above.
+                Err(_) => unreachable!(),
+            })
+        })
+    }
+}
+
+impl<'d> embedded_io::asynch::Write for TcpSocket<'d> {
+    type WriteFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
+    where
+        Self: 'a;
+
+    fn write<'a>(&'a mut self, buf: &'a [u8]) -> Self::WriteFuture<'a> {
+        poll_fn(move |cx| {
+            self.with(|s, _| match s.send_slice(buf) {
+                // Not ready to send (no space in the tx buffer)
+                Ok(0) => {
+                    s.register_send_waker(cx.waker());
+                    Poll::Pending
+                }
+                // Some data sent
+                Ok(n) => Poll::Ready(Ok(n)),
+                // Connection reset. TODO: this can also be timeouts etc, investigate.
+                Err(smoltcp::Error::Illegal) => Poll::Ready(Err(Error::ConnectionReset)),
+                // smoltcp returns no errors other than the above.
+                Err(_) => unreachable!(),
+            })
+        })
+    }
+
+    type FlushFuture<'a> = impl Future<Output = Result<(), Self::Error>>
+    where
+        Self: 'a;
+
+    fn flush<'a>(&'a mut self) -> Self::FlushFuture<'a> {
+        poll_fn(move |_| {
+            Poll::Ready(Ok(())) // TODO: Is there a better implementation for this?
+        })
+    }
+}
diff --git a/embassy-net/src/tcp_socket.rs b/embassy-net/src/tcp/mod.rs
index 5637505d4..3bfd4c7b6 100644
--- a/embassy-net/src/tcp_socket.rs
+++ b/embassy-net/src/tcp/mod.rs
@@ -1,17 +1,46 @@
 use core::marker::PhantomData;
 use core::mem;
-use core::pin::Pin;
-use core::task::{Context, Poll};
-use embassy::io;
-use embassy::io::{AsyncBufRead, AsyncWrite};
+use core::task::Poll;
 use smoltcp::iface::{Context as SmolContext, SocketHandle};
 use smoltcp::socket::TcpSocket as SyncTcpSocket;
 use smoltcp::socket::{TcpSocketBuffer, TcpState};
 use smoltcp::time::Duration;
 use smoltcp::wire::IpEndpoint;
 
+#[cfg(feature = "nightly")]
+mod io_impl;
+
 use super::stack::Stack;
-use crate::{Error, Result};
+
+#[derive(PartialEq, Eq, Clone, Copy, Debug)]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
+pub enum Error {
+    ConnectionReset,
+}
+
+#[derive(PartialEq, Eq, Clone, Copy, Debug)]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
+pub enum ConnectError {
+    /// The socket is already connected or listening.
+    InvalidState,
+    /// The remote host rejected the connection with a RST packet.
+    ConnectionReset,
+    /// Connect timed out.
+    TimedOut,
+    /// No route to host.
+    NoRoute,
+}
+
+#[derive(PartialEq, Eq, Clone, Copy, Debug)]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
+pub enum AcceptError {
+    /// The socket is already connected or listening.
+    InvalidState,
+    /// Invalid listen port
+    InvalidPort,
+    /// The remote host rejected the connection with a RST packet.
+    ConnectionReset,
+}
 
 pub struct TcpSocket<'a> {
     handle: SocketHandle,
@@ -37,17 +66,25 @@ impl<'a> TcpSocket<'a> {
         }
     }
 
-    pub async fn connect<T>(&mut self, remote_endpoint: T) -> Result<()>
+    pub async fn connect<T>(&mut self, remote_endpoint: T) -> Result<(), ConnectError>
     where
         T: Into<IpEndpoint>,
     {
         let local_port = Stack::with(|stack| stack.get_local_port());
-        self.with(|s, cx| s.connect(cx, remote_endpoint, local_port))?;
+        match self.with(|s, cx| s.connect(cx, remote_endpoint, local_port)) {
+            Ok(()) => {}
+            Err(smoltcp::Error::Illegal) => return Err(ConnectError::InvalidState),
+            Err(smoltcp::Error::Unaddressable) => return Err(ConnectError::NoRoute),
+            // smoltcp returns no errors other than the above.
+            Err(_) => unreachable!(),
+        }
 
         futures::future::poll_fn(|cx| {
             self.with(|s, _| match s.state() {
-                TcpState::Closed | TcpState::TimeWait => Poll::Ready(Err(Error::Unaddressable)),
-                TcpState::Listen => Poll::Ready(Err(Error::Illegal)),
+                TcpState::Closed | TcpState::TimeWait => {
+                    Poll::Ready(Err(ConnectError::ConnectionReset))
+                }
+                TcpState::Listen => unreachable!(),
                 TcpState::SynSent | TcpState::SynReceived => {
                     s.register_send_waker(cx.waker());
                     Poll::Pending
@@ -58,11 +95,17 @@ impl<'a> TcpSocket<'a> {
         .await
     }
 
-    pub async fn accept<T>(&mut self, local_endpoint: T) -> Result<()>
+    pub async fn accept<T>(&mut self, local_endpoint: T) -> Result<(), AcceptError>
     where
         T: Into<IpEndpoint>,
     {
-        self.with(|s, _| s.listen(local_endpoint))?;
+        match self.with(|s, _| s.listen(local_endpoint)) {
+            Ok(()) => {}
+            Err(smoltcp::Error::Illegal) => return Err(AcceptError::InvalidState),
+            Err(smoltcp::Error::Unaddressable) => return Err(AcceptError::InvalidPort),
+            // smoltcp returns no errors other than the above.
+            Err(_) => unreachable!(),
+        }
 
         futures::future::poll_fn(|cx| {
             self.with(|s, _| match s.state() {
@@ -130,11 +173,6 @@ impl<'a> TcpSocket<'a> {
     }
 }
 
-fn to_ioerr(_err: Error) -> io::Error {
-    // todo
-    io::Error::Other
-}
-
 impl<'a> Drop for TcpSocket<'a> {
     fn drop(&mut self) {
         Stack::with(|stack| {
@@ -143,63 +181,12 @@ impl<'a> Drop for TcpSocket<'a> {
     }
 }
 
-impl<'a> AsyncBufRead for TcpSocket<'a> {
-    fn poll_fill_buf<'z>(
-        self: Pin<&'z mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<io::Result<&'z [u8]>> {
-        self.with(|s, _| match s.peek(1 << 30) {
-            // No data ready
-            Ok(buf) if buf.is_empty() => {
-                s.register_recv_waker(cx.waker());
-                Poll::Pending
-            }
-            // Data ready!
-            Ok(buf) => {
-                // Safety:
-                // - User can't touch the inner TcpSocket directly at all.
-                // - The socket itself won't touch these bytes until consume() is called, which
-                //   requires the user to release this borrow.
-                let buf: &'z [u8] = unsafe { core::mem::transmute(&*buf) };
-                Poll::Ready(Ok(buf))
-            }
-            // EOF
-            Err(Error::Finished) => Poll::Ready(Ok(&[][..])),
-            // Error
-            Err(e) => Poll::Ready(Err(to_ioerr(e))),
-        })
-    }
-
-    fn consume(self: Pin<&mut Self>, amt: usize) {
-        if amt == 0 {
-            // smoltcp's recv returns Finished if we're at EOF,
-            // even if we're "reading" 0 bytes.
-            return;
-        }
-        self.with(|s, _| s.recv(|_| (amt, ()))).unwrap()
+impl embedded_io::Error for Error {
+    fn kind(&self) -> embedded_io::ErrorKind {
+        embedded_io::ErrorKind::Other
     }
 }
 
-impl<'a> AsyncWrite for TcpSocket<'a> {
-    fn poll_write(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &[u8],
-    ) -> Poll<io::Result<usize>> {
-        self.with(|s, _| match s.send_slice(buf) {
-            // Not ready to send (no space in the tx buffer)
-            Ok(0) => {
-                s.register_send_waker(cx.waker());
-                Poll::Pending
-            }
-            // Some data sent
-            Ok(n) => Poll::Ready(Ok(n)),
-            // Error
-            Err(e) => Poll::Ready(Err(to_ioerr(e))),
-        })
-    }
-
-    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        Poll::Ready(Ok(())) // TODO: Is there a better implementation for this?
-    }
+impl<'d> embedded_io::Io for TcpSocket<'d> {
+    type Error = Error;
 }