Towards reliable communication

src/frame.rs

@@ -1,8 +1,8 @@
-use std::{mem, fmt};
+use std::{mem, fmt, io};
 
-use bytes::{Bytes, IntoBuf, BufMut};
+use bytes::{Bytes, BufMut};
 
-use {varint, FromBytes};
+use {varint, FromBytes, TransportError};
 
 #[derive(Debug, Copy, Clone, Eq, PartialEq)]
 pub struct Type(u8);
@@ -48,38 +48,53 @@ frame_types!{
     PADDING = 0x00,
     RST_STREAM = 0x01,
     CONNECTION_CLOSE = 0x02,
+    STOP_SENDING = 0x0c,
+    ACK = 0x0d,
 }
 
 #[derive(Debug)]
 pub enum Frame {
     Padding,
     RstStream {
-        id: u64,
+        id: StreamId,
         app_error_code: u16,
         final_offset: u64,
     },
     ConnectionClose {
-        error_code: u16,
+        error_code: TransportError,
         reason: Bytes,
     },
-    Stream {
-        id: u64,
-        offset: Option<u64>,
-        fin: bool,
-        data: Bytes,
-    },
+    Ack(Ack),
+    Stream(Stream),
     Invalid,
 }
 
+#[derive(Debug, Clone)]
+pub struct Ack {
+    pub delay: u64,
+    pub largest: u64,
+    pub packets: AckIter
+}
+
+#[derive(Debug, Clone)]
+pub struct Stream {
+    pub id: StreamId,
+    pub offset: u64,
+    pub fin: bool,
+    pub data: Bytes,
+}
+
 pub struct Iter(Bytes);
 
 impl Iter {
     pub fn new(payload: Bytes) -> Self { Iter(payload) }
 
     fn get_var(&mut self) -> Option<u64> {
-        let mut buf = self.0.clone().into_buf();
-        let x: u64 = varint::read(&mut buf)?;
-        self.0.advance(buf.position() as usize);
+        let (x, advance) = {
+            let mut buf = io::Cursor::new(&self.0[..]);
+            (varint::read(&mut buf)?, buf.position())
+        };
+        self.0.advance(advance as usize);
         Some(x)
     }
 
@@ -97,21 +112,29 @@ impl Iter {
         Some(match ty {
             Type::PADDING => Frame::Padding,
             Type::RST_STREAM => Frame::RstStream {
-                id: self.get_var()?,
+                id: self.get_var()?.into(),
                 app_error_code: self.get()?,
                 final_offset: self.get_var()?,
             },
             Type::CONNECTION_CLOSE => Frame::ConnectionClose {
-                error_code: self.get()?,
+                error_code: self.get::<u16>()?.into(),
                 reason: self.take_len()?,
             },
+            Type::ACK => {
+                let largest = self.get_var()?;
+                let delay = self.get_var()?;
+                Frame::Ack(Ack {
+                    delay, largest,
+                    packets: AckIter::new(largest, self)?,
+                })
+            }
             _ => match ty.stream() {
-                Some(s) => Frame::Stream {
-                    id: self.get_var()?,
-                    offset: if s.off() { Some(self.get_var()?) } else { None },
+                Some(s) => Frame::Stream(Stream {
+                    id: self.get_var()?.into(),
+                    offset: if s.off() { self.get_var()? } else { 0 },
                     fin: s.fin(),
                     data: if s.len() { self.take_len()? } else { mem::replace(&mut self.0, Bytes::new()) }
-                },
+                }),
                 None => return None,
             }
         })
@@ -133,14 +156,82 @@ impl Iterator for Iter {
     }
 }
 
-pub fn stream(out: &mut Vec<u8>, id: u64, offset: Option<u64>, length: bool, fin: bool, data: &[u8]) {
+pub fn stream(out: &mut Vec<u8>, id: StreamId, offset: Option<u64>, length: bool, fin: bool, data: &[u8]) {
     let mut ty = 0x10;
     if offset.is_some() { ty |= 0x04; }
     if length { ty |= 0x02; }
     if fin { ty |= 0x01; }
     out.put_u8(ty);
-    varint::write(id, out).unwrap();
+    varint::write(id.0, out).unwrap();
     if let Some(o) = offset { varint::write(o, out).unwrap(); }
     if length { varint::write(data.len() as u64, out).unwrap(); }
     out.extend_from_slice(data);
 }
+
+#[derive(Debug, Clone)]
+pub struct AckIter {
+    next: u64,
+    block_size: u64,
+    data: Bytes,
+}
+
+impl AckIter {
+    fn new(largest: u64, packet: &mut Iter) -> Option<Self> {
+        let extra_blocks = packet.get_var()? + 1;
+        let first_block = packet.get_var()?;
+        let len = {
+            let mut buf = io::Cursor::new(&packet.0[..]);
+            for i in 0..extra_blocks {
+                varint::read(&mut buf)?; // gap
+                varint::read(&mut buf)?; // block
+            }
+            buf.position()
+        };
+        
+        Some(Self {
+            next: largest,
+            block_size: first_block + 1,
+            data: packet.0.slice(0, len as usize),
+        })
+    }
+
+    pub fn peek(&self) -> Option<u64> {
+        if self.block_size == 0 { None } else { Some(self.next) }
+    }
+}
+
+impl Iterator for AckIter {
+    type Item = u64;
+    fn next(&mut self) -> Option<u64> {
+        if self.block_size == 0 { return None; }
+        let result = self.next;
+        self.next -= 1;
+        self.block_size -= 1;
+        if self.block_size == 0 && !self.data.is_empty() {
+            let advance = {
+                let mut buf = io::Cursor::new(&self.data[..]);
+                self.next -= varint::read(&mut buf).unwrap() + 1;
+                self.block_size = varint::read(&mut buf).unwrap() + 1;
+                buf.position()
+            };
+            self.data.advance(advance as usize);
+        }
+        Some(result)
+    }
+}
+
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
+pub struct StreamId(pub u64);
+
+#[derive(Debug, Copy, Clone, Eq, PartialEq)]
+pub enum Side { Client, Server }
+
+#[derive(Debug, Copy, Clone, Eq, PartialEq)]
+pub enum Directionality { Uni, Bi }
+
+impl StreamId {
+    pub fn initiator(&self) -> Side { if self.0 & 0x1 == 0 { Side::Client } else { Side::Server } }
+    pub fn directionality(&self) -> Directionality { if self.0 & 0x2 == 0 { Directionality::Bi } else { Directionality::Uni } }
+}
+
+impl From<u64> for StreamId { fn from(x: u64) -> Self { StreamId(x) } }

src/transport_error.rs

@@ -5,6 +5,8 @@ use frame;
 #[derive(Debug, Copy, Clone, Eq, PartialEq)]
 pub struct Error(u16);
 
+impl From<u16> for Error { fn from(x: u16) -> Self { Error(x) } }
+
 macro_rules! errors {
     {$($name:ident($val:expr) $desc:expr;)*} => {
         impl Error {

src/transport_parameters.rs

@@ -179,7 +179,7 @@ impl TransportParameters {
             }
         }
 
-        if initial_max_stream_data && initial_max_data && idle_timeout && (am_server || params.stateless_reset_token.is_some()) {
+        if initial_max_stream_data && initial_max_data && idle_timeout && (am_server ^ params.stateless_reset_token.is_some()) {
             Ok(params)
         } else {
             Err(Error::IllegalValue)

tests/tests.rs

@@ -6,13 +6,10 @@ extern crate slog;
 extern crate slog_term;
 #[macro_use]
 extern crate assert_matches;
-#[macro_use]
-extern crate lazy_static;
 
 use std::net::SocketAddrV6;
 
-use std::ops::Deref;
-use openssl::pkey::{PKey, PKeyRef, Private};
+use openssl::pkey::{PKey};
 use openssl::rsa::Rsa;
 use openssl::x509::X509;
 use slog::{Logger, Drain};
@@ -25,19 +22,6 @@ fn logger() -> Logger {
     Logger::root(drain, o!())
 }
 
-
-
-// lazy_static! {
-//     static ref PRIVATE_KEY: Rsa<Private> = {
-        
-//     };
-//     static ref CERT: X509 = {
-//         let key = PKey::from_rsa(PRIVATE_KEY.clone()).unwrap();
-        
-//     };
-// }
-
-
 struct Pair {
     log: Logger,
     server: Endpoint,
@@ -79,15 +63,17 @@ impl Pair {
                 None => {}
                 Some(Io::Transmit { destination, packet }) => {
                     trace!(self.log, "server -> client");
-                    self.client.handle(self.server_addr, destination, Vec::from(packet).into());
+                    self.client.handle(0, self.server_addr, destination, Vec::from(packet).into());
                 }
+                Some(Io::TimerStart { .. }) | Some(Io::TimerStop { .. }) => {} // No time passes
             }
             match c {
                 None => {}
                 Some(Io::Transmit { destination, packet }) => {
                     trace!(self.log, "client -> server");
-                    self.server.handle(self.client_addr, destination, Vec::from(packet).into())
+                    self.server.handle(0, self.client_addr, destination, Vec::from(packet).into())
                 }
+                Some(Io::TimerStart { .. }) | Some(Io::TimerStop { .. }) => {} // No time passes
             }
         }
     }
@@ -101,6 +87,6 @@ fn connect() {
         panic!("{}", e);
     }
     pair.drive();
-    assert_matches!(pair.server.poll().unwrap(), Event::Connected(_));
-    assert_matches!(pair.client.poll().unwrap(), Event::Connected(_));
+    assert_matches!(pair.server.poll(), Some(Event::Connected(_)));
+    assert_matches!(pair.client.poll(), Some(Event::Connected(_)));
 }