node: Add wire protocol

Signed-off-by: Alexis Sellier <alexis@radicle.xyz>
This commit is contained in:
Alexis Sellier 2022-09-15 21:37:28 +02:00
parent a925fb0e02
commit 8e160f170c
No known key found for this signature in database
4 changed files with 108 additions and 19 deletions

View File

@ -8,6 +8,7 @@ use crate::clock::RefClock;
use crate::collections::HashMap;
use crate::crypto::Signer;
use crate::protocol;
use crate::protocol::wire::Wire;
use crate::storage::git::Storage;
use crate::transport::Transport;
@ -97,7 +98,7 @@ impl<R: Reactor, G: Signer> Client<R, G> {
);
self.reactor.run(
&config.listen,
Transport::new(protocol),
Transport::new(Wire::new(protocol)),
self.events,
self.commands,
)?;

View File

@ -1,17 +1,23 @@
use std::collections::BTreeMap;
use std::convert::TryFrom;
use std::ops::Deref;
use std::net;
use std::ops::{Deref, DerefMut};
use std::string::FromUtf8Error;
use std::{io, mem};
use byteorder::{NetworkEndian, ReadBytesExt, WriteBytesExt};
use nakamoto_net as nakamoto;
use nakamoto_net::{Link, LocalTime};
use crate::crypto::{PublicKey, Signature};
use crate::address_book;
use crate::crypto::{PublicKey, Signature, Signer};
use crate::git;
use crate::git::fmt;
use crate::hash::Digest;
use crate::identity::Id;
use crate::protocol;
use crate::storage::refs::Refs;
use crate::storage::WriteStorage;
/// The default type we use to represent sizes.
/// Four bytes is more than enough for anything sent over the wire.
@ -379,6 +385,86 @@ impl Decode for Digest {
}
}
#[derive(Debug)]
pub struct Wire<S, T, G> {
inner: protocol::Protocol<S, T, G>,
}
impl<S, T, G> Wire<S, T, G> {
pub fn new(inner: protocol::Protocol<S, T, G>) -> Self {
Self { inner }
}
}
impl<'r, S, T, G> Wire<S, T, G>
where
S: address_book::Store,
T: WriteStorage<'r> + 'static,
G: Signer,
{
fn initialize(&mut self, time: LocalTime) {
self.inner.initialize(time)
}
fn tick(&mut self, now: LocalTime) {
self.inner.tick(now)
}
fn wake(&mut self) {
self.inner.wake()
}
fn command(&mut self, cmd: protocol::Command) {
self.inner.command(cmd)
}
fn attempted(&mut self, addr: &net::SocketAddr) {
self.inner.attempted(addr)
}
fn connected(
&mut self,
addr: std::net::SocketAddr,
local_addr: &std::net::SocketAddr,
link: Link,
) {
self.inner.connected(addr, local_addr, link)
}
fn disconnected(
&mut self,
addr: &std::net::SocketAddr,
reason: nakamoto::DisconnectReason<protocol::DisconnectReason>,
) {
self.inner.disconnected(addr, reason)
}
fn received_bytes(&mut self, addr: &std::net::SocketAddr, bytes: &[u8]) {
self.inner.received_bytes(addr, bytes)
}
}
impl<S, T, G> Iterator for Wire<S, T, G> {
type Item = nakamoto::Io<protocol::Event, protocol::DisconnectReason>;
fn next(&mut self) -> Option<Self::Item> {
self.inner.next()
}
}
impl<S, T, G> Deref for Wire<S, T, G> {
type Target = protocol::Protocol<S, T, G>;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
impl<S, T, G> DerefMut for Wire<S, T, G> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.inner
}
}
#[cfg(test)]
mod tests {
use super::*;

View File

@ -12,6 +12,7 @@ use crate::collections::HashMap;
use crate::decoder::Decoder;
use crate::protocol::config::*;
use crate::protocol::message::*;
use crate::protocol::wire::Wire;
use crate::protocol::*;
use crate::storage::WriteStorage;
use crate::test::crypto::MockSigner;
@ -93,14 +94,14 @@ where
let local_time = LocalTime::now();
let clock = RefClock::from(local_time);
let signer = MockSigner::new(&mut rng);
let protocol = Transport::new(Protocol::new(
let protocol = Transport::new(Wire::new(Protocol::new(
config,
clock,
storage,
addrs,
signer,
rng.clone(),
));
)));
let ip = ip.into();
let local_addr = net::SocketAddr::new(ip, rng.u16(..));

View File

@ -8,6 +8,7 @@ use nakamoto_net::{Io, Link};
use crate::address_book;
use crate::collections::HashMap;
use crate::crypto;
use crate::protocol::wire::Wire;
use crate::protocol::{Command, DisconnectReason, Event, Protocol};
use crate::storage::WriteStorage;
@ -19,14 +20,14 @@ struct Peer {
#[derive(Debug)]
pub struct Transport<S, T, G> {
peers: HashMap<net::IpAddr, Peer>,
protocol: Protocol<S, T, G>,
inner: Wire<S, T, G>,
}
impl<S, T, G> Transport<S, T, G> {
pub fn new(protocol: Protocol<S, T, G>) -> Self {
pub fn new(inner: Wire<S, T, G>) -> Self {
Self {
peers: HashMap::default(),
protocol,
inner,
}
}
}
@ -42,23 +43,23 @@ where
type DisconnectReason = DisconnectReason;
fn initialize(&mut self, time: LocalTime) {
self.protocol.initialize(time)
self.inner.initialize(time)
}
fn tick(&mut self, now: nakamoto::LocalTime) {
self.protocol.tick(now)
self.inner.tick(now)
}
fn wake(&mut self) {
self.protocol.wake()
self.inner.wake()
}
fn command(&mut self, cmd: Self::Command) {
self.protocol.command(cmd)
self.inner.command(cmd)
}
fn attempted(&mut self, addr: &std::net::SocketAddr) {
self.protocol.attempted(addr)
self.inner.attempted(addr)
}
fn connected(
@ -67,7 +68,7 @@ where
local_addr: &std::net::SocketAddr,
link: Link,
) {
self.protocol.connected(addr, local_addr, link)
self.inner.connected(addr, local_addr, link)
}
fn disconnected(
@ -75,11 +76,11 @@ where
addr: &std::net::SocketAddr,
reason: nakamoto::DisconnectReason<Self::DisconnectReason>,
) {
self.protocol.disconnected(addr, reason)
self.inner.disconnected(addr, reason)
}
fn received_bytes(&mut self, addr: &std::net::SocketAddr, bytes: &[u8]) {
self.protocol.received_bytes(addr, bytes)
self.inner.received_bytes(addr, bytes)
}
}
@ -87,7 +88,7 @@ impl<S, T, G> Iterator for Transport<S, T, G> {
type Item = Io<Event, DisconnectReason>;
fn next(&mut self) -> Option<Self::Item> {
self.protocol.next()
self.inner.next()
}
}
@ -95,12 +96,12 @@ impl<S, T, G> Deref for Transport<S, T, G> {
type Target = Protocol<S, T, G>;
fn deref(&self) -> &Self::Target {
&self.protocol
&self.inner
}
}
impl<S, T, G> DerefMut for Transport<S, T, G> {
fn deref_mut(&mut self) -> &mut Self::Target {
&mut self.protocol
&mut self.inner
}
}