mirror of
https://github.com/YGGverse/aquatic.git
synced 2026-04-03 03:05:31 +00:00
aquatic_ws: remove DebugCallback and lots of debug output
Some error messages are still printed, though
This commit is contained in:
parent
cb0c299298
commit
b42ff9a23b
5 changed files with 19 additions and 56 deletions
1
TODO.md
1
TODO.md
|
|
@ -2,7 +2,6 @@
|
||||||
|
|
||||||
## aquatic_ws
|
## aquatic_ws
|
||||||
* ipv4 / ipv6 split state?
|
* ipv4 / ipv6 split state?
|
||||||
* remove debug prints and DebugCallback
|
|
||||||
* network
|
* network
|
||||||
* handle tls certificate parse errors etc better
|
* handle tls certificate parse errors etc better
|
||||||
* parse once only
|
* parse once only
|
||||||
|
|
|
||||||
|
|
@ -7,29 +7,12 @@ use mio::Token;
|
||||||
use mio::net::TcpStream;
|
use mio::net::TcpStream;
|
||||||
use native_tls::{TlsAcceptor, TlsStream, MidHandshakeTlsStream};
|
use native_tls::{TlsAcceptor, TlsStream, MidHandshakeTlsStream};
|
||||||
use tungstenite::WebSocket;
|
use tungstenite::WebSocket;
|
||||||
use tungstenite::handshake::{MidHandshake, HandshakeError};
|
use tungstenite::handshake::{MidHandshake, HandshakeError, server::NoCallback};
|
||||||
use tungstenite::server::ServerHandshake;
|
use tungstenite::server::{ServerHandshake};
|
||||||
|
|
||||||
use crate::common::*;
|
use crate::common::*;
|
||||||
|
|
||||||
|
|
||||||
#[derive(Clone, Copy, Debug)]
|
|
||||||
pub struct DebugCallback;
|
|
||||||
|
|
||||||
impl ::tungstenite::handshake::server::Callback for DebugCallback {
|
|
||||||
fn on_request(
|
|
||||||
self,
|
|
||||||
request: &::tungstenite::handshake::server::Request,
|
|
||||||
response: ::tungstenite::handshake::server::Response,
|
|
||||||
) -> Result<::tungstenite::handshake::server::Response, ::tungstenite::handshake::server::ErrorResponse> {
|
|
||||||
println!("request: {:#?}", request);
|
|
||||||
println!("response: {:#?}", response);
|
|
||||||
|
|
||||||
Ok(response)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
pub enum Stream {
|
pub enum Stream {
|
||||||
TcpStream(TcpStream),
|
TcpStream(TcpStream),
|
||||||
TlsStream(TlsStream<TcpStream>),
|
TlsStream(TlsStream<TcpStream>),
|
||||||
|
|
@ -105,7 +88,7 @@ enum HandshakeMachine {
|
||||||
TcpStream(TcpStream),
|
TcpStream(TcpStream),
|
||||||
TlsStream(TlsStream<TcpStream>),
|
TlsStream(TlsStream<TcpStream>),
|
||||||
TlsMidHandshake(MidHandshakeTlsStream<TcpStream>),
|
TlsMidHandshake(MidHandshakeTlsStream<TcpStream>),
|
||||||
WsMidHandshake(MidHandshake<ServerHandshake<Stream, DebugCallback>>),
|
WsMidHandshake(MidHandshake<ServerHandshake<Stream, NoCallback>>),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -127,18 +110,16 @@ impl HandshakeMachine {
|
||||||
tls_acceptor.accept(stream)
|
tls_acceptor.accept(stream)
|
||||||
)
|
)
|
||||||
} else {
|
} else {
|
||||||
let handshake_result = ::tungstenite::server::accept_hdr(
|
let handshake_result = ::tungstenite::server::accept(
|
||||||
Stream::TcpStream(stream),
|
Stream::TcpStream(stream),
|
||||||
DebugCallback
|
|
||||||
);
|
);
|
||||||
|
|
||||||
Self::handle_ws_handshake_result(handshake_result)
|
Self::handle_ws_handshake_result(handshake_result)
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
HandshakeMachine::TlsStream(stream) => {
|
HandshakeMachine::TlsStream(stream) => {
|
||||||
let handshake_result = ::tungstenite::server::accept_hdr(
|
let handshake_result = ::tungstenite::server::accept(
|
||||||
Stream::TlsStream(stream),
|
Stream::TlsStream(stream),
|
||||||
DebugCallback
|
|
||||||
);
|
);
|
||||||
|
|
||||||
Self::handle_ws_handshake_result(handshake_result)
|
Self::handle_ws_handshake_result(handshake_result)
|
||||||
|
|
@ -158,17 +139,13 @@ impl HandshakeMachine {
|
||||||
) -> (Option<Either<EstablishedWs, Self>>, bool) {
|
) -> (Option<Either<EstablishedWs, Self>>, bool) {
|
||||||
match result {
|
match result {
|
||||||
Ok(stream) => {
|
Ok(stream) => {
|
||||||
println!("handshake established");
|
|
||||||
|
|
||||||
(Some(Either::Right(Self::TlsStream(stream))), false)
|
(Some(Either::Right(Self::TlsStream(stream))), false)
|
||||||
},
|
},
|
||||||
Err(native_tls::HandshakeError::WouldBlock(handshake)) => {
|
Err(native_tls::HandshakeError::WouldBlock(handshake)) => {
|
||||||
println!("interrupted");
|
|
||||||
|
|
||||||
(Some(Either::Right(Self::TlsMidHandshake(handshake))), true)
|
(Some(Either::Right(Self::TlsMidHandshake(handshake))), true)
|
||||||
},
|
},
|
||||||
Err(native_tls::HandshakeError::Failure(err)) => {
|
Err(native_tls::HandshakeError::Failure(err)) => {
|
||||||
dbg!(err);
|
eprintln!("tls handshake error: {}", err);
|
||||||
|
|
||||||
(None, false)
|
(None, false)
|
||||||
}
|
}
|
||||||
|
|
@ -177,12 +154,10 @@ impl HandshakeMachine {
|
||||||
|
|
||||||
#[inline]
|
#[inline]
|
||||||
fn handle_ws_handshake_result(
|
fn handle_ws_handshake_result(
|
||||||
result: Result<WebSocket<Stream>, HandshakeError<ServerHandshake<Stream, DebugCallback>>> ,
|
result: Result<WebSocket<Stream>, HandshakeError<ServerHandshake<Stream, NoCallback>>> ,
|
||||||
) -> (Option<Either<EstablishedWs, Self>>, bool) {
|
) -> (Option<Either<EstablishedWs, Self>>, bool) {
|
||||||
match result {
|
match result {
|
||||||
Ok(mut ws) => {
|
Ok(mut ws) => {
|
||||||
println!("handshake established");
|
|
||||||
|
|
||||||
let peer_addr = ws.get_mut().get_peer_addr();
|
let peer_addr = ws.get_mut().get_peer_addr();
|
||||||
|
|
||||||
let established_ws = EstablishedWs {
|
let established_ws = EstablishedWs {
|
||||||
|
|
@ -193,12 +168,10 @@ impl HandshakeMachine {
|
||||||
(Some(Either::Left(established_ws)), false)
|
(Some(Either::Left(established_ws)), false)
|
||||||
},
|
},
|
||||||
Err(HandshakeError::Interrupted(handshake)) => {
|
Err(HandshakeError::Interrupted(handshake)) => {
|
||||||
println!("interrupted");
|
|
||||||
|
|
||||||
(Some(Either::Right(HandshakeMachine::WsMidHandshake(handshake))), true)
|
(Some(Either::Right(HandshakeMachine::WsMidHandshake(handshake))), true)
|
||||||
},
|
},
|
||||||
Err(HandshakeError::Failure(err)) => {
|
Err(HandshakeError::Failure(err)) => {
|
||||||
dbg!(err);
|
eprintln!("ws handshake error: {}", err);
|
||||||
|
|
||||||
(None, false)
|
(None, false)
|
||||||
}
|
}
|
||||||
|
|
@ -268,9 +241,12 @@ impl Connection {
|
||||||
if ews.ws.can_read(){
|
if ews.ws.can_read(){
|
||||||
ews.ws.close(None).unwrap();
|
ews.ws.close(None).unwrap();
|
||||||
|
|
||||||
// Needs to be done after ws.close()
|
// Required after ws.close()
|
||||||
if let Err(err) = ews.ws.write_pending(){
|
if let Err(err) = ews.ws.write_pending(){
|
||||||
dbg!(err);
|
eprintln!(
|
||||||
|
"error writing pending messages after closing ws: {}",
|
||||||
|
err
|
||||||
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -124,7 +124,7 @@ fn accept_new_streams(
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
eprint!("{}", err);
|
eprint!("error while accepting streams: {}", err);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -149,14 +149,10 @@ pub fn run_handshakes_and_read_messages(
|
||||||
|
|
||||||
match established_ws.ws.read_message(){
|
match established_ws.ws.read_message(){
|
||||||
Ok(ws_message) => {
|
Ok(ws_message) => {
|
||||||
dbg!(ws_message.clone());
|
|
||||||
|
|
||||||
if let Some(in_message) = InMessage::from_ws_message(ws_message){
|
if let Some(in_message) = InMessage::from_ws_message(ws_message){
|
||||||
dbg!(in_message.clone());
|
|
||||||
|
|
||||||
let meta = ConnectionMeta {
|
let meta = ConnectionMeta {
|
||||||
worker_index: socket_worker_index,
|
worker_index: socket_worker_index,
|
||||||
poll_token: poll_token,
|
poll_token,
|
||||||
peer_addr: established_ws.peer_addr
|
peer_addr: established_ws.peer_addr
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
@ -164,10 +160,10 @@ pub fn run_handshakes_and_read_messages(
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
Err(Io(err)) if err.kind() == ErrorKind::WouldBlock => {
|
Err(Io(err)) if err.kind() == ErrorKind::WouldBlock => {
|
||||||
break
|
break;
|
||||||
}
|
}
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
dbg!(err);
|
eprintln!("error reading messages: {}", err);
|
||||||
|
|
||||||
remove_connection_if_exists(connections, poll_token);
|
remove_connection_if_exists(connections, poll_token);
|
||||||
|
|
||||||
|
|
@ -203,12 +199,10 @@ pub fn send_out_messages(
|
||||||
|
|
||||||
if let Some(established_ws) = opt_established_ws {
|
if let Some(established_ws) = opt_established_ws {
|
||||||
if established_ws.peer_addr != meta.peer_addr {
|
if established_ws.peer_addr != meta.peer_addr {
|
||||||
eprintln!("socket worker: peer socket addrs didn't match");
|
eprintln!("socket worker error: peer socket addrs didn't match");
|
||||||
|
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
dbg!(out_message.clone());
|
|
||||||
|
|
||||||
use ::tungstenite::Error::Io;
|
use ::tungstenite::Error::Io;
|
||||||
|
|
||||||
|
|
@ -218,7 +212,7 @@ pub fn send_out_messages(
|
||||||
continue;
|
continue;
|
||||||
},
|
},
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
dbg!(err);
|
eprintln!("error writing ws message: {}", err);
|
||||||
|
|
||||||
remove_connection_if_exists(
|
remove_connection_if_exists(
|
||||||
connections,
|
connections,
|
||||||
|
|
|
||||||
|
|
@ -82,12 +82,8 @@ pub fn remove_inactive_connections(
|
||||||
if connection.valid_until.0 < now {
|
if connection.valid_until.0 < now {
|
||||||
connection.close();
|
connection.close();
|
||||||
|
|
||||||
println!("closing connection, it is inactive");
|
|
||||||
|
|
||||||
false
|
false
|
||||||
} else {
|
} else {
|
||||||
println!("keeping connection, it is still active");
|
|
||||||
|
|
||||||
true
|
true
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
|
||||||
|
|
@ -234,8 +234,6 @@ impl InMessage {
|
||||||
|
|
||||||
if let Ok(ActionWrapper { action: Action::Announce, inner }) = res {
|
if let Ok(ActionWrapper { action: Action::Announce, inner }) = res {
|
||||||
return Some(InMessage::AnnounceRequest(inner));
|
return Some(InMessage::AnnounceRequest(inner));
|
||||||
} else {
|
|
||||||
dbg!(res);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
let res: Result<ActionWrapper<ScrapeRequest>, _> = serde_json::from_str(&text);
|
let res: Result<ActionWrapper<ScrapeRequest>, _> = serde_json::from_str(&text);
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue