Generate replies for received message packets

Signed-off-by: Lee Smet <lee.smet@hotmail.com>

src/message.rs

@@ -22,6 +22,7 @@ use crate::{
     crypto::PacketBuffer,
     data::DataPlane,
     message::{chunk::MessageChunk, done::MessageDone, init::MessageInit},
+    packet::{DataPacket, Packet},
 };
 
 mod chunk;
@@ -283,7 +284,7 @@ impl MessageStack {
         let header = mp.header();
         let message_id = header.message_id();
         let flags = header.flags();
-        if flags.init() {
+        let reply = if flags.init() {
             // We receive a new message with an ID. If we already have a complete message, ignore
             // it.
             let mut inbox = self.inbox.lock().unwrap();
@@ -308,6 +309,8 @@ impl MessageStack {
             if inbox.pending_msges.insert(message_id, message).is_some() {
                 debug!("Dropped current pending message because we received a new message with INIT flag set for the same ID");
             }
+
+            Some(mi.into_reply().into_inner())
         } else if flags.chunk() {
             // A chunk can only be received for incomplete messages. We don't have to check the
             // completed messages. Either there is none, so no problem, or there is one, in which
@@ -348,6 +351,10 @@ impl MessageStack {
                     chunk_idx: mc.chunk_idx(),
                     data: mc.data().to_vec(),
                 });
+
+                Some(mc.into_reply().into_inner())
+            } else {
+                None
             }
         } else if flags.done() {
             let mut inbox = self.inbox.lock().unwrap();
@@ -397,6 +404,10 @@ impl MessageStack {
                 // Move message to be read.
                 inbox.complete_msges.insert(message_id, message);
                 inbox.pending_msges.remove(&message_id);
+
+                Some(md.into_reply().into_inner())
+            } else {
+                None
             }
         } else if flags.read() {
             let mut outbox = self.outbox.lock().unwrap();
@@ -407,6 +418,7 @@ impl MessageStack {
                 }
                 message.state = TransmissionState::Read;
             }
+            None
         } else if flags.aborted() {
             // If the message is not finished yet, discard it completely.
             // But if it is finished, ignore this, i.e, nothing to do.
@@ -414,10 +426,25 @@ impl MessageStack {
             if inbox.pending_msges.remove(&message_id).is_some() {
                 debug!("Dropping pending message because we received an ABORT");
             }
+            None
         } else {
-            debug!("Received unknown ACK message flags {:x}", flags.flags);
+            debug!("Received unknown message flags {:x}", flags.flags);
+            None
+        };
+        if let Some(reply) = reply {
+            // This is a reply, so SRC -> DST and DST -> SRC
+            // FIXME: this can be fixed once the dataplane accepts generic IpAddr addresses.
+            match (src, dst) {
+                (IpAddr::V6(src), IpAddr::V6(dst)) => {
+                    self.data_plane.lock().unwrap().inject_message_packet(
+                        dst,
+                        src,
+                        reply.into_inner(),
+                    );
+                }
+                _ => debug!("can only reply to message fragments if both src and dst are IPv6"),
+            }
         }
-        todo!("reply");
     }
 }
 
@@ -511,6 +538,11 @@ impl MessagePacket {
             .expect("Packet contains enough data for a header; qed"),
         }
     }
+
+    /// Consumes this `MessagePacket`, returning the underlying [`PacketBuffer`].
+    pub fn into_inner(self) -> PacketBuffer {
+        self.packet
+    }
 }
 
 /// A MessagePacketMut exposes the mutable useable body of a `PacketBuffer` for working with messages.

src/message/chunk.rs

@@ -92,6 +92,15 @@ impl MessageChunk {
         Ok(())
     }
 
+    /// Convert the `MessageChunk` into a reply. This does nothing if it is already a reply.
+    pub fn into_reply(mut self) -> Self {
+        self.buffer.header_mut().flags_mut().set_ack();
+        // We want to leave the length field in tact but don't want to copy the data in the reply.
+        // This needs additional work on the underlying buffer.
+        // TODO
+        self
+    }
+
     /// Consumes this `MessageChunk`, returning the underlying [`MessagePacket`].
     pub fn into_inner(self) -> MessagePacket {
         self.buffer

src/message/done.rs

@@ -45,6 +45,12 @@ impl MessageDone {
             .copy_from_slice(checksum.as_bytes())
     }
 
+    /// Convert the `MessageDone` into a reply. This does nothing if it is already a reply.
+    pub fn into_reply(mut self) -> Self {
+        self.buffer.header_mut().flags_mut().set_ack();
+        self
+    }
+
     /// Consumes this `MessageDone`, returning the underlying [`MessagePacket`].
     pub fn into_inner(self) -> MessagePacket {
         self.buffer

src/message/init.rs

@@ -29,6 +29,12 @@ impl MessageInit {
         self.buffer.buffer_mut()[..8].copy_from_slice(&length.to_be_bytes())
     }
 
+    /// Convert the `MessageInit` into a reply. This does nothing if it is already a reply.
+    pub fn into_reply(mut self) -> Self {
+        self.buffer.header_mut().flags_mut().set_ack();
+        self
+    }
+
     /// Consumes this `MessageInit`, returning the underlying [`MessagePacket`].
     pub fn into_inner(self) -> MessagePacket {
         self.buffer