node: Move wire code

Signed-off-by: Alexis Sellier <alexis@radicle.xyz>

node/src/client.rs

@@ -8,9 +8,9 @@ use crate::clock::RefClock;
 use crate::collections::HashMap;
 use crate::crypto::Signer;
 use crate::service;
-use crate::service::wire::Wire;
 use crate::storage::git::Storage;
 use crate::transport::Transport;
+use crate::wire::Wire;
 
 pub mod handle;
 

node/src/decoder.rs

@@ -2,7 +2,7 @@ use std::io;
 use std::marker::PhantomData;
 
 use crate::service::message::Envelope;
-use crate::service::wire;
+use crate::wire;
 
 /// Message stream decoder.
 ///

node/src/lib.rs

@@ -4,6 +4,7 @@ pub use nakamoto_net::{Io, Link, LocalDuration, LocalTime};
 pub mod client;
 pub mod control;
 pub mod crypto;
+pub mod storage;
 
 mod address_book;
 mod address_manager;
@@ -17,7 +18,18 @@ mod logger;
 mod rad;
 mod serde_ext;
 mod service;
-mod storage;
 #[cfg(test)]
 mod test;
 mod transport;
+mod wire;
+
+pub mod prelude {
+    pub use crate::crypto::{PublicKey, Signature, Signer};
+    pub use crate::decoder::Decoder;
+    pub use crate::hash::Digest;
+    pub use crate::identity::{Did, Id};
+    pub use crate::service::filter::Filter;
+    pub use crate::service::{NodeId, Timestamp};
+    pub use crate::storage::refs::Refs;
+    pub use crate::storage::WriteStorage;
+}

node/src/service.rs

@@ -3,7 +3,6 @@ pub mod config;
 pub mod filter;
 pub mod message;
 pub mod peer;
-pub mod wire;
 
 use std::ops::{Deref, DerefMut};
 use std::{collections::VecDeque, fmt, net, net::IpAddr};

node/src/service/filter.rs

@@ -1,10 +1,8 @@
-use std::io;
 use std::ops::{Deref, DerefMut};
 
-use bloomy::BloomFilter;
+pub use bloomy::BloomFilter;
 
 use crate::identity::Id;
-use crate::service::wire;
 
 /// Size in bytes of subscription bloom filter.
 pub const FILTER_SIZE: usize = 1024 * 16;
@@ -51,34 +49,8 @@ impl DerefMut for Filter {
     }
 }
 
-#[cfg(test)]
 impl From<BloomFilter<Id>> for Filter {
     fn from(bloom: BloomFilter<Id>) -> Self {
         Self(bloom)
     }
 }
-
-impl wire::Encode for Filter {
-    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
-        let mut n = 0;
-
-        n += self.0.as_bytes().encode(writer)?;
-
-        Ok(n)
-    }
-}
-
-impl wire::Decode for Filter {
-    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
-        let size: wire::Size = wire::Decode::decode(reader)?;
-        if size as usize != FILTER_SIZE {
-            return Err(wire::Error::InvalidFilterSize(size as usize));
-        }
-        let bytes: [u8; FILTER_SIZE] = wire::Decode::decode(reader)?;
-        let bf = BloomFilter::from(Vec::from(bytes));
-
-        debug_assert_eq!(bf.hashes(), FILTER_HASHES);
-
-        Ok(Self(bf))
-    }
-}

node/src/service/message.rs

@@ -1,14 +1,12 @@
 use std::{fmt, io, net};
 
-use byteorder::{NetworkEndian, ReadBytesExt};
-
 use crate::crypto;
 use crate::git;
 use crate::identity::Id;
 use crate::service::filter::Filter;
-use crate::service::wire;
 use crate::service::{NodeId, Timestamp, PROTOCOL_VERSION};
 use crate::storage::refs::Refs;
+use crate::wire;
 
 /// Message envelope. All messages sent over the network are wrapped in this type.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -26,68 +24,6 @@ pub type NodeFeatures = [u8; 32];
 // TODO: We should check the length and charset when deserializing.
 pub struct Hostname(String);
 
