2018-03-16 11:12:14 -07:00
|
|
|
//! Networking controls of the consensus node.
|
|
|
|
use std::sync::mpsc;
|
|
|
|
use std::fmt::Debug;
|
|
|
|
use std::collections::{HashMap, HashSet};
|
2018-03-16 14:04:06 -07:00
|
|
|
use std::marker::{Send, Sync};
|
2018-03-16 11:12:14 -07:00
|
|
|
use std::net::{TcpStream, TcpListener, SocketAddr};
|
2018-03-16 14:04:06 -07:00
|
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use std::sync::mpsc::{Sender, Receiver, channel};
|
2018-03-16 11:12:14 -07:00
|
|
|
use broadcast::*;
|
2018-03-16 14:04:06 -07:00
|
|
|
use broadcast::stage::Stage as BroadcastStage;
|
|
|
|
use proto::Message;
|
2018-03-16 11:12:14 -07:00
|
|
|
|
|
|
|
/// This is a structure to start a consensus node.
|
|
|
|
pub struct Node {
|
|
|
|
/// Incoming connection socket.
|
|
|
|
addr: SocketAddr,
|
2018-03-16 17:36:32 -07:00
|
|
|
/// Sockets of remote nodes. TODO.
|
2018-03-16 11:12:14 -07:00
|
|
|
remotes: Vec<SocketAddr>
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Node {
|
|
|
|
pub fn new(addr: SocketAddr, remotes: Vec<SocketAddr>) -> Self {
|
|
|
|
Node {addr, remotes}
|
|
|
|
}
|
|
|
|
|
2018-03-16 14:04:06 -07:00
|
|
|
pub fn run<T: Clone + Debug + Send + Sync + 'static>(&self) {
|
2018-03-16 11:12:14 -07:00
|
|
|
// Listen for incoming connections on a given TCP port.
|
|
|
|
let listener = TcpListener::bind(&self.addr).unwrap();
|
2018-03-16 14:04:06 -07:00
|
|
|
let broadcast_stage: Arc<Mutex<BroadcastStage<T>>> =
|
|
|
|
Arc::new(Mutex::new(BroadcastStage::new(Vec::new())));
|
2018-03-16 11:12:14 -07:00
|
|
|
|
|
|
|
for stream in listener.incoming() {
|
|
|
|
match stream {
|
|
|
|
Ok(stream) => {
|
|
|
|
info!("New connection from {:?}",
|
|
|
|
stream.peer_addr().unwrap());
|
2018-03-16 14:04:06 -07:00
|
|
|
let (tx, rx): (Sender<Message<T>>, Receiver<Message<T>>) =
|
|
|
|
channel();
|
2018-03-16 17:36:32 -07:00
|
|
|
let rx = Arc::new(Mutex::new(rx));
|
2018-03-16 14:04:06 -07:00
|
|
|
let stage = Arc::clone(&broadcast_stage);
|
|
|
|
// Insert the transmit handle connected to this task into
|
|
|
|
// the shared list of senders.
|
|
|
|
stage.lock().unwrap().senders.push(tx);
|
|
|
|
let task = BroadcastTask::new(stream, rx, stage);
|
2018-03-16 11:12:14 -07:00
|
|
|
|
|
|
|
// TODO: spawn a thread for the connected socket
|
|
|
|
}
|
|
|
|
Err(e) => {
|
|
|
|
warn!("Failed to connect: {}", e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|