Ignore max-peers min-peers option when a user call the "connect" RPC

Author: majecty

network/src/p2p/handler.rs

@@ -59,9 +59,15 @@ const LAST_CONNECTION_TOKEN: TimerToken = FIRST_CONNECTION_TOKEN + MAX_CONNECTIO
 const CREATE_CONNECTIONS_TOKEN: TimerToken = 0;
 const PULL_CONNECTIONS_MS: u64 = 10 * 1000;
 
-#[derive(Clone, Debug, PartialOrd, PartialEq)]
+#[derive(Clone, Debug, PartialEq)]
+pub enum IgnoreConnectionLimit {
+    Ignore,
+    Not,
+}
+
+#[derive(Clone, Debug, PartialEq)]
 pub enum Message {
-    RequestConnection(SocketAddr),
+    RequestConnection(SocketAddr, IgnoreConnectionLimit),
 
     RequestNegotiation {
         node_id: NodeId,
@@ -355,7 +361,7 @@ impl IoHandler<Message> for Handler {
                     let count = (self.min_peers - number_of_connections + 1) / 2;
                     let addresses = manager.routing_table.unestablished_addresses(count);
                     for address in addresses {
-                        io.message(Message::RequestConnection(address))?;
+                        io.message(Message::RequestConnection(address, IgnoreConnectionLimit::Not))?;
                     }
                 }
                 Ok(())
@@ -366,12 +372,14 @@ impl IoHandler<Message> for Handler {
 
     fn message(&self, io: &IoContext<Message>, message: &Message) -> IoHandlerResult<()> {
         match message {
-            Message::RequestConnection(socket_address) => {
+            Message::RequestConnection(socket_address, ignore_connection_limit) => {
                 let mut manager = self.manager.lock();
-                let number_of_connections = manager.connections.len();
-                if self.max_peers <= manager.connections.len() {
-                    ctrace!(NET, "Already has maximum peers({})", number_of_connections);
-                    return Ok(())
+                if ignore_connection_limit == &IgnoreConnectionLimit::Not {
+                    let number_of_connections = manager.connections.len();
+                    if self.max_peers <= manager.connections.len() {
+                        ctrace!(NET, "Already has maximum peers({})", number_of_connections);
+                        return Ok(())
+                    }
                 }
 
                 ctrace!(NET, "Connecting to {:?}", socket_address);

network/src/p2p/mod.rs

@@ -21,7 +21,7 @@ mod listener;
 mod message;
 mod stream;
 
-pub use self::handler::{Handler, Message};
+pub use self::handler::{Handler, IgnoreConnectionLimit, Message};
 use self::message::ExtensionMessage;
 use self::message::NegotiationBody;
 use self::message::NegotiationMessage;

network/src/service.rs

@@ -58,7 +58,8 @@ impl Service {
 
         timer.register_handler(Arc::new(timer::Handler::new(Arc::clone(&client))))?;
 
-        let session_initiator_handler = Arc::new(session_initiator::Handler::new(address, Arc::clone(&routing_table)));
+        let session_initiator_handler =
+            Arc::new(session_initiator::Handler::new(address, Arc::clone(&routing_table), p2p.channel()));
         session_initiator.register_handler(session_initiator_handler)?;
 
         Ok(Arc::new(Self {

network/src/session_initiator/handler.rs

@@ -22,14 +22,15 @@ use std::sync::Arc;
 
 use ccrypto::aes::SymmetricCipherError;
 use cfinally::finally;
-use cio::{IoContext, IoError as CIoError, IoHandler, IoHandlerResult, IoManager, StreamToken, TimerToken};
+use cio::{IoChannel, IoContext, IoError as CIoError, IoHandler, IoHandlerResult, IoManager, StreamToken, TimerToken};
 use ckey::Error as KeysError;
 use ctypes::Secret;
 use mio::deprecated::EventLoop;
 use mio::Token;
 use parking_lot::Mutex;
 use rlp::DecoderError;
 
+use super::super::p2p;
 use super::super::token_generator::TokenGenerator;
 use super::super::RoutingTable;
 use super::super::SocketAddr;
@@ -90,6 +91,7 @@ struct SessionInitiator {
 
     routing_table: Arc<RoutingTable>,
     requests: Requests,
+    channel_to_p2p: IoChannel<p2p::Message>,
 }
 
 #[derive(Debug)]
@@ -191,12 +193,17 @@ pub enum Message {
 const MESSAGE_TIMEOUT_MS: u64 = 10_000;
 
 impl SessionInitiator {
-    fn bind(socket_address: &SocketAddr, routing_table: Arc<RoutingTable>) -> Result<Self> {
+    fn bind(
+        socket_address: &SocketAddr,
+        routing_table: Arc<RoutingTable>,
+        channel_to_p2p: IoChannel<p2p::Message>,
+    ) -> Result<Self> {
         let server = Server::bind(socket_address)?;
         Ok(Self {
             server,
             routing_table,
             requests: Requests::new(),
+            channel_to_p2p,
         })
     }
 
@@ -341,7 +348,10 @@ impl SessionInitiator {
                     return Ok(())
                 }
 
-                self.requests.manually_connected_address.take(from);
+                if self.requests.manually_connected_address.take(from).is_some() {
+                    self.channel_to_p2p
+                        .send(p2p::Message::RequestConnection(from.clone(), p2p::IgnoreConnectionLimit::Ignore))?;
+                }
 
                 if !self.routing_table.create_allowed_session(from, &encrypted_nonce) {
                     cwarn!(NET, "Cannot create session to {:?}", from);
@@ -376,9 +386,14 @@ pub struct Handler {
 }
 
 impl Handler {
-    pub fn new(socket_address: SocketAddr, routing_table: Arc<RoutingTable>) -> Self {
-        let session_initiator =
-            Mutex::new(SessionInitiator::bind(&socket_address, routing_table).expect("Cannot bind UDP port"));
+    pub fn new(
+        socket_address: SocketAddr,
+        routing_table: Arc<RoutingTable>,
+        channel_to_p2p: IoChannel<p2p::Message>,
+    ) -> Self {
+        let session_initiator = Mutex::new(
+            SessionInitiator::bind(&socket_address, routing_table, channel_to_p2p).expect("Cannot bind UDP port"),
+        );
         Self {
             session_initiator,
         }