WIP connection manager

README.md

@@ -105,7 +105,7 @@ use laminar::{Socket, Packet};
 
 // create the socket
 let mut socket = Socket::bind("127.0.0.1:12345")?;
-let packet_sender = socket.get_packet_sender();
+let packet_sender = socket.get_event_sender();
 // this will start the socket, which will start a poll mechanism to receive and send messages.
 let _thread = thread::spawn(move || socket.start_polling());
 

examples/server_client.rs

@@ -5,41 +5,48 @@ use std::io::stdin;
 use std::thread;
 use std::time::Instant;
 
-use laminar::{ErrorKind, Packet, Socket, SocketEvent};
+use laminar::{
+    managers::SimpleSocketManager, ConnectionEvent, ErrorKind, Packet, ReceiveEvent, Socket,
+    SocketEventSender,
+};
 
 const SERVER: &str = "127.0.0.1:12351";
 
 fn server() -> Result<(), ErrorKind> {
-    let mut socket = Socket::bind(SERVER)?;
-    let (sender, receiver) = (socket.get_packet_sender(), socket.get_event_receiver());
+    // create socket manager, that will use SimpleConnectionManager, that actually initiates connection by exchanging methods
+    let mut socket = Socket::bind(SERVER, Box::new(SimpleSocketManager(false)))?;
+    let (sender, receiver) = (
+        SocketEventSender(socket.get_event_sender()),
+        socket.get_event_receiver(),
+    );
     let _thread = thread::spawn(move || socket.start_polling());
 
     loop {
-        if let Ok(event) = receiver.recv() {
+        if let Ok(ConnectionEvent(addr, event)) = receiver.recv() {
             match event {
-                SocketEvent::Packet(packet) => {
+                ReceiveEvent::Connected(data) => {
+                    println!(
+                        "{:?} -> Connected msg:{}",
+                        addr,
+                        String::from_utf8_lossy(data.as_ref())
+                    );
+                }
+                ReceiveEvent::Packet(packet) => {
                     let msg = packet.payload();
-
                     if msg == b"Bye!" {
                         break;
                     }
-
                     let msg = String::from_utf8_lossy(msg);
-                    let ip = packet.addr().ip();
-
-                    println!("Received {:?} from {:?}", msg, ip);
+                    println!("{:?} -> Packet msg:{}", addr, msg);
 
                     sender
                         .send(Packet::reliable_unordered(
                             packet.addr(),
-                            "Copy that!".as_bytes().to_vec(),
+                            ["Echo: ".as_bytes(), msg.as_bytes()].concat(),
                         ))
                         .expect("This should send");
                 }
-                SocketEvent::Timeout(address) => {
-                    println!("Client timed out: {}", address);
-                }
-                _ => {}
+                _ => println!("{:?} -> {:?}", addr, event),
             }
         }
     }
@@ -49,43 +56,58 @@ fn server() -> Result<(), ErrorKind> {
 
 fn client() -> Result<(), ErrorKind> {
     let addr = "127.0.0.1:12352";
-    let mut socket = Socket::bind(addr)?;
+    let mut socket = Socket::bind(addr, Box::new(SimpleSocketManager(false)))?;
     println!("Connected on {}", addr);
 
-    let server = SERVER.parse().unwrap();
+    let sender = SocketEventSender(socket.get_event_sender());
+    let _thread = thread::spawn(move || loop {
+        socket.manual_poll(Instant::now());
 
-    println!("Type a message and press Enter to send. Send `Bye!` to quit.");
+        if let Some(ConnectionEvent(addr, event)) = socket.recv() {
+            match event {
+                ReceiveEvent::Connected(data) => {
+                    println!(
+                        "{:?} -> Connected msg:{}",
+                        addr,
+                        String::from_utf8_lossy(data.as_ref())
+                    );
+                }
+                ReceiveEvent::Packet(packet) => {
+                    let msg = String::from_utf8_lossy(packet.payload());
+                    println!("{:?} -> Packet msg:{}", addr, msg);
+                }
+                _ => println!("{:?} -> {:?}", addr, event),
+            }
+        }
+    });
 
     let stdin = stdin();
     let mut s_buffer = String::new();
+    s_buffer.clear();
+
+    let server = SERVER.parse().unwrap();
+    println!("Type a `:c<message>` to connect");
+    println!("Type a `<message>` to send a packet");
+    println!("Type a `:d` to disconnect");
+    println!("Type a `:q` to quit.");
 
     loop {
-        s_buffer.clear();
         stdin.read_line(&mut s_buffer)?;
         let line = s_buffer.replace(|x| x == '\n' || x == '\r', "");
-
-        socket.send(Packet::reliable_unordered(
-            server,
-            line.clone().into_bytes(),
-        ))?;
-
-        socket.manual_poll(Instant::now());
-
-        if line == "Bye!" {
+        if line == ":q" {
             break;
+        } else if line.starts_with(":c") {
+            sender
+                .connect(server, Box::from(line.split_at(2).1.as_bytes()))
+                .expect("sending should not fail");
+        } else if line == ":d" {
+            sender.disconnect(server).expect("sending should not fail");
+        } else {
+            sender
+                .send(Packet::reliable_unordered(server, line.into_bytes()))
+                .expect("sending should not fail");
         }
-
-        match socket.recv() {
-            Some(SocketEvent::Packet(packet)) => {
-                if packet.addr() == server {
-                    println!("Server sent: {}", String::from_utf8_lossy(packet.payload()));
-                } else {
-                    println!("Unknown sender.");
-                }
-            }
-            Some(SocketEvent::Timeout(_)) => {}
-            _ => println!("Silence.."),
-        }
+        s_buffer.clear();
     }
 
     Ok(())

examples/simple_udp.rs

@@ -3,7 +3,7 @@
 //! 2. setting up client to send data.
 //! 3. serialize data to send and deserialize when received.
 use bincode::{deserialize, serialize};
-use laminar::{Packet, Socket, SocketEvent};
+use laminar::{ConnectionEventSender, Packet, Socket};
 use serde_derive::{Deserialize, Serialize};
 use std::net::SocketAddr;
 use std::time::Instant;
@@ -27,7 +27,7 @@ pub fn main() {
     let mut server = Socket::bind(server_address()).unwrap();
 
     /*  setup or `Client` and send some test data. */
-    let mut client = Socket::bind(client_address()).unwrap();
+    let mut client = ConnectionEventSender(Socket::bind(client_address()).unwrap());
 
     client.send(Packet::unreliable(
         server_address(),

src/either.rs

@@ -1,4 +1,5 @@
-pub(crate) enum Either<L, R> {
+#[derive(Debug)]
+pub enum Either<L, R> {
     Left(L),
     Right(R),
 }

src/error.rs

@@ -1,6 +1,8 @@
 //! This module contains the laminar error handling logic.
 
-use crate::SocketEvent;
+use crate::either::Either;
+use crate::net::events::{ConnectionEvent, ReceiveEvent, SendEvent};
+use crate::net::managers::ConnectionManagerError;
 use crossbeam_channel::SendError;
 use std::{
     error::Error,
@@ -27,9 +29,11 @@ pub enum ErrorKind {
     /// Protocol versions did not match
     ProtocolVersionMismatch,
     /// Could not send on `SendChannel`.
-    SendError(SendError<SocketEvent>),
+    SendError(SendError<Either<ConnectionEvent<SendEvent>, ConnectionEvent<ReceiveEvent>>>),
     /// Expected header but could not be read from buffer.
     CouldNotReadHeader(String),
+    /// Errors that is returned from `ConnectionManager` either preprocessing data or processing packet
+    ConnectionError(ConnectionManagerError),
 }
 
 impl Display for ErrorKind {
@@ -67,6 +71,11 @@ impl Display for ErrorKind {
                 "Expected {} header but could not be read from buffer.",
                 header
             ),
+            ErrorKind::ConnectionError(err) => write!(
+                fmt,
+                "Something went wrong in ConnectionManager. Reason: {:?}.",
+                err
+            ),
         }
     }
 }
@@ -103,6 +112,8 @@ impl Display for DecodingErrorKind {
 pub enum PacketErrorKind {
     /// The maximal allowed size of the packet was exceeded
     ExceededMaxPacketSize,
+    /// Only user packets (a.k.a PacketType::Packet) can be fragmented
+    PacketTypeCannotBeFragmented,
 }
 
 impl Display for PacketErrorKind {
@@ -111,6 +122,10 @@ impl Display for PacketErrorKind {
             PacketErrorKind::ExceededMaxPacketSize => {
                 write!(fmt, "The packet size was bigger than the max allowed size.")
             }
+            PacketErrorKind::PacketTypeCannotBeFragmented => write!(
+                fmt,
+                "Only user packets (PacketType::Packet) can be fragmented."
+            ),
         }
     }
 }
@@ -173,9 +188,21 @@ impl From<FragmentErrorKind> for ErrorKind {
     }
 }
 
-impl From<crossbeam_channel::SendError<SocketEvent>> for ErrorKind {
-    fn from(inner: SendError<SocketEvent>) -> Self {
-        ErrorKind::SendError(inner)
+impl From<SendError<ConnectionEvent<SendEvent>>> for ErrorKind {
+    fn from(inner: SendError<ConnectionEvent<SendEvent>>) -> Self {
+        ErrorKind::SendError(SendError(Either::Left(inner.0)))
+    }
+}
+
+impl From<SendError<ConnectionEvent<ReceiveEvent>>> for ErrorKind {
+    fn from(inner: SendError<ConnectionEvent<ReceiveEvent>>) -> Self {
+        ErrorKind::SendError(SendError(Either::Right(inner.0)))
+    }
+}
+
+impl From<ConnectionManagerError> for ErrorKind {
+    fn from(inner: ConnectionManagerError) -> Self {
+        ErrorKind::ConnectionError(inner)
     }
 }
 

src/infrastructure/acknowledgment.rs

@@ -1,5 +1,4 @@
-use crate::packet::OrderingGuarantee;
-use crate::packet::SequenceNumber;
+use crate::packet::{OrderingGuarantee, PacketType, SequenceNumber};
 use crate::sequence_buffer::{sequence_greater_than, sequence_less_than, SequenceBuffer};
 use std::collections::HashMap;
 
@@ -101,13 +100,15 @@ impl AcknowledgmentHandler {
     /// Enqueue the outgoing packet for acknowledgment.
     pub fn process_outgoing(
         &mut self,
+        packet_type: PacketType,
         payload: &[u8],
         ordering_guarantee: OrderingGuarantee,
         item_identifier: Option<SequenceNumber>,
     ) {
         self.sent_packets.insert(
             self.sequence_number,
             SentPacket {
+                packet_type,
                 payload: Box::from(payload),
                 ordering_guarantee,
                 item_identifier,
@@ -138,8 +139,9 @@ impl AcknowledgmentHandler {
     }
 }
 
-#[derive(Clone, Debug, Default, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq)]
 pub struct SentPacket {
+    pub packet_type: PacketType,
     pub payload: Box<[u8]>,
     pub ordering_guarantee: OrderingGuarantee,
     pub item_identifier: Option<SequenceNumber>,

src/infrastructure/fragmenter.rs

@@ -71,7 +71,7 @@ impl Fragmentation {
             Fragmentation::fragments_needed(payload_length, config.fragment_size) as u8;
 
         if num_fragments > config.max_fragments {
-            Err(FragmentErrorKind::ExceededMaxFragments)?;
+            return Err(FragmentErrorKind::ExceededMaxFragments.into());
         }
 
         for fragment_id in 0..num_fragments {
@@ -112,16 +112,16 @@ impl Fragmentation {
             // get entry of previous received fragments
             let reassembly_data = match self.fragments.get_mut(fragment_header.sequence()) {
                 Some(val) => val,
-                None => Err(FragmentErrorKind::CouldNotFindFragmentById)?,
+                None => return Err(FragmentErrorKind::CouldNotFindFragmentById.into()),
             };
 
             // Got the data
             if reassembly_data.num_fragments_total != fragment_header.fragment_count() {
-                Err(FragmentErrorKind::FragmentWithUnevenNumberOfFragemts)?
+                return Err(FragmentErrorKind::FragmentWithUnevenNumberOfFragemts.into());
             }
 
             if reassembly_data.fragments_received[usize::from(fragment_header.id())] {
-                Err(FragmentErrorKind::AlreadyProcessedFragment)?
+                return Err(FragmentErrorKind::AlreadyProcessedFragment.into());
             }
 
             // increase number of received fragments and set the specific fragment to received.

src/lib.rs

@@ -26,6 +26,7 @@ mod config;
 mod either;
 mod error;
 mod infrastructure;
+pub mod managers;
 mod net;
 mod packet;
 mod protocol_version;
@@ -39,5 +40,7 @@ pub use self::throughput::ThroughputMonitoring;
 
 pub use self::config::Config;
 pub use self::error::{ErrorKind, Result};
-pub use self::net::{LinkConditioner, Socket, SocketEvent};
+pub use self::net::events::*;
+pub use self::net::managers::{ConnectionManager, ConnectionManagerError};
+pub use self::net::{LinkConditioner, Socket, SocketEventSender};
 pub use self::packet::{DeliveryGuarantee, OrderingGuarantee, Packet};

src/managers.rs

@@ -0,0 +1,4 @@
+//! This module provides socket managers.
+mod simple;
+
+pub use self::simple::SimpleSocketManager;