-/// Message type.
-#[repr(u16)]
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum MessageType {
-    Initialize = 0,
-    NodeAnnouncement = 2,
-    InventoryAnnouncement = 4,
-    RefsAnnouncement = 6,
-    Subscribe = 8,
-}
-
-impl From<MessageType> for u16 {
-    fn from(other: MessageType) -> Self {
-        other as u16
-    }
-}
-
-impl TryFrom<u16> for MessageType {
-    type Error = u16;
-
-    fn try_from(other: u16) -> Result<Self, Self::Error> {
-        match other {
-            0 => Ok(MessageType::Initialize),
-            2 => Ok(MessageType::NodeAnnouncement),
-            4 => Ok(MessageType::InventoryAnnouncement),
-            6 => Ok(MessageType::RefsAnnouncement),
-            8 => Ok(MessageType::Subscribe),
-            _ => Err(other),
-        }
-    }
-}
-
-/// Address type.
-#[repr(u8)]
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum AddressType {
-    Ipv4 = 1,
-    Ipv6 = 2,
-    Hostname = 3,
-    Onion = 4,
-}
-
-impl From<AddressType> for u8 {
-    fn from(other: AddressType) -> Self {
-        other as u8
-    }
-}
-
-impl TryFrom<u8> for AddressType {
-    type Error = u8;
-
-    fn try_from(other: u8) -> Result<Self, Self::Error> {
-        match other {
-            1 => Ok(AddressType::Ipv4),
-            2 => Ok(AddressType::Ipv6),
-            3 => Ok(AddressType::Hostname),
-            4 => Ok(AddressType::Onion),
-            _ => Err(other),
-        }
-    }
-}
-
 /// Peer public protocol address.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum Address {
@@ -123,78 +59,6 @@ impl From<net::SocketAddr> for Address {
     }
 }
 
-impl wire::Encode for Envelope {
-    fn encode<W: std::io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, std::io::Error> {
-        let mut n = 0;
-
-        n += self.magic.encode(writer)?;
-        n += self.msg.encode(writer)?;
-
-        Ok(n)
-    }
-}
-
-impl wire::Decode for Envelope {
-    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
-        let magic = u32::decode(reader)?;
-        let msg = Message::decode(reader)?;
-
-        Ok(Self { magic, msg })
-    }
-}
-
-impl wire::Encode for Address {
-    fn encode<W: std::io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, std::io::Error> {
-        let mut n = 0;
-
-        match self {
-            Self::Ipv4 { ip, port } => {
-                n += u8::from(AddressType::Ipv4).encode(writer)?;
-                n += ip.octets().encode(writer)?;
-                n += port.encode(writer)?;
-            }
-            Self::Ipv6 { ip, port } => {
-                n += u8::from(AddressType::Ipv6).encode(writer)?;
-                n += ip.octets().encode(writer)?;
-                n += port.encode(writer)?;
-            }
-            Self::Hostname { .. } => todo!(),
-            Self::Onion { .. } => todo!(),
-        }
-        Ok(n)
-    }
-}
-
-impl wire::Decode for Address {
-    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
-        let addrtype = reader.read_u8()?;
-
-        match AddressType::try_from(addrtype) {
-            Ok(AddressType::Ipv4) => {
-                let octets: [u8; 4] = wire::Decode::decode(reader)?;
-                let ip = net::Ipv4Addr::from(octets);
-                let port = u16::decode(reader)?;
-
-                Ok(Self::Ipv4 { ip, port })
-            }
-            Ok(AddressType::Ipv6) => {
-                let octets: [u8; 16] = wire::Decode::decode(reader)?;
-                let ip = net::Ipv6Addr::from(octets);
-                let port = u16::decode(reader)?;
-
-                Ok(Self::Ipv6 { ip, port })
-            }
-            Ok(AddressType::Hostname) => {
-                todo!();
-            }
-            Ok(AddressType::Onion) => {
-                todo!();
-            }
-            Err(other) => Err(wire::Error::UnknownAddressType(other)),
-        }
-    }
-}
-
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Subscribe {
     /// Subscribe to events matching this filter.
@@ -276,26 +140,6 @@ impl RefsAnnouncement {
     }
 }
 
