Merge #1226

1226: embassy-net: Implement flush for TcpSocket r=Dirbaio a=kbleeke

Implements flush for TcpSocket by checking the send queue. 

Flushing is implemented by checking if smoltcp's send_queue/tx_buffer is empty. The flush is completed when all outstanding octets are acknowledged. Smoltcp wakes the send waker [here](https://docs.rs/smoltcp/latest/src/smoltcp/socket/tcp.rs.html#1712) when ACKs are processed and data is removed from the send buffer. So we can re-check in our flush implementation, if the buffer is now empty.

fixes #1223



Co-authored-by: kbleeke <[email protected]>

1 files changed, 20 insertions, 3 deletions

diff --git a/embassy-net/src/tcp.rs b/embassy-net/src/tcp.rs
index d46bd4dbf..c3d8764b0 100644
--- a/embassy-net/src/tcp.rs
+++ b/embassy-net/src/tcp.rs
@@ -63,6 +63,10 @@ impl<'a> TcpWriter<'a> {
     pub async fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
         self.io.write(buf).await
     }
+
+    pub async fn flush(&mut self) -> Result<(), Error> {
+        self.io.flush().await
+    }
 }
 
 impl<'a> TcpSocket<'a> {
@@ -146,6 +150,10 @@ impl<'a> TcpSocket<'a> {
         self.io.write(buf).await
     }
 
+    pub async fn flush(&mut self) -> Result<(), Error> {
+        self.io.flush().await
+    }
+
     pub fn set_timeout(&mut self, duration: Option<Duration>) {
         self.io.with_mut(|s, _| s.set_timeout(duration))
     }
@@ -254,10 +262,19 @@ impl<'d> TcpIo<'d> {
         .await
     }
 
-    #[allow(unused)]
     async fn flush(&mut self) -> Result<(), Error> {
-        poll_fn(move |_| {
-            Poll::Ready(Ok(())) // TODO: Is there a better implementation for this?
+        poll_fn(move |cx| {
+            self.with_mut(|s, _| {
+                // If there are outstanding send operations, register for wake up and wait
+                // smoltcp issues wake-ups when octets are dequeued from the send buffer
+                if s.send_queue() > 0 {
+                    s.register_send_waker(cx.waker());
+                    Poll::Pending
+                // No outstanding sends, socket is flushed
+                } else {
+                    Poll::Ready(Ok(()))
+                }
+            })
         })
         .await
     }