PollStream/PollDataChannel: flush before shutting down (#340)

* PollStream/PollDataChannel: flush before shutting down

"Invocation of a shutdown implies an invocation of flush. Once this
method returns Ready it implies that a flush successfully happened
before the shutdown happened. That is, callers don’t need to call flush
before calling shutdown. They can rely that by calling shutdown any
pending buffered data will be written out."

https://docs.rs/tokio/1.21.2/tokio/io/trait.AsyncWrite.html#tymethod.poll_shutdown

* format code

* fix clippy warning

* fix another clippy warning

* add changelog entries

Author: melekes

data/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Unreleased
 
 * Increased minimum support rust version to `1.60.0`.
+* `PollDataChannel::poll_shutdown`: make sure to flush any writes before shutting down [#340](https://github.com/webrtc-rs/webrtc/pull/340)
 
 ## 0.5.0
 

data/src/data_channel/mod.rs

@@ -626,6 +626,11 @@ impl AsyncWrite for PollDataChannel {
     }
 
     fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        match self.as_mut().poll_flush(cx) {
+            Poll::Pending => return Poll::Pending,
+            Poll::Ready(_) => {}
+        }
+
         let fut = match self.shutdown_fut.as_mut() {
             Some(fut) => fut,
             None => {

sctp/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Unreleased
 
 * Increased minimum support rust version to `1.60.0`.
+* `PollStream::poll_shutdown`: make sure to flush any writes before shutting down [#340](https://github.com/webrtc-rs/webrtc/pull/340)
 
 ## v0.6.1
 
@@ -12,4 +13,3 @@
 ## Prior to 0.6.1
 
 Before 0.6.1 there was no changelog, previous changes are sometimes, but not always, available in the [GitHub Releases](https://github.com/webrtc-rs/sctp/releases).
-

sctp/src/stream/mod.rs

@@ -4,21 +4,24 @@ mod stream_test;
 use crate::association::AssociationState;
 use crate::chunk::chunk_payload_data::{ChunkPayloadData, PayloadProtocolIdentifier};
 use crate::error::{Error, Result};
-use crate::queue::reassembly_queue::ReassemblyQueue;
-
 use crate::queue::pending_queue::PendingQueue;
+use crate::queue::reassembly_queue::ReassemblyQueue;
 
 use bytes::Bytes;
-use std::fmt;
-use std::future::Future;
-use std::io;
-use std::net::Shutdown;
-use std::pin::Pin;
-use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU32, AtomicU8, AtomicUsize, Ordering};
-use std::sync::Arc;
-use std::task::{Context, Poll};
-use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
-use tokio::sync::{mpsc, Mutex, Notify};
+use std::{
+    fmt,
+    future::Future,
+    io,
+    net::Shutdown,
+    pin::Pin,
+    sync::atomic::{AtomicBool, AtomicU16, AtomicU32, AtomicU8, AtomicUsize, Ordering},
+    sync::Arc,
+    task::{Context, Poll},
+};
+use tokio::{
+    io::{AsyncRead, AsyncWrite, ReadBuf},
+    sync::{mpsc, Mutex, Notify},
+};
 
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 #[repr(C)]
@@ -760,6 +763,11 @@ impl AsyncWrite for PollStream {
     }
 
     fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        match self.as_mut().poll_flush(cx) {
+            Poll::Pending => return Poll::Pending,
+            Poll::Ready(_) => {}
+        }
+
         let fut = match self.shutdown_fut.as_mut() {
             Some(fut) => fut,
             None => {

stun/src/message.rs

@@ -64,7 +64,7 @@ pub struct Message {
 
 impl fmt::Display for Message {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        let t_id = base64::encode(&self.transaction_id.0);
+        let t_id = base64::encode(self.transaction_id.0);
         write!(
             f,
             "{} l={} attrs={} id={}",

util/src/fixed_big_int/mod.rs

@@ -81,11 +81,7 @@ impl FixedBigInt {
         }
         let chunk = i / 64;
         let pos = i % 64;
-        if self.bits[chunk] & (1 << pos) != 0 {
-            1
-        } else {
-            0
-        }
+        usize::from(self.bits[chunk] & (1 << pos) != 0)
     }
 
     // set_bit sets i-th bit to 1.