Remove `Envelope` type
This type that was wrapping `Message` isn't very useful here. If we decide that we do need something like that, it's best handled one layer below. Signed-off-by: Alexis Sellier <alexis@radicle.xyz>
This commit is contained in:
parent
156a3a576a
commit
09c438f8fb
|
|
@ -1,14 +1,14 @@
|
|||
use std::io;
|
||||
use std::marker::PhantomData;
|
||||
|
||||
use crate::service::message::Envelope;
|
||||
use crate::service::message::Message;
|
||||
use crate::wire;
|
||||
|
||||
/// Message stream decoder.
|
||||
///
|
||||
/// Used to for example turn a byte stream into network messages.
|
||||
#[derive(Debug)]
|
||||
pub struct Decoder<D = Envelope> {
|
||||
pub struct Decoder<D = Message> {
|
||||
unparsed: Vec<u8>,
|
||||
item: PhantomData<D>,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ pub mod prelude {
|
|||
pub use crate::identity::{Did, Id};
|
||||
pub use crate::service::filter::Filter;
|
||||
pub use crate::service::message::Address;
|
||||
pub use crate::service::{DisconnectReason, Envelope, Event, Message, Network, NodeId};
|
||||
pub use crate::service::{DisconnectReason, Event, Message, Network, NodeId};
|
||||
pub use crate::storage::refs::Refs;
|
||||
pub use crate::storage::WriteStorage;
|
||||
pub use crate::{LocalDuration, LocalTime};
|
||||
|
|
|
|||
|
|
@ -38,7 +38,7 @@ use crate::storage;
|
|||
use crate::storage::{Inventory, ReadRepository, RefUpdate, WriteRepository, WriteStorage};
|
||||
|
||||
pub use crate::service::config::{Config, Network};
|
||||
pub use crate::service::message::{Envelope, Message, ZeroBytes};
|
||||
pub use crate::service::message::{Message, ZeroBytes};
|
||||
pub use crate::service::peer::Session;
|
||||
|
||||
use self::gossip::Gossip;
|
||||
|
|
@ -246,7 +246,6 @@ where
|
|||
rng: Rng,
|
||||
) -> Self {
|
||||
let sessions = Sessions::new(rng.clone());
|
||||
let network = config.network;
|
||||
|
||||
Self {
|
||||
config,
|
||||
|
|
@ -259,7 +258,7 @@ where
|
|||
gossip: Gossip::default(),
|
||||
// FIXME: This should be loaded from the address store.
|
||||
nodes: BTreeMap::new(),
|
||||
reactor: Reactor::new(network),
|
||||
reactor: Reactor::default(),
|
||||
sessions,
|
||||
out_of_sync: false,
|
||||
last_idle: LocalTime::default(),
|
||||
|
|
@ -586,8 +585,8 @@ where
|
|||
}
|
||||
}
|
||||
|
||||
pub fn received_message(&mut self, addr: &net::SocketAddr, envelope: Envelope) {
|
||||
match self.handle_message(addr, envelope) {
|
||||
pub fn received_message(&mut self, addr: &net::SocketAddr, message: Message) {
|
||||
match self.handle_message(addr, message) {
|
||||
Err(SessionError::NotFound(ip)) => {
|
||||
error!("Session not found for {ip}");
|
||||
}
|
||||
|
|
@ -736,7 +735,7 @@ where
|
|||
pub fn handle_message(
|
||||
&mut self,
|
||||
remote: &net::SocketAddr,
|
||||
envelope: Envelope,
|
||||
message: Message,
|
||||
) -> Result<(), peer::SessionError> {
|
||||
let peer_ip = remote.ip();
|
||||
let peer = if let Some(peer) = self.sessions.get_mut(&peer_ip) {
|
||||
|
|
@ -746,12 +745,9 @@ where
|
|||
};
|
||||
peer.last_active = self.clock.local_time();
|
||||
|
||||
if envelope.magic != self.config.network.magic() {
|
||||
return Err(SessionError::WrongMagic(envelope.magic));
|
||||
}
|
||||
debug!("Received {:?} from {}", &envelope.msg, peer.ip());
|
||||
debug!("Received {:?} from {}", &message, peer.ip());
|
||||
|
||||
match (&mut peer.state, envelope.msg) {
|
||||
match (&mut peer.state, message) {
|
||||
(
|
||||
SessionState::Initial,
|
||||
Message::Initialize {
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ use crate::git;
|
|||
use crate::git::Url;
|
||||
use crate::identity::{Id, PublicKey};
|
||||
use crate::service::filter::Filter;
|
||||
use crate::service::message::{Address, Envelope, Message};
|
||||
use crate::service::message::Address;
|
||||
|
||||
/// Peer-to-peer network.
|
||||
#[derive(Default, Debug, Copy, Clone, PartialEq, Eq)]
|
||||
|
|
@ -13,22 +13,6 @@ pub enum Network {
|
|||
Test,
|
||||
}
|
||||
|
||||
impl Network {
|
||||
pub fn magic(&self) -> u32 {
|
||||
match self {
|
||||
Self::Main => 0x819b43d9,
|
||||
Self::Test => 0x717ebaf8,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn envelope(&self, msg: Message) -> Envelope {
|
||||
Envelope {
|
||||
magic: self.magic(),
|
||||
msg,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Project tracking policy.
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum ProjectTracking {
|
||||
|
|
|
|||
|
|
@ -12,15 +12,6 @@ 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)]
|
||||
pub struct Envelope {
|
||||
/// Network magic constant. Used to differentiate networks.
|
||||
pub magic: u32,
|
||||
/// The message payload.
|
||||
pub msg: Message,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
// TODO: We should check the length and charset when deserializing.
|
||||
pub struct Hostname(String);
|
||||
|
|
|
|||
|
|
@ -37,8 +37,6 @@ pub enum SessionState {
|
|||
|
||||
#[derive(thiserror::Error, Debug)]
|
||||
pub enum SessionError {
|
||||
#[error("wrong network constant in message: {0}")]
|
||||
WrongMagic(u32),
|
||||
#[error("wrong protocol version in message: {0}")]
|
||||
WrongVersion(u32),
|
||||
#[error("invalid announcement timestamp: {0}")]
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ use super::message::{Announcement, AnnouncementMessage};
|
|||
#[derive(Debug)]
|
||||
pub enum Io {
|
||||
/// There are some messages ready to be sent to a peer.
|
||||
Write(net::SocketAddr, Vec<Envelope>),
|
||||
Write(net::SocketAddr, Vec<Message>),
|
||||
/// Connect to a peer.
|
||||
Connect(net::SocketAddr),
|
||||
/// Disconnect from a peer.
|
||||
|
|
@ -24,22 +24,13 @@ pub enum Io {
|
|||
}
|
||||
|
||||
/// Interface to the network reactor.
|
||||
#[derive(Debug)]
|
||||
#[derive(Debug, Default)]
|
||||
pub struct Reactor {
|
||||
/// The network we're on.
|
||||
network: Network,
|
||||
/// Outgoing I/O queue.
|
||||
io: VecDeque<Io>,
|
||||
}
|
||||
|
||||
impl Reactor {
|
||||
pub fn new(network: Network) -> Self {
|
||||
Self {
|
||||
network,
|
||||
io: VecDeque::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Emit an event.
|
||||
pub fn event(&mut self, event: Event) {
|
||||
self.io.push_back(Io::Event(event));
|
||||
|
|
@ -71,16 +62,12 @@ impl Reactor {
|
|||
pub fn write(&mut self, remote: net::SocketAddr, msg: Message) {
|
||||
debug!("Write {:?} to {}", &msg, remote.ip());
|
||||
|
||||
let envelope = self.network.envelope(msg);
|
||||
self.io.push_back(Io::Write(remote, vec![envelope]));
|
||||
self.io.push_back(Io::Write(remote, vec![msg]));
|
||||
}
|
||||
|
||||
pub fn write_all(&mut self, remote: net::SocketAddr, msgs: impl IntoIterator<Item = Message>) {
|
||||
let envelopes = msgs
|
||||
.into_iter()
|
||||
.map(|msg| self.network.envelope(msg))
|
||||
.collect();
|
||||
self.io.push_back(Io::Write(remote, envelopes));
|
||||
self.io
|
||||
.push_back(Io::Write(remote, msgs.into_iter().collect()));
|
||||
}
|
||||
|
||||
pub fn wakeup(&mut self, after: LocalDuration) {
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ use crate::crypto;
|
|||
use crate::prelude::{Id, NodeId, Refs, Timestamp};
|
||||
use crate::service::filter::{Filter, FILTER_SIZE_L, FILTER_SIZE_M, FILTER_SIZE_S};
|
||||
use crate::service::message::{
|
||||
Address, Announcement, Envelope, InventoryAnnouncement, Message, NodeAnnouncement, Ping,
|
||||
Address, Announcement, InventoryAnnouncement, Message, NodeAnnouncement, Ping,
|
||||
RefsAnnouncement, Subscribe, ZeroBytes,
|
||||
};
|
||||
use crate::wire::message::MessageType;
|
||||
|
|
@ -28,15 +28,6 @@ impl Arbitrary for Filter {
|
|||
}
|
||||
}
|
||||
|
||||
impl Arbitrary for Envelope {
|
||||
fn arbitrary(g: &mut quickcheck::Gen) -> Self {
|
||||
Self {
|
||||
magic: u32::arbitrary(g),
|
||||
msg: Message::arbitrary(g),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Arbitrary for Message {
|
||||
fn arbitrary(g: &mut quickcheck::Gen) -> Self {
|
||||
let type_id = g
|
||||
|
|
|
|||
|
|
@ -154,8 +154,7 @@ where
|
|||
}
|
||||
|
||||
pub fn receive(&mut self, peer: &net::SocketAddr, msg: Message) {
|
||||
self.service
|
||||
.received_message(peer, self.config().network.envelope(msg));
|
||||
self.service.received_message(peer, msg);
|
||||
}
|
||||
|
||||
pub fn inventory_announcement(&self) -> Message {
|
||||
|
|
@ -262,8 +261,8 @@ where
|
|||
let mut msgs = Vec::new();
|
||||
|
||||
self.service.reactor().outbox().retain(|o| match o {
|
||||
Io::Write(a, envelopes) if a == remote => {
|
||||
msgs.extend(envelopes.iter().map(|e| e.msg.clone()));
|
||||
Io::Write(a, messages) if a == remote => {
|
||||
msgs.extend(messages.clone());
|
||||
false
|
||||
}
|
||||
_ => true,
|
||||
|
|
|
|||
|
|
@ -17,7 +17,7 @@ use nakamoto_net::{Link, LocalDuration, LocalTime};
|
|||
|
||||
use crate::crypto::Signer;
|
||||
use crate::service::reactor::Io;
|
||||
use crate::service::{DisconnectReason, Envelope, Event};
|
||||
use crate::service::{DisconnectReason, Event, Message};
|
||||
use crate::storage::WriteStorage;
|
||||
use crate::test::peer::Service;
|
||||
|
||||
|
|
@ -64,7 +64,7 @@ pub enum Input {
|
|||
Rc<nakamoto::DisconnectReason<DisconnectReason>>,
|
||||
),
|
||||
/// Received a message from a remote peer.
|
||||
Received(net::SocketAddr, Vec<Envelope>),
|
||||
Received(net::SocketAddr, Vec<Message>),
|
||||
/// Used to advance the state machine after some wall time has passed.
|
||||
Wake,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -296,26 +296,6 @@ impl wire::Decode for Message {
|
|||
}
|
||||
}
|
||||
|
||||
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;
|
||||
|
|
@ -404,18 +384,10 @@ mod tests {
|
|||
);
|
||||
}
|
||||
|
||||
#[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);
|
||||
fn prop_message_decoder() {
|
||||
fn property(items: Vec<Message>) {
|
||||
let mut decoder = Decoder::<Message>::new(8);
|
||||
|
||||
for item in &items {
|
||||
item.encode(&mut decoder).unwrap();
|
||||
|
|
@ -427,7 +399,7 @@ mod tests {
|
|||
|
||||
quickcheck::QuickCheck::new()
|
||||
.gen(quickcheck::Gen::new(16))
|
||||
.quickcheck(property as fn(items: Vec<Envelope>));
|
||||
.quickcheck(property as fn(items: Vec<Message>));
|
||||
}
|
||||
|
||||
#[quickcheck]
|
||||
|
|
|
|||
Loading…
Reference in New Issue