diff --git a/Cargo.lock b/Cargo.lock
index 7ca56d078be092e336f6801d53242460651d80ec..ea1da06de14e68eeb2101a4531e0773b5ad14755 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -553,6 +553,7 @@ dependencies = [
"bincode 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
"dubp-documents 0.12.0",
"dup-crypto 0.6.0",
+ "durs-common-tests-tools 0.1.0",
"durs-common-tools 0.1.0",
"durs-conf 0.1.0-a0.1",
"durs-message 0.1.0-a0.1",
diff --git a/lib/modules/ws2p/ws2p-messages/lib.rs b/lib/modules/ws2p/ws2p-messages/lib.rs
index 8f439c3c629bef0560773f238f9212cc1ad470c1..3ca69abfde200b0c8b663f2e379b41f3595e59ce 100644
--- a/lib/modules/ws2p/ws2p-messages/lib.rs
+++ b/lib/modules/ws2p/ws2p-messages/lib.rs
@@ -80,7 +80,7 @@ impl WS2PMessage {
pub fn parse_and_check_bin_message(bin_msg: &[u8]) -> Result<WS2PMessage, WS2PMessageError> {
let msg: WS2PMessage = bincode::deserialize(&bin_msg)?;
let hash = msg.hash();
- //println!("DEBUG: parse_and_check_bin_message: hash={:?}", hash);
+ //debug!("parse_and_check_bin_message: hash={:?}", hash);
// Compute hash len
let hash_len = 33;
// Compute signature len
diff --git a/lib/modules/ws2p/ws2p-messages/v2/mod.rs b/lib/modules/ws2p/ws2p-messages/v2/mod.rs
index fc3f21d24c34aa713f0c1983419022e931c9dc9c..412d334594ccc12e236a13dc497905d1768ecb83 100644
--- a/lib/modules/ws2p/ws2p-messages/v2/mod.rs
+++ b/lib/modules/ws2p/ws2p-messages/v2/mod.rs
@@ -105,7 +105,9 @@ mod tests {
#[test]
fn test_ws2p_message_ack() {
- test_ws2p_message(WS2Pv2MessagePayload::Ack(Hash::random()));
+ test_ws2p_message(WS2Pv2MessagePayload::Ack {
+ challenge: Hash::random(),
+ });
}
#[test]
diff --git a/lib/modules/ws2p/ws2p-messages/v2/payload_container.rs b/lib/modules/ws2p/ws2p-messages/v2/payload_container.rs
index 92bce7660269b8696d9158a1c1d6627cbc50f250..8da911778eae86df49926af76aaea6100f40b88c 100644
--- a/lib/modules/ws2p/ws2p-messages/v2/payload_container.rs
+++ b/lib/modules/ws2p/ws2p-messages/v2/payload_container.rs
@@ -38,7 +38,10 @@ pub enum WS2Pv2MessagePayload {
/// CONNECT message
Connect(Box<WS2Pv2ConnectMsg>),
/// ACK message
- Ack(Hash),
+ Ack {
+ /// Hash previously sent in CONNECT message
+ challenge: Hash,
+ },
/// SECRET_FLAGS Message
SecretFlags(WS2Pv2SecretFlagsMsg),
/// OK Message
diff --git a/lib/modules/ws2p/ws2p/Cargo.toml b/lib/modules/ws2p/ws2p/Cargo.toml
index c35642daf719677ecce344cc20770e5403be2cce..b368910eafc892433a35ab8e34fe82f744007b76 100644
--- a/lib/modules/ws2p/ws2p/Cargo.toml
+++ b/lib/modules/ws2p/ws2p/Cargo.toml
@@ -29,5 +29,8 @@ serde_json = "1.0.*"
structopt= "0.2.*"
ws = { version = "0.7.*", features = ["permessage-deflate"] }
+[dev-dependencies]
+durs-common-tests-tools = { path = "../../../tests-tools/common-tests-tools" }
+
[features]
-ssl = ["ws/ssl"]
\ No newline at end of file
+ssl = ["ws/ssl"]
diff --git a/lib/modules/ws2p/ws2p/src/controllers/handler/ack_msg.rs b/lib/modules/ws2p/ws2p/src/controllers/handler/ack_msg.rs
new file mode 100644
index 0000000000000000000000000000000000000000..fa3e948be1e626f6022a99771c2893e9394e8461
--- /dev/null
+++ b/lib/modules/ws2p/ws2p/src/controllers/handler/ack_msg.rs
@@ -0,0 +1,91 @@
+// Copyright (C) 2018 The Durs Project Developers.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! Process WS2P ACK message.
+
+use crate::controllers::handler::*;
+use crate::controllers::*;
+use dup_crypto::hashs::Hash;
+use durs_common_tools::fatal_error;
+use durs_ws2p_messages::v2::ok::WS2Pv2OkMsg;
+use ws::CloseCode;
+
+/// Process WS2pv2 ACK Message
+pub fn process_ws2p_v2_ack_msg(
+ handler: &mut Ws2pConnectionHandler, // handler contains original challenge
+ ack_msg_challenge: Hash,
+) {
+ debug!("Receive ACK message !");
+
+ match handler.conn_datas.state {
+ WS2PConnectionState::OkMsgOkWaitingAckMsg => {
+ // already sent ack message and received ok response
+ process(handler, ack_msg_challenge, WS2PConnectionState::Established);
+ }
+ WS2PConnectionState::ConnectMessOk => {
+ // ack message not yet sent
+ process(handler, ack_msg_challenge, WS2PConnectionState::AckMsgOk);
+ }
+ _ => {
+ let _ = handler
+ .ws
+ .0
+ .close_with_reason(CloseCode::Invalid, "Unexpected ACK message !");
+ }
+ }
+}
+
+#[inline]
+// process and apply given status in case of success
+fn process(
+ handler: &mut Ws2pConnectionHandler,
+ ack_msg_challenge: Hash,
+ success_status: WS2PConnectionState,
+) {
+ if handler.conn_datas.challenge != ack_msg_challenge {
+ handler.update_status(WS2PConnectionState::Denial);
+ } else {
+ handler.update_status(success_status);
+ send_ok_msg(handler);
+ }
+}
+
+// send ok message
+fn send_ok_msg(handler: &mut Ws2pConnectionHandler) {
+ // generate empty Ok message
+ let ok_msg: WS2Pv2OkMsg = Default::default();
+
+ // Encapsulate and binarize OK message
+ if let Ok((_, bin_ok_msg)) = WS2Pv2Message::encapsulate_payload(
+ handler.currency.clone(),
+ handler.local_node.my_node_id,
+ handler.local_node.my_key_pair,
+ WS2Pv2MessagePayload::Ok(ok_msg),
+ ) {
+ // Send Ok Message
+ match handler.ws.0.send(Message::binary(bin_ok_msg)) {
+ Ok(()) => {}
+ Err(_) => {
+ handler.conn_datas.state = WS2PConnectionState::Unreachable;
+ let _ = handler
+ .ws
+ .0
+ .close_with_reason(CloseCode::Error, "Fail to send Ok message !");
+ }
+ }
+ } else {
+ fatal_error!("Dev error: Fail to sign own ok message !");
+ }
+}
diff --git a/lib/modules/ws2p/ws2p/src/controllers/handler/connect_msg.rs b/lib/modules/ws2p/ws2p/src/controllers/handler/connect_msg.rs
index 7ba8205d57ecbed4870f1931bc613b92d20a357e..e1ce00566a5901d78f163de0b07449bacad1619c 100644
--- a/lib/modules/ws2p/ws2p/src/controllers/handler/connect_msg.rs
+++ b/lib/modules/ws2p/ws2p/src/controllers/handler/connect_msg.rs
@@ -13,10 +13,11 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.
-//! Process WS2P CONNECT mesage.
+//! Process WS2P CONNECT message.
use crate::controllers::handler::*;
use crate::controllers::*;
+use durs_common_tools::fatal_error;
use durs_network_documents::NodeFullId;
use durs_ws2p_messages::v2::connect::WS2Pv2ConnectMsg;
use ws::CloseCode;
@@ -29,7 +30,7 @@ pub fn process_ws2p_v2_connect_msg(
remote_full_id: NodeFullId,
connect_msg: &WS2Pv2ConnectMsg,
) {
- println!("DEBUG: Receive CONNECT message !");
+ debug!("Receive CONNECT message !");
// Get remote node datas
let remote_challenge = connect_msg.challenge;
@@ -38,7 +39,7 @@ pub fn process_ws2p_v2_connect_msg(
peer_card: None,
current_blockstamp: None,
};
- if let WS2PConnectionState::WaitingConnectMess = handler.conn_datas.state {
+ if let WS2PConnectionState::WaitingConnectMsg = handler.conn_datas.state {
// Check remote node datas
if let WS2Pv2ConnectType::Incoming = handler.conn_datas.connect_type {
handler.conn_datas.remote_full_id = Some(remote_full_id);
@@ -97,25 +98,26 @@ pub fn process_ws2p_v2_connect_msg(
handler.send_new_conn_state_to_service();
// Encapsulate and binarize ACK message
- let (_, bin_ack_msg) = WS2Pv2Message::encapsulate_payload(
+ if let Ok((_, bin_ack_msg)) = WS2Pv2Message::encapsulate_payload(
handler.currency.clone(),
handler.local_node.my_node_id,
handler.local_node.my_key_pair,
- WS2Pv2MessagePayload::Ack(remote_challenge),
- )
- .expect("WS2P : Fail to sign own ack message !");
- // Send ACk Message
- match handler.ws.0.send(Message::binary(bin_ack_msg)) {
- Ok(()) => {
+ WS2Pv2MessagePayload::Ack {
+ challenge: remote_challenge,
+ },
+ ) {
+ // Send Ack Message
+ if let Ok(()) = handler.ws.0.send(Message::binary(bin_ack_msg)) {
// Update state
handler.conn_datas.state = WS2PConnectionState::ConnectMessOk;
- }
- Err(_) => {
+ } else {
handler.conn_datas.state = WS2PConnectionState::Unreachable;
let _ = handler
.ws
.0
- .close_with_reason(CloseCode::Error, "Fail to send ACk message !");
+ .close_with_reason(CloseCode::Error, "Fail to send Ack message !");
}
+ } else {
+ fatal_error!("Dev error: Fail to sign own ack message !")
}
}
diff --git a/lib/modules/ws2p/ws2p/src/controllers/handler/mod.rs b/lib/modules/ws2p/ws2p/src/controllers/handler/mod.rs
index 45e9fc97497ca44c67a6a72a4d79bd4388e5af2d..a30a4ceff07cb30a70c67e0c7d5e7321601c8b7b 100644
--- a/lib/modules/ws2p/ws2p/src/controllers/handler/mod.rs
+++ b/lib/modules/ws2p/ws2p/src/controllers/handler/mod.rs
@@ -15,17 +15,17 @@
//! WS2P connection handler.
+pub mod ack_msg;
pub mod connect_msg;
+pub mod ok_msg;
use crate::constants::*;
use crate::controllers::*;
+use crate::services::Ws2pServiceSender;
use crate::services::*;
-use ws::{util::Token, CloseCode, /*Frame,*/ Handler, Handshake, Message};
-//use dup_crypto::keys::KeyPairEnum;
use dubp_documents::CurrencyName;
+use durs_common_tools::fatal_error;
use durs_network_documents::NodeFullId;
-//use durs_ws2p_messages::v2::api_features::WS2PFeatures;
-use crate::services::Ws2pServiceSender;
use durs_ws2p_messages::v2::connect::generate_connect_message;
use durs_ws2p_messages::v2::payload_container::WS2Pv2MessagePayload;
use durs_ws2p_messages::v2::WS2Pv2Message;
@@ -35,6 +35,7 @@ use std::ops::Deref;
use std::sync::mpsc;
use std::thread;
use std::time::Duration;
+use ws::{util::Token, CloseCode, Handler, Handshake, Message};
const CONNECT: Token = Token(1);
const EXPIRE: Token = Token(2);
@@ -77,7 +78,8 @@ impl Ws2pConnectionHandler {
) = mpsc::channel();
// Send controller sender to service
- println!("DEBUG: Send controller sender to service");
+ debug!("Send controller sender to service");
+
service_sender.send(Ws2pServiceSender::ControllerSender(sender))?;
Ok(Ws2pConnectionHandler {
@@ -103,6 +105,11 @@ impl Ws2pConnectionHandler {
))
.expect("WS2p Service unreacheable !");
}
+ #[inline]
+ fn update_status(&mut self, status: WS2PConnectionState) {
+ self.conn_datas.state = status;
+ self.send_new_conn_state_to_service();
+ }
}
fn print_opt_addr(addr: Option<SocketAddr>) -> String {
@@ -121,14 +128,13 @@ impl Handler for Ws2pConnectionHandler {
// Handler state or reject the connection based on the details of the Request
// or Response, such as by checking cookies or Auth headers.
fn on_open(&mut self, handshake: Handshake) -> ws::Result<()> {
- #[cfg(test)]
- println!(
- "TESTS: open websocket from {}",
+ debug!(
+ "open websocket from {}",
print_opt_addr(handshake.peer_addr)
);
// Update connection state
- self.conn_datas.state = WS2PConnectionState::TryToSendConnectMess;
+ self.conn_datas.state = WS2PConnectionState::TryToSendConnectMsg;
self.send_new_conn_state_to_service();
// Generate connect message
@@ -140,77 +146,58 @@ impl Handler for Ws2pConnectionHandler {
);
// Encapsulate and binarize connect message
- let (_ws2p_full_msg, bin_connect_msg) = WS2Pv2Message::encapsulate_payload(
+ if let Ok((_ws2p_full_msg, bin_connect_msg)) = WS2Pv2Message::encapsulate_payload(
self.currency.clone(),
self.local_node.my_node_id,
self.local_node.my_key_pair,
WS2Pv2MessagePayload::Connect(Box::new(connect_msg)),
- )
- .expect("WS2P : Fail to sign own connect message !");
-
- // Start negociation timeouts
- self.ws.0.timeout(*WS2P_NEGOTIATION_TIMEOUT, CONNECT)?;
- // Start expire timeout
- self.ws
- .0
- .timeout(*WS2P_EXPIRE_TIMEOUT_IN_SECS * 1_000, EXPIRE)?;
+ ) {
+ // Start negociation timeouts
+ self.ws.0.timeout(*WS2P_NEGOTIATION_TIMEOUT, CONNECT)?;
+ // Start expire timeout
+ self.ws
+ .0
+ .timeout(*WS2P_EXPIRE_TIMEOUT_IN_SECS * 1_000, EXPIRE)?;
- // Send connect message
- match self.ws.0.send(Message::binary(bin_connect_msg)) {
- Ok(()) => {
- // Update state
- if let WS2PConnectionState::TryToSendConnectMess = self.conn_datas.state {
- self.conn_datas.state = WS2PConnectionState::WaitingConnectMess;
- self.send_new_conn_state_to_service();
+ // Send connect message
+ match self.ws.0.send(Message::binary(bin_connect_msg)) {
+ Ok(()) => {
+ // Update state
+ if let WS2PConnectionState::TryToSendConnectMsg = self.conn_datas.state {
+ self.conn_datas.state = WS2PConnectionState::WaitingConnectMsg;
+ self.send_new_conn_state_to_service();
+ }
+ // Log
+ info!(
+ "Send CONNECT message to {}",
+ print_opt_addr(handshake.peer_addr)
+ );
+ debug!(
+ "Succesfully send CONNECT message to {}",
+ print_opt_addr(handshake.peer_addr)
+ );
+ }
+ Err(e) => {
+ self.conn_datas.state = WS2PConnectionState::Unreachable;
+ warn!(
+ "Fail to send CONNECT message to {} : {}",
+ print_opt_addr(handshake.peer_addr),
+ e
+ );
+ debug!(
+ "Fail send CONNECT message to {}",
+ print_opt_addr(handshake.peer_addr)
+ );
+ let _ = self
+ .ws
+ .0
+ .close_with_reason(CloseCode::Error, "Fail to send CONNECT message !");
}
- // Log
- info!(
- "Send CONNECT message to {}",
- print_opt_addr(handshake.peer_addr)
- );
- #[cfg(test)]
- println!(
- "TESTS: Succesfully send CONNECT message to {}",
- print_opt_addr(handshake.peer_addr)
- );
- }
- Err(e) => {
- self.conn_datas.state = WS2PConnectionState::Unreachable;
- warn!(
- "Fail to send CONNECT message to {} : {}",
- print_opt_addr(handshake.peer_addr),
- e
- );
- #[cfg(test)]
- println!(
- "TESTS: Fail send CONNECT message to {}",
- print_opt_addr(handshake.peer_addr)
- );
- let _ = self
- .ws
- .0
- .close_with_reason(CloseCode::Error, "Fail to send CONNECT message !");
}
- }
- /*
- // Send ws::Sender to WS2PConductor
- let result = self
- .conductor_sender
- .send(WS2PThreadSignal::WS2PConnectionMessage(
- WS2PConnectionMessage(
- self.conn_datas.node_full_id(),
- WS2PConnectionMessagePayload::WebsocketOk(WsSender(self.ws.clone())),
- ),
- ));
- // If WS2PConductor is unrechable, close connection.
- if result.is_err() {
- debug!("Close ws2p connection because ws2p main thread is unrechable !");
- self.ws.close(CloseCode::Normal)
+ Ok(())
} else {
- // Send CONNECT Message
- self.ws.send(self.connect_message.clone())
- }*/
- Ok(())
+ fatal_error!("Dev error: Fail to sign own connect message !");
+ }
}
// `on_message` is roughly equivalent to the Handler closure. It takes a `Message`
@@ -243,7 +230,7 @@ impl Handler for Ws2pConnectionHandler {
self.conn_datas.last_mess_time = SystemTime::now();
if msg.is_binary() {
- println!("DEBUG: Receive new message there is not a spam !");
+ debug!("Receive new message there is not a spam !");
match WS2PMessage::parse_and_check_bin_message(&msg.into_data()) {
Ok(valid_msg) => match valid_msg {
WS2PMessage::V2(msg_v2) => {
@@ -260,9 +247,17 @@ impl Handler for Ws2pConnectionHandler {
connect_msg,
);
}
- WS2Pv2MessagePayload::Ack(_) => {}
+ WS2Pv2MessagePayload::Ack {
+ challenge: ack_msg_challenge,
+ } => {
+ // Process ack message
+ ack_msg::process_ws2p_v2_ack_msg(self, ack_msg_challenge);
+ }
WS2Pv2MessagePayload::SecretFlags(_) => {}
- WS2Pv2MessagePayload::Ok(_) => {}
+ WS2Pv2MessagePayload::Ok(_) => {
+ // Process ok message
+ ok_msg::process_ws2p_v2_ok_msg(self);
+ }
WS2Pv2MessagePayload::Ko(_) => {}
_ => {
if let WS2PConnectionState::Established = self.conn_datas.state {
@@ -278,7 +273,7 @@ impl Handler for Ws2pConnectionHandler {
}
},
Err(ws2p_msg_err) => {
- println!("DEBUG: Message is invalid : {:?}", ws2p_msg_err);
+ warn!("Message is invalid : {:?}", ws2p_msg_err);
self.count_invalid_msgs += 1;
if self.count_invalid_msgs >= *WS2P_INVALID_MSGS_LIMIT {
let _ = self.ws.0.close_with_reason(
diff --git a/lib/modules/ws2p/ws2p/src/controllers/handler/ok_msg.rs b/lib/modules/ws2p/ws2p/src/controllers/handler/ok_msg.rs
new file mode 100644
index 0000000000000000000000000000000000000000..a59b596b6f73f4f70424bc424d92f95a1918a578
--- /dev/null
+++ b/lib/modules/ws2p/ws2p/src/controllers/handler/ok_msg.rs
@@ -0,0 +1,40 @@
+// Copyright (C) 2018 The Durs Project Developers.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! Process WS2P OK message.
+
+use crate::controllers::handler::*;
+use crate::controllers::*;
+use ws::CloseCode;
+
+/// Process WS2pv2 OK Message
+pub fn process_ws2p_v2_ok_msg(handler: &mut Ws2pConnectionHandler) {
+ debug!("Receive OK message !");
+
+ match handler.conn_datas.state {
+ WS2PConnectionState::ConnectMessOk => {
+ handler.update_status(WS2PConnectionState::OkMsgOkWaitingAckMsg);
+ }
+ WS2PConnectionState::AckMsgOk => {
+ handler.update_status(WS2PConnectionState::Established);
+ }
+ _ => {
+ let _ = handler
+ .ws
+ .0
+ .close_with_reason(CloseCode::Invalid, "Unexpected OK message !");
+ }
+ }
+}
diff --git a/lib/modules/ws2p/ws2p/src/controllers/incoming_connections/mod.rs b/lib/modules/ws2p/ws2p/src/controllers/incoming_connections/mod.rs
index 702e97295a0187b406eea7b55b4760609e695e83..81797f1c1014b4e63a118d6be3554aa3afcb760c 100644
--- a/lib/modules/ws2p/ws2p/src/controllers/incoming_connections/mod.rs
+++ b/lib/modules/ws2p/ws2p/src/controllers/incoming_connections/mod.rs
@@ -39,11 +39,10 @@ pub fn listen_on_ws2p_v2_endpoint(
// Log
info!("Listen on {} ...", ws_url);
- println!("DEBUG: call function listen({}) ...", ws_url);
// Connect to websocket
listen(ws_url, move |ws| {
- println!("DEBUG: Listen on host:port...");
+ info!("Listen on host:port...");
DeflateBuilder::new().build(
Ws2pConnectionHandler::try_new(
WsSender(ws),
diff --git a/lib/modules/ws2p/ws2p/src/controllers/mod.rs b/lib/modules/ws2p/ws2p/src/controllers/mod.rs
index 602c7731f721026ed7cf1cb535801e6195ad8fdf..778a4e485bdee3bf8af60e565b1707a5a3d69265 100644
--- a/lib/modules/ws2p/ws2p/src/controllers/mod.rs
+++ b/lib/modules/ws2p/ws2p/src/controllers/mod.rs
@@ -60,11 +60,11 @@ pub enum WS2PConnectionState {
/// Websocket error
WSError,
/// Try to send connect message
- TryToSendConnectMess,
+ TryToSendConnectMsg,
/// Endpoint unreachable
Unreachable,
/// Waiting connect message
- WaitingConnectMess,
+ WaitingConnectMsg,
/// No response
NoResponse,
/// Negociation timeout
@@ -72,14 +72,14 @@ pub enum WS2PConnectionState {
/// Receive valid connect message
ConnectMessOk,
/// Receive valid OK message but wait ACK message
- OkMessOkWaitingAckMess,
+ OkMsgOkWaitingAckMsg,
/// Receive valid ACK message
- AckMessOk,
+ AckMsgOk,
/// Connection denial (maybe due to many different reasons : receive wrong message, wrong format, wrong signature, etc)
Denial,
/// Connection closed
Close,
- /// Connection succesfully established
+ /// Connection successfully established
Established,
}
diff --git a/lib/modules/ws2p/ws2p/src/controllers/outgoing_connections/mod.rs b/lib/modules/ws2p/ws2p/src/controllers/outgoing_connections/mod.rs
index dd186d3e61af9d9d3dbf326b5becf261b0164bb0..ec854a85af66db401f0492b537ef69cc7f4e04e3 100644
--- a/lib/modules/ws2p/ws2p/src/controllers/outgoing_connections/mod.rs
+++ b/lib/modules/ws2p/ws2p/src/controllers/outgoing_connections/mod.rs
@@ -15,11 +15,11 @@
//! WS2P outgoing connections controllers.
-use dubp_documents::CurrencyName;
-//use durs_module::ModuleReqId;
use crate::controllers::handler::Ws2pConnectionHandler;
use crate::controllers::*;
use crate::services::*;
+use dubp_documents::CurrencyName;
+use durs_common_tools::fatal_error;
use durs_network_documents::network_endpoint::EndpointEnum;
use durs_network_documents::NodeFullId;
use ws::connect;
@@ -47,19 +47,18 @@ pub fn connect_to_ws2p_v2_endpoint(
// Log
info!("Try connection to {} ...", ws_url);
- println!("DEBUG: Try connection to {} ...", ws_url);
// Connect to websocket
connect(ws_url, move |ws| {
- DeflateBuilder::new().build(
- Ws2pConnectionHandler::try_new(
- WsSender(ws),
- service_sender.clone(),
- currency.clone(),
- self_node.clone(),
- conn_meta_datas.clone(),
- )
- .expect("WS2P Service unrechable"),
- )
+ match Ws2pConnectionHandler::try_new(
+ WsSender(ws),
+ service_sender.clone(),
+ currency.clone(),
+ self_node.clone(),
+ conn_meta_datas.clone(),
+ ) {
+ Ok(handler) => DeflateBuilder::new().build(handler),
+ Err(_e) => fatal_error!("WS2P Service unreachable"),
+ }
})
}
diff --git a/lib/modules/ws2p/ws2p/tests/connection_negociation.rs b/lib/modules/ws2p/ws2p/tests/connection_negociation.rs
index e5443a3caede90701b97adda84eae8c2fd1d08df..7006a1b2247a36f938055dc407d9b07b48aa7786 100644
--- a/lib/modules/ws2p/ws2p/tests/connection_negociation.rs
+++ b/lib/modules/ws2p/ws2p/tests/connection_negociation.rs
@@ -16,6 +16,7 @@
use dubp_documents::CurrencyName;
use dup_crypto::keys::KeyPair;
use dup_crypto::keys::*;
+use durs_common_tests_tools::logger::init_logger_stdout;
use durs_network_documents::network_endpoint::*;
use durs_network_documents::*;
use durs_ws2p::controllers::incoming_connections::*;
@@ -36,15 +37,15 @@ pub fn currency() -> CurrencyName {
pub fn keypair1() -> ed25519::KeyPair {
ed25519::KeyPairFromSaltedPasswordGenerator::with_default_parameters().generate(
- "JhxtHB7UcsDbA9wMSyMKXUzBZUQvqVyB32KwzS9SWoLkjrUhHV".as_bytes(),
- "JhxtHB7UcsDbA9wMSyMKXUzBZUQvqVyB32KwzS9SWoLkjrUhHV_".as_bytes(),
+ "JhxtHB7UcsDbA9wMSyMKXUzBZUQvqVyB32KwzS9SWoLkjrUhHV1".as_bytes(),
+ "JhxtHB7UcsDbA9wMSyMKXUzBZUQvqVyB32KwzS9SWoLkjrUhHV1_".as_bytes(),
)
}
pub fn keypair2() -> ed25519::KeyPair {
ed25519::KeyPairFromSaltedPasswordGenerator::with_default_parameters().generate(
- "JhxtHB7UcsDbA9wMSyMKXUzBZUQvqVyB32KwzS9SWdLkjrUhHV".as_bytes(),
- "JhxtHB7UcsDbA9wMSyMKXUzBZUQvqVyB32KwzS9SWdLkjrUhHV_".as_bytes(),
+ "JhxtHB7UcsDbA9wMSyMKXUzBZUQvqVyB32KwzS9SWdLkjrUhHV2".as_bytes(),
+ "JhxtHB7UcsDbA9wMSyMKXUzBZUQvqVyB32KwzS9SWdLkjrUhHV2_".as_bytes(),
)
}
@@ -68,6 +69,11 @@ fn client_node() -> MySelfWs2pNode {
#[test]
#[cfg(unix)]
fn test_connection_negociation() {
+ init_logger_stdout();
+
+ // ===== initialization =====
+ // client and server are initialized and launched in separate threads
+
let server_node = server_node();
let client_node = client_node();
@@ -116,70 +122,81 @@ fn test_connection_negociation() {
)
});
- // Listen client service channel : we must receive controller sender
- if let Ok(Ws2pServiceSender::ControllerSender(_)) = client_service_channel
- .1
- .recv_timeout(Duration::from_millis(*TIMEOUT_IN_MS))
- {
- } else {
- panic!("Not receive client controller sender");
- }
-
- // Listen client service channel : we must receive status TryToSendConnectMess
- test_expected_states(
- &client_service_channel.1,
- vec![WS2PConnectionState::TryToSendConnectMess],
+ // ===== opening connection =====
+ // we must get Ws2pServiceSender::ControllerSender from the client and server threads (but we ignore them)
+ // we also test that the statuses match expected ones
+
+ let _client_controller = get_controller(&client_service_channel.1);
+ let _server_controller = get_controller(&server_service_channel.1);
+
+ // TryToSendConnectMsg
+ let state = get_state(&client_service_channel.1); // client
+ assert!(state == WS2PConnectionState::TryToSendConnectMsg);
+ let state = get_state(&server_service_channel.1); // server
+ assert!(state == WS2PConnectionState::TryToSendConnectMsg);
+
+ // WaitingConnectMsg
+ let state = get_state(&client_service_channel.1); // client
+ assert!(state == WS2PConnectionState::WaitingConnectMsg);
+ let state = get_state(&server_service_channel.1); // server
+ assert!(state == WS2PConnectionState::WaitingConnectMsg);
+
+ // ConnectMessOk
+ let state = get_state(&client_service_channel.1); // client
+ assert!(state == WS2PConnectionState::ConnectMessOk);
+ let state = get_state(&server_service_channel.1); // server
+ assert!(state == WS2PConnectionState::ConnectMessOk);
+
+ // Ack message
+ let state_1 = get_state(&client_service_channel.1); // client
+ let state_2 = get_state(&server_service_channel.1); // server
+
+ println!("state_1: {:?}", &state_1);
+ println!("state_2: {:?}", &state_2);
+
+ assert!(
+ // client faster
+ ( state_1 == WS2PConnectionState::OkMsgOkWaitingAckMsg &&
+ state_2 == WS2PConnectionState::AckMsgOk ) ||
+ // server faster
+ ( state_1 == WS2PConnectionState::AckMsgOk &&
+ state_2 == WS2PConnectionState::OkMsgOkWaitingAckMsg ) ||
+ // ack messages received at the same time
+ ( state_1 == WS2PConnectionState::AckMsgOk &&
+ state_2 == WS2PConnectionState::AckMsgOk )
);
- // Listen client service channel : we must receive status WaitingConnectMess
- test_expected_states(
- &client_service_channel.1,
- vec![WS2PConnectionState::WaitingConnectMess],
- );
+ // Established
+ let state = get_state(&client_service_channel.1); // client
+ assert!(state == WS2PConnectionState::Established);
+ let state = get_state(&server_service_channel.1); // server
+ assert!(state == WS2PConnectionState::Established);
+}
- // Listen server service channel : we must receive controller sender
- if let Ok(Ws2pServiceSender::ControllerSender(_)) = server_service_channel
- .1
+// === functions used in above test ===
+
+// get the state in a receiver
+fn get_state(service_receiver: &mpsc::Receiver<Ws2pServiceSender>) -> WS2PConnectionState {
+ if let Ws2pServiceSender::ChangeConnectionState(_, new_state) = service_receiver
.recv_timeout(Duration::from_millis(*TIMEOUT_IN_MS))
+ .expect("Receive nothing from controller :")
{
+ return new_state;
} else {
- panic!("Not receive server controller sender");
+ panic!("Expect signal ChangeConnectionState, receive other !");
}
-
- // Listen server service channel : we must receive status TryToSendConnectMess
- test_expected_states(
- &server_service_channel.1,
- vec![WS2PConnectionState::TryToSendConnectMess],
- );
-
- // Listen server service channel : we must receive status WaitingConnectMess
- test_expected_states(
- &server_service_channel.1,
- vec![WS2PConnectionState::WaitingConnectMess],
- );
-
- // Listen server service channel : we must receive status ConnectMessOk
- test_expected_states(
- &server_service_channel.1,
- vec![WS2PConnectionState::ConnectMessOk],
- );
}
-fn test_expected_states(
+// get the controller from the thread
+fn get_controller(
service_receiver: &mpsc::Receiver<Ws2pServiceSender>,
- expected_states: Vec<WS2PConnectionState>,
-) -> WS2PConnectionState {
- if let Ws2pServiceSender::ChangeConnectionState(_, new_state) = service_receiver
- .recv_timeout(Duration::from_millis(*TIMEOUT_IN_MS))
- .expect("Receive nothing from controller :")
+) -> mpsc::Sender<Ws2pControllerOrder> {
+ // we must receive controller sender
+ if let Ok(Ws2pServiceSender::ControllerSender(controller)) =
+ service_receiver.recv_timeout(Duration::from_millis(*TIMEOUT_IN_MS))
{
- for expected_state in expected_states {
- if new_state == expected_state {
- return new_state;
- }
- }
- panic!("Receive unexpected state: {:?} !", new_state);
+ return controller;
} else {
- panic!("Expect signal ChangeConnectionState, receive other !");
+ panic!("Not receive client controller sender");
}
}