-impl wire::Encode for RefsAnnouncement {
-    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
-        let mut n = 0;
-
-        n += self.id.encode(writer)?;
-        n += self.refs.encode(writer)?;
-
-        Ok(n)
-    }
-}
-
-impl wire::Decode for RefsAnnouncement {
-    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
-        let id = Id::decode(reader)?;
-        let refs = Refs::decode(reader)?;
-
-        Ok(Self { id, refs })
-    }
-}
-
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct InventoryAnnouncement {
     pub inventory: Vec<Id>,
@@ -310,29 +154,6 @@ impl InventoryAnnouncement {
     }
 }
 
-impl wire::Encode for InventoryAnnouncement {
-    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
-        let mut n = 0;
-
-        n += self.inventory.as_slice().encode(writer)?;
-        n += self.timestamp.encode(writer)?;
-
-        Ok(n)
-    }
-}
-
-impl wire::Decode for InventoryAnnouncement {
-    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
-        let inventory = Vec::<Id>::decode(reader)?;
-        let timestamp = Timestamp::decode(reader)?;
-
-        Ok(Self {
-            inventory,
-            timestamp,
-        })
-    }
-}
-
 /// Message payload.
 /// These are the messages peers send to each other.
 #[derive(Clone, PartialEq, Eq)]
@@ -425,17 +246,6 @@ impl Message {
             until,
         })
     }
-
-    pub fn type_id(&self) -> u16 {
-        match self {
-            Self::Initialize { .. } => MessageType::Initialize,
-            Self::Subscribe { .. } => MessageType::Subscribe,
-            Self::NodeAnnouncement { .. } => MessageType::NodeAnnouncement,
-            Self::InventoryAnnouncement { .. } => MessageType::InventoryAnnouncement,
-            Self::RefsAnnouncement { .. } => MessageType::RefsAnnouncement,
-        }
-        .into()
-    }
 }
 
 impl fmt::Debug for Message {
@@ -472,186 +282,14 @@ impl fmt::Debug for Message {
     }
 }
 
