astro/zynq-rs
1
0
Fork 0
forked from M-Labs/zynq-rs
Code Issues Pull Requests Releases Wiki Activity

libasync: doc TcpStream

Browse Source
This commit is contained in:
Astro 2020-04-01 23:34:53 +02:00
parent 75a8889d28
commit 6dde8c3b02
1 changed files with 18 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
libasync/src/smoltcp/tcp_stream.rs
View File

@ -7,7 +7,7 @@ use core::{
pin::Pin, pin::Pin,
task::{Context, Poll}, task::{Context, Poll},
}; };
use alloc::{boxed::Box, vec, vec::Vec}; use alloc::vec::Vec;
use smoltcp::{ use smoltcp::{
socket::{ socket::{
SocketHandle, SocketRef, SocketHandle, SocketRef,
@ -22,7 +22,7 @@ pub struct TcpStream {
handle: SocketHandle, handle: SocketHandle,
} }
/// Wait while polling a stream /// Wait while letting `$f()` poll a stream's socket
macro_rules! poll_stream { macro_rules! poll_stream {
($stream: expr, $output: ty, $f: expr) => (async { ($stream: expr, $output: ty, $f: expr) => (async {
struct Adhoc<'a> { struct Adhoc<'a> {
@ -46,6 +46,11 @@ macro_rules! poll_stream {
} }
impl TcpStream { impl TcpStream {
/// Allocates sockets and its buffers, registers it in the
/// SocketSet.
///
/// Not `pub` as the result can not yet be used. Use `listen()` or
/// `connect()` to obtain a valid TcpStream.
fn new(rx_bufsize: usize, tx_bufsize: usize) -> Self { fn new(rx_bufsize: usize, tx_bufsize: usize) -> Self {
fn uninit_vec<T>(size: usize) -> Vec<T> { fn uninit_vec<T>(size: usize) -> Vec<T> {
let mut result = Vec::with_capacity(size); let mut result = Vec::with_capacity(size);
@ -100,7 +105,10 @@ impl TcpStream {
/// `listen()` with a backlog instead. /// `listen()` with a backlog instead.
pub async fn accept(port: u16, rx_bufsize: usize, tx_bufsize: usize) -> Self { pub async fn accept(port: u16, rx_bufsize: usize, tx_bufsize: usize) -> Self {
let stream = Self::new(rx_bufsize, tx_bufsize); let stream = Self::new(rx_bufsize, tx_bufsize);
stream.with_socket(|mut s| s.listen(port)).expect("listen"); // Set socket to listen
stream.with_socket(|mut s| s.listen(port))
.expect("listen");
// Wait for a connection
poll_stream!(&stream, (), |socket| { poll_stream!(&stream, (), |socket| {
if socket.is_active() { if socket.is_active() {
Poll::Ready(()) Poll::Ready(())
@ -133,8 +141,11 @@ impl TcpStream {
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> { fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let result = self.stream.with_socket(|mut socket| { let result = self.stream.with_socket(|mut socket| {
socket.recv(|buf| match (self.f)(buf) { socket.recv(|buf| match (self.f)(buf) {
Poll::Ready((amount, result)) => (amount, Poll::Ready(Ok(result))), Poll::Ready((amount, result)) =>
Poll::Pending => (0, Poll::Pending), (amount, Poll::Ready(Ok(result))),
Poll::Pending =>
// 0 bytes consumed
(0, Poll::Pending),
}) })
}); });
match result { match result {
@ -211,6 +222,8 @@ impl TcpStream {
} }
impl Drop for TcpStream { impl Drop for TcpStream {
/// Free item in the socket set, which leads to deallocation of
/// the rx/tx buffers associated with this socket.
fn drop(&mut self) { fn drop(&mut self) {
Sockets::instance().sockets.borrow_mut() Sockets::instance().sockets.borrow_mut()
.remove(self.handle); .remove(self.handle);