association: only reset stored_init once ACK is received (#24)

* association: only reset stored_init once ACK is received

BEFORE: after `sent_init` is called, `stored_init` becomes None.
Consequently, all retransmit attempts was failing because there's no
init anymore.

AFTER: after `sent_init` is called, `stored_init` is still Some(v).

Closes #23

Author: melekes

src/association/association_internal.rs

@@ -169,7 +169,7 @@ impl AssociationInternal {
 
     /// caller must hold self.lock
     pub(crate) fn send_init(&mut self) -> Result<()> {
-        if let Some(stored_init) = self.stored_init.take() {
+        if let Some(stored_init) = self.stored_init.clone() {
             log::debug!("[{}] sending INIT", self.name);
 
             self.source_port = 5000; // Spec??

src/chunk/chunk_init.rs

@@ -74,6 +74,20 @@ pub(crate) struct ChunkInit {
     pub(crate) params: Vec<Box<dyn Param + Send + Sync>>,
 }
 
+impl Clone for ChunkInit {
+    fn clone(&self) -> Self {
+        ChunkInit {
+            is_ack: self.is_ack,
+            initiate_tag: self.initiate_tag,
+            advertised_receiver_window_credit: self.advertised_receiver_window_credit,
+            num_outbound_streams: self.num_outbound_streams,
+            num_inbound_streams: self.num_inbound_streams,
+            initial_tsn: self.initial_tsn,
+            params: self.params.to_vec(),
+        }
+    }
+}
+
 pub(crate) const INIT_CHUNK_MIN_LENGTH: usize = 16;
 pub(crate) const INIT_OPTIONAL_VAR_HEADER_LENGTH: usize = 4;
 

src/chunk/chunk_payload_data.rs

@@ -15,7 +15,7 @@ pub(crate) const PAYLOAD_DATA_HEADER_SIZE: usize = 12;
 /// PayloadProtocolIdentifier is an enum for DataChannel payload types
 /// PayloadProtocolIdentifier enums
 /// https://www.iana.org/assignments/sctp-parameters/sctp-parameters.xhtml#sctp-parameters-25
-#[derive(Debug, Copy, Clone, PartialEq)]
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
 #[repr(C)]
 pub enum PayloadProtocolIdentifier {
     Dcep = 50,

src/error.rs

@@ -3,7 +3,7 @@ use thiserror::Error;
 
 pub type Result<T> = std::result::Result<T, Error>;
 
-#[derive(Debug, Error, PartialEq)]
+#[derive(Debug, Error, PartialEq, Eq)]
 #[non_exhaustive]
 pub enum Error {
     #[error("raw is too small for a SCTP chunk")]

src/packet.rs

@@ -170,7 +170,7 @@ impl Packet {
 
         let hasher = Crc::<u32>::new(&CRC_32_ISCSI);
         let mut digest = hasher.digest();
-        digest.update(&writer.to_vec());
+        digest.update(writer);
         digest.update(&FOUR_ZEROES);
         digest.update(&raw[..]);
         let checksum = digest.finalize();

src/param/param_unknown.rs

@@ -10,7 +10,7 @@ use std::fmt::{Debug, Display, Formatter};
 /// This means we do not really understand the semantics of the param but can represent it.
 ///
 /// This is useful for usage in e.g.`ParamUnrecognized` where we want to report some unrecognized params back to the sender.
-#[derive(Clone, Debug, PartialEq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ParamUnknown {
     typ: u16,
     value: Bytes,
@@ -33,7 +33,7 @@ impl Param for ParamUnknown {
     }
 
     fn as_any(&self) -> &(dyn Any + Send + Sync) {
-        &*self
+        self
     }
 
     fn unmarshal(raw: &Bytes) -> crate::error::Result<Self>

src/param/param_unrecognized.rs

@@ -35,7 +35,7 @@ impl Param for ParamUnrecognized {
     }
 
     fn as_any(&self) -> &(dyn Any + Send + Sync) {
-        &*self
+        self
     }
 
     fn unmarshal(raw: &Bytes) -> crate::error::Result<Self>

src/stream/mod.rs

@@ -20,7 +20,7 @@ use std::task::{Context, Poll};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::sync::{mpsc, Mutex, Notify};
 
-#[derive(Debug, Copy, Clone, PartialEq)]
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
 #[repr(C)]
 pub enum ReliabilityType {
     /// ReliabilityTypeReliable is used for reliable transmission
@@ -795,6 +795,6 @@ impl fmt::Debug for PollStream {
 
 impl AsRef<Stream> for PollStream {
     fn as_ref(&self) -> &Stream {
-        &*self.stream
+        &self.stream
     }
 }

src/timer/timer_test.rs

@@ -1,13 +1,9 @@
 use async_trait::async_trait;
-use std::sync::atomic::{AtomicU32, Ordering};
-use std::sync::Arc;
-use std::time::SystemTime;
-
-use tokio::sync::mpsc;
 use tokio::sync::Mutex;
 use tokio::time::{sleep, Duration};
 
-use crate::error::Result;
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::Arc;
 
 ///////////////////////////////////////////////////////////////////
 //ack_timer_test
@@ -165,6 +161,10 @@ mod test_rto_manager {
 mod test_rtx_timer {
     use super::*;
     use crate::association::RtxTimerId;
+    use crate::error::Result;
+
+    use std::time::SystemTime;
+    use tokio::sync::mpsc;
 
     struct TestTimerObserver {
         ncbs: Arc<AtomicU32>,