diff --git a/zebra-network/src/peer/server.rs b/zebra-network/src/peer/server.rs index f7782770..743ca6b6 100644 --- a/zebra-network/src/peer/server.rs +++ b/zebra-network/src/peer/server.rs @@ -304,6 +304,7 @@ where // and try to construct an appropriate request object. let req = match msg { Message::Addr(addrs) => Some(Request::PushPeers(addrs)), + Message::GetAddr => Some(Request::GetPeers), _ => { debug!("unhandled message type"); None diff --git a/zebrad/src/commands/seed.rs b/zebrad/src/commands/seed.rs index 0ac7352a..78e82d2e 100644 --- a/zebrad/src/commands/seed.rs +++ b/zebrad/src/commands/seed.rs @@ -3,12 +3,69 @@ use std::{ future::Future, pin::Pin, + sync::{Arc, Mutex}, task::{Context, Poll}, }; +use abscissa_core::{config, Command, FrameworkError, Options, Runnable}; +use futures::stream::StreamExt; +use tower::{buffer::Buffer, Service, ServiceExt}; +use zebra_network::{AddressBook, BoxedStdError, Request, Response}; + use crate::{config::ZebradConfig, prelude::*}; -use abscissa_core::{config, Command, FrameworkError, Options, Runnable}; +#[derive(Clone)] +struct SeedService { + address_book: Option>>, +} + +impl SeedService { + fn set_address_book(&mut self, address_book: Arc>) { + debug!("Settings SeedService.address_book: {:?}", address_book); + self.address_book = Some(address_book); + } +} + +impl Service for SeedService { + type Response = Response; + type Error = BoxedStdError; + type Future = + Pin> + Send + 'static>>; + + fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll> { + Ok(()).into() + } + + fn call(&mut self, req: Request) -> Self::Future { + info!("SeedService handling a request: {:?}", req); + + match &self.address_book { + Some(address_book) => trace!( + "SeedService address_book total: {:?}", + address_book.lock().unwrap().len() + ), + _ => (), + }; + + let response = match req { + Request::GetPeers => match &self.address_book { + Some(address_book) => { + info!("Responding to GetPeers"); + + Ok::(Response::Peers( + address_book.lock().unwrap().peers().collect(), + )) + } + _ => Ok::(Response::Ok), + }, + _ => Ok::(Response::Ok), + }; + + info!("SeedService response: {:?}", response); + + return Box::pin(futures::future::ready(response)); + } +} /// `seed` subcommand /// @@ -30,6 +87,7 @@ impl config::Override for SeedCmd { config.tracing.filter = self.filters.join(","); } + info!("{:?}", config); Ok(config) } } @@ -69,54 +127,18 @@ impl Runnable for SeedCmd { impl SeedCmd { async fn seed(&self) -> Result<(), failure::Error> { use failure::Error; - use futures::stream::StreamExt; - use tower::{buffer::Buffer, Service, ServiceExt}; - use zebra_network::{AddressBook, Request, Response}; info!("begin tower-based peer handling test stub"); - struct SeedService { - address_book: Option, - } - - impl SeedService { - fn set_address_book(&mut self, address_book: AddressBook) { - self.address_book = Some(address_book); - } - } - - impl Service for SeedService { - type Response = Response; - type Error = Error; - type Future = - Pin> + Send + 'static>>; - - fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll> { - Ok(()).into() - } - - fn call(&mut self, req: Request) -> Self::Future { - let response = match req { - Request::GetPeers => match &self.address_book { - Some(address_book) => Ok::(Response::Peers( - address_book.peers().collect(), - )), - _ => Ok::(Response::Ok), - }, - _ => Ok::(Response::Ok), - }; - - return Box::pin(futures::future::ready(response)); - } - } - - let node = Buffer::new(SeedService { address_book: None }, 1); + let mut seed_service = SeedService { address_book: None }; + // let node = Buffer::new(seed_service, 1); let config = app_config().network.clone(); + info!("{:?}", config); - // XXX How do I create a service above that answers questions - // about this specific address book? - let (mut peer_set, address_book) = zebra_network::init(config, node).await; + let (mut peer_set, address_book) = zebra_network::init(config, seed_service.clone()).await; + + seed_service.set_address_book(address_book.clone()); // XXX Do not tell our DNS seed queries about gossiped addrs // that we have not connected to before? @@ -125,6 +147,21 @@ impl SeedCmd { info!("peer_set became ready"); + use std::time::Duration; + use tokio::timer::Interval; + + //#[cfg(dos)] + // Fire GetPeers requests at ourselves, for testing. + tokio::spawn(async move { + let mut interval_stream = Interval::new_interval(Duration::from_secs(1)); + + loop { + interval_stream.next().await; + + let _ = seed_service.call(Request::GetPeers); + } + }); + let eternity = tokio::future::pending::<()>(); eternity.await;