-impl wire::Encode for Message {
-    fn encode<W: std::io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, std::io::Error> {
-        let mut n = self.type_id().encode(writer)?;
-
-        match self {
-            Self::Initialize {
-                id,
-                timestamp,
-                version,
-                addrs,
-                git,
-            } => {
-                n += id.encode(writer)?;
-                n += timestamp.encode(writer)?;
-                n += version.encode(writer)?;
-                n += addrs.as_slice().encode(writer)?;
-                n += git.encode(writer)?;
-            }
-            Self::Subscribe(Subscribe {
-                filter,
-                since,
-                until,
-            }) => {
-                n += filter.encode(writer)?;
-                n += since.encode(writer)?;
-                n += until.encode(writer)?;
-            }
-            Self::RefsAnnouncement {
-                node,
-                message,
-                signature,
-            } => {
-                n += node.encode(writer)?;
-                n += message.encode(writer)?;
-                n += signature.encode(writer)?;
-            }
-            Self::InventoryAnnouncement {
-                node,
-                message,
-                signature,
-            } => {
-                n += node.encode(writer)?;
-                n += message.encode(writer)?;
-                n += signature.encode(writer)?;
-            }
-            Self::NodeAnnouncement {
-                node,
-                message,
-                signature,
-            } => {
-                n += node.encode(writer)?;
-                n += message.encode(writer)?;
-                n += signature.encode(writer)?;
-            }
-        }
-        Ok(n)
-    }
-}
-
-impl wire::Decode for Message {
-    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
-        let type_id = reader.read_u16::<NetworkEndian>()?;
-
-        match MessageType::try_from(type_id) {
-            Ok(MessageType::Initialize) => {
-                let id = NodeId::decode(reader)?;
-                let timestamp = Timestamp::decode(reader)?;
-                let version = u32::decode(reader)?;
-                let addrs = Vec::<Address>::decode(reader)?;
-                let git = git::Url::decode(reader)?;
-
-                Ok(Self::Initialize {
-                    id,
-                    timestamp,
-                    version,
-                    addrs,
-                    git,
-                })
-            }
-            Ok(MessageType::Subscribe) => {
-                let filter = Filter::decode(reader)?;
-                let since = Timestamp::decode(reader)?;
-                let until = Timestamp::decode(reader)?;
-
-                Ok(Self::Subscribe(Subscribe {
-                    filter,
-                    since,
-                    until,
-                }))
-            }
-            Ok(MessageType::NodeAnnouncement) => {
-                let node = NodeId::decode(reader)?;
-                let message = NodeAnnouncement::decode(reader)?;
-                let signature = crypto::Signature::decode(reader)?;
-
-                Ok(Self::NodeAnnouncement {
-                    node,
-                    message,
-                    signature,
-                })
-            }
-            Ok(MessageType::InventoryAnnouncement) => {
-                let node = NodeId::decode(reader)?;
-                let message = InventoryAnnouncement::decode(reader)?;
-                let signature = crypto::Signature::decode(reader)?;
-
-                Ok(Self::InventoryAnnouncement {
-                    node,
-                    message,
-                    signature,
-                })
-            }
-            Ok(MessageType::RefsAnnouncement) => {
-                let node = NodeId::decode(reader)?;
-                let message = RefsAnnouncement::decode(reader)?;
-                let signature = crypto::Signature::decode(reader)?;
-
-                Ok(Self::RefsAnnouncement {
-                    node,
-                    message,
-                    signature,
-                })
-            }
-            Err(other) => Err(wire::Error::UnknownMessageType(other)),
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
     use quickcheck_macros::quickcheck;
 
     use crate::crypto::Signer;
-    use crate::decoder::Decoder;
-    use crate::service::wire::{self, Encode};
     use crate::test::crypto::MockSigner;
 
-    #[quickcheck]
-    fn prop_message_encode_decode(message: Message) {
-        assert_eq!(
-            wire::deserialize::<Message>(&wire::serialize(&message)).unwrap(),
-            message
-        );
-    }
-
-    #[quickcheck]
-    fn prop_envelope_encode_decode(envelope: Envelope) {
-        assert_eq!(
-            wire::deserialize::<Envelope>(&wire::serialize(&envelope)).unwrap(),
-            envelope
-        );
-    }
-
-    #[test]
-    fn prop_envelope_decoder() {
-        fn property(items: Vec<Envelope>) {
-            let mut decoder = Decoder::<Envelope>::new(8);
-
-            for item in &items {
-                item.encode(&mut decoder).unwrap();
-            }
-            for item in items {
-                assert_eq!(decoder.next().unwrap().unwrap(), item);
-            }
-        }
-
-        quickcheck::QuickCheck::new()
-            .gen(quickcheck::Gen::new(16))
-            .quickcheck(property as fn(items: Vec<Envelope>));
-    }
-
-    #[quickcheck]
-    fn prop_addr(addr: Address) {
-        assert_eq!(
-            wire::deserialize::<Address>(&wire::serialize(&addr)).unwrap(),
-            addr
-        );
-    }
-
     #[quickcheck]
     fn prop_refs_announcement_signing(id: Id, refs: Refs) {
         let signer = MockSigner::new(&mut fastrand::Rng::new());

node/src/storage/refs.rs

@@ -15,9 +15,9 @@ use crate::crypto;
 use crate::crypto::{PublicKey, Signature, Signer, Unverified, Verified};
 use crate::git;
 use crate::git::Oid;
-use crate::service::wire;
 use crate::storage;
 use crate::storage::{ReadRepository, RemoteId, WriteRepository};
+use crate::wire;
 
 pub static SIGNATURE_REF: Lazy<git::RefString> = Lazy::new(|| git::refname!("radicle/signature"));
 pub const REFS_BLOB_PATH: &str = "refs";

node/src/test/arbitrary.rs

@@ -16,13 +16,14 @@ use crate::hash;
 use crate::identity::{Delegate, Did, Doc, Id, Project};
 use crate::service::filter::{Filter, FILTER_SIZE};
 use crate::service::message::{
-    Address, Envelope, InventoryAnnouncement, Message, MessageType, NodeAnnouncement,
-    RefsAnnouncement, Subscribe,
+    Address, Envelope, InventoryAnnouncement, Message, NodeAnnouncement, RefsAnnouncement,
+    Subscribe,
 };
 use crate::service::{NodeId, Timestamp};
 use crate::storage;
 use crate::storage::refs::{Refs, SignedRefs};
 use crate::test::storage::MockStorage;
+use crate::wire::message::MessageType;
 
 use super::crypto::MockSigner;
 

node/src/transport.rs

@@ -8,9 +8,9 @@ use nakamoto_net::{Io, Link};
 use crate::address_book;
 use crate::collections::HashMap;
 use crate::crypto;
-use crate::service::wire::Wire;
 use crate::service::{Command, DisconnectReason, Event, Service};
 use crate::storage::WriteStorage;
+use crate::wire::Wire;
 
 #[derive(Debug)]
 struct Peer {

node/src/wire.rs

@@ -1,3 +1,5 @@
+pub mod message;
+
 use std::collections::{BTreeMap, HashMap};
 use std::convert::TryFrom;
 use std::net::IpAddr;
@@ -17,6 +19,7 @@ use crate::git::fmt;
 use crate::hash::Digest;
 use crate::identity::Id;
 use crate::service;
+use crate::service::filter;
 use crate::storage::refs::Refs;
 use crate::storage::WriteStorage;
 
@@ -386,6 +389,31 @@ impl Decode for Digest {
     }
 }
 
+impl Encode for filter::Filter {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        let mut n = 0;
+
+        n += self.deref().as_bytes().encode(writer)?;
+
+        Ok(n)
+    }
+}
+
+impl Decode for filter::Filter {
+    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, Error> {
+        let size: Size = Decode::decode(reader)?;
+        if size as usize != filter::FILTER_SIZE {
+            return Err(Error::InvalidFilterSize(size as usize));
+        }
+        let bytes: [u8; filter::FILTER_SIZE] = Decode::decode(reader)?;
+        let bf = filter::BloomFilter::from(Vec::from(bytes));
+
+        debug_assert_eq!(bf.hashes(), filter::FILTER_HASHES);
+
+        Ok(Self::from(bf))
+    }
+}
+
 #[derive(Debug)]
 pub struct Wire<S, T, G> {
     inboxes: HashMap<IpAddr, Decoder>,

node/src/wire/message.rs

@@ -0,0 +1,377 @@
+use std::{io, net};
+
+use byteorder::{NetworkEndian, ReadBytesExt};
+
+use crate::git;
+use crate::prelude::*;
+use crate::service::message::*;
+use crate::wire;
+
+/// Message type.
+#[repr(u16)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum MessageType {
+    Initialize = 0,
+    NodeAnnouncement = 2,
+    InventoryAnnouncement = 4,
+    RefsAnnouncement = 6,
+    Subscribe = 8,
+}
+
+impl From<MessageType> for u16 {
+    fn from(other: MessageType) -> Self {
+        other as u16
+    }
+}
+
+impl TryFrom<u16> for MessageType {
+    type Error = u16;
+
+    fn try_from(other: u16) -> Result<Self, Self::Error> {
+        match other {
+            0 => Ok(MessageType::Initialize),
+            2 => Ok(MessageType::NodeAnnouncement),
+            4 => Ok(MessageType::InventoryAnnouncement),
+            6 => Ok(MessageType::RefsAnnouncement),
+            8 => Ok(MessageType::Subscribe),
+            _ => Err(other),
+        }
+    }
+}
+
+impl Message {
+    pub fn type_id(&self) -> u16 {
+        match self {
+            Self::Initialize { .. } => MessageType::Initialize,
+            Self::Subscribe { .. } => MessageType::Subscribe,
+            Self::NodeAnnouncement { .. } => MessageType::NodeAnnouncement,
+            Self::InventoryAnnouncement { .. } => MessageType::InventoryAnnouncement,
+            Self::RefsAnnouncement { .. } => MessageType::RefsAnnouncement,
+        }
+        .into()
+    }
+}
+
+/// Address type.
+#[repr(u8)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AddressType {
+    Ipv4 = 1,
+    Ipv6 = 2,
+    Hostname = 3,
+    Onion = 4,
+}
+
+impl From<AddressType> for u8 {
+    fn from(other: AddressType) -> Self {
+        other as u8
+    }
+}
+
+impl TryFrom<u8> for AddressType {
+    type Error = u8;
+
+    fn try_from(other: u8) -> Result<Self, Self::Error> {
+        match other {
+            1 => Ok(AddressType::Ipv4),
+            2 => Ok(AddressType::Ipv6),
+            3 => Ok(AddressType::Hostname),
+            4 => Ok(AddressType::Onion),
+            _ => Err(other),
+        }
+    }
+}
+
+impl wire::Encode for RefsAnnouncement {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        let mut n = 0;
+
+        n += self.id.encode(writer)?;
+        n += self.refs.encode(writer)?;
+
+        Ok(n)
+    }
+}
+
+impl wire::Decode for RefsAnnouncement {
+    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
+        let id = Id::decode(reader)?;
+        let refs = Refs::decode(reader)?;
+
+        Ok(Self { id, refs })
+    }
+}
+
+impl wire::Encode for InventoryAnnouncement {
+    fn encode<W: io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, io::Error> {
+        let mut n = 0;
+
+        n += self.inventory.as_slice().encode(writer)?;
+        n += self.timestamp.encode(writer)?;
+
+        Ok(n)
+    }
+}
+
+impl wire::Decode for InventoryAnnouncement {
+    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
+        let inventory = Vec::<Id>::decode(reader)?;
+        let timestamp = Timestamp::decode(reader)?;
+
+        Ok(Self {
+            inventory,
+            timestamp,
+        })
+    }
+}
+
+impl wire::Encode for Message {
+    fn encode<W: std::io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, std::io::Error> {
+        let mut n = self.type_id().encode(writer)?;
+
+        match self {
+            Self::Initialize {
+                id,
+                timestamp,
+                version,
+                addrs,
+                git,
+            } => {
+                n += id.encode(writer)?;
+                n += timestamp.encode(writer)?;
+                n += version.encode(writer)?;
+                n += addrs.as_slice().encode(writer)?;
+                n += git.encode(writer)?;
+            }
+            Self::Subscribe(Subscribe {
+                filter,
+                since,
+                until,
+            }) => {
+                n += filter.encode(writer)?;
+                n += since.encode(writer)?;
+                n += until.encode(writer)?;
+            }
+            Self::RefsAnnouncement {
+                node,
+                message,
+                signature,
+            } => {
+                n += node.encode(writer)?;
+                n += message.encode(writer)?;
+                n += signature.encode(writer)?;
+            }
+            Self::InventoryAnnouncement {
+                node,
+                message,
+                signature,
+            } => {
+                n += node.encode(writer)?;
+                n += message.encode(writer)?;
+                n += signature.encode(writer)?;
+            }
+            Self::NodeAnnouncement {
+                node,
+                message,
+                signature,
+            } => {
+                n += node.encode(writer)?;
+                n += message.encode(writer)?;
+                n += signature.encode(writer)?;
+            }
+        }
+        Ok(n)
+    }
+}
+
+impl wire::Decode for Message {
+    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
+        let type_id = reader.read_u16::<NetworkEndian>()?;
+
+        match MessageType::try_from(type_id) {
+            Ok(MessageType::Initialize) => {
+                let id = NodeId::decode(reader)?;
+                let timestamp = Timestamp::decode(reader)?;
+                let version = u32::decode(reader)?;
+                let addrs = Vec::<Address>::decode(reader)?;
+                let git = git::Url::decode(reader)?;
+
+                Ok(Self::Initialize {
+                    id,
+                    timestamp,
+                    version,
+                    addrs,
+                    git,
+                })
+            }
+            Ok(MessageType::Subscribe) => {
+                let filter = Filter::decode(reader)?;
+                let since = Timestamp::decode(reader)?;
+                let until = Timestamp::decode(reader)?;
+
+                Ok(Self::Subscribe(Subscribe {
+                    filter,
+                    since,
+                    until,
+                }))
+            }
+            Ok(MessageType::NodeAnnouncement) => {
+                let node = NodeId::decode(reader)?;
+                let message = NodeAnnouncement::decode(reader)?;
+                let signature = Signature::decode(reader)?;
+
+                Ok(Self::NodeAnnouncement {
+                    node,
+                    message,
+                    signature,
+                })
+            }
+            Ok(MessageType::InventoryAnnouncement) => {
+                let node = NodeId::decode(reader)?;
+                let message = InventoryAnnouncement::decode(reader)?;
+                let signature = Signature::decode(reader)?;
+
+                Ok(Self::InventoryAnnouncement {
+                    node,
+                    message,
+                    signature,
+                })
+            }
+            Ok(MessageType::RefsAnnouncement) => {
+                let node = NodeId::decode(reader)?;
+                let message = RefsAnnouncement::decode(reader)?;
+                let signature = Signature::decode(reader)?;
+
+                Ok(Self::RefsAnnouncement {
+                    node,
+                    message,
+                    signature,
+                })
+            }
+            Err(other) => Err(wire::Error::UnknownMessageType(other)),
+        }
+    }
+}
+
+impl wire::Encode for Envelope {
+    fn encode<W: std::io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, std::io::Error> {
+        let mut n = 0;
+
+        n += self.magic.encode(writer)?;
+        n += self.msg.encode(writer)?;
+
+        Ok(n)
+    }
+}
+
+impl wire::Decode for Envelope {
+    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
+        let magic = u32::decode(reader)?;
+        let msg = Message::decode(reader)?;
+
+        Ok(Self { magic, msg })
+    }
+}
+
+impl wire::Encode for Address {
+    fn encode<W: std::io::Write + ?Sized>(&self, writer: &mut W) -> Result<usize, std::io::Error> {
+        let mut n = 0;
+
+        match self {
+            Self::Ipv4 { ip, port } => {
+                n += u8::from(AddressType::Ipv4).encode(writer)?;
+                n += ip.octets().encode(writer)?;
+                n += port.encode(writer)?;
+            }
+            Self::Ipv6 { ip, port } => {
+                n += u8::from(AddressType::Ipv6).encode(writer)?;
+                n += ip.octets().encode(writer)?;
+                n += port.encode(writer)?;
+            }
+            Self::Hostname { .. } => todo!(),
+            Self::Onion { .. } => todo!(),
+        }
+        Ok(n)
+    }
+}
+
+impl wire::Decode for Address {
+    fn decode<R: std::io::Read + ?Sized>(reader: &mut R) -> Result<Self, wire::Error> {
+        let addrtype = reader.read_u8()?;
+
+        match AddressType::try_from(addrtype) {
+            Ok(AddressType::Ipv4) => {
+                let octets: [u8; 4] = wire::Decode::decode(reader)?;
+                let ip = net::Ipv4Addr::from(octets);
+                let port = u16::decode(reader)?;
+
+                Ok(Self::Ipv4 { ip, port })
+            }
+            Ok(AddressType::Ipv6) => {
+                let octets: [u8; 16] = wire::Decode::decode(reader)?;
+                let ip = net::Ipv6Addr::from(octets);
+                let port = u16::decode(reader)?;
+
+                Ok(Self::Ipv6 { ip, port })
+            }
+            Ok(AddressType::Hostname) => {
+                todo!();
+            }
+            Ok(AddressType::Onion) => {
+                todo!();
+            }
+            Err(other) => Err(wire::Error::UnknownAddressType(other)),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use quickcheck_macros::quickcheck;
+
+    use crate::decoder::Decoder;
+    use crate::wire::{self, Encode};
+
+    #[quickcheck]
+    fn prop_message_encode_decode(message: Message) {
+        assert_eq!(
+            wire::deserialize::<Message>(&wire::serialize(&message)).unwrap(),
+            message
+        );
+    }
+
+    #[quickcheck]
+    fn prop_envelope_encode_decode(envelope: Envelope) {
+        assert_eq!(
+            wire::deserialize::<Envelope>(&wire::serialize(&envelope)).unwrap(),
+            envelope
+        );
+    }
+
+    #[test]
+    fn prop_envelope_decoder() {
+        fn property(items: Vec<Envelope>) {
+            let mut decoder = Decoder::<Envelope>::new(8);
+
+            for item in &items {
+                item.encode(&mut decoder).unwrap();
+            }
+            for item in items {
+                assert_eq!(decoder.next().unwrap().unwrap(), item);
+            }
+        }
+
+        quickcheck::QuickCheck::new()
+            .gen(quickcheck::Gen::new(16))
+            .quickcheck(property as fn(items: Vec<Envelope>));
+    }
+
+    #[quickcheck]
+    fn prop_addr(addr: Address) {
+        assert_eq!(
+            wire::deserialize::<Address>(&wire::serialize(&addr)).unwrap(),
+            addr
+        );
+    }
+}