aboutsummaryrefslogtreecommitdiff
path: root/src/conn.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/conn.rs')
-rw-r--r--src/conn.rs34
1 files changed, 13 insertions, 21 deletions
diff --git a/src/conn.rs b/src/conn.rs
index 474032d..73bc6a1 100644
--- a/src/conn.rs
+++ b/src/conn.rs
@@ -6,15 +6,16 @@ use std::sync::{Arc, Mutex};
use bytes::Bytes;
use log::{debug, error, trace};
-use tokio::io::split;
use tokio::net::TcpStream;
use tokio::sync::{mpsc, oneshot, watch};
+use tokio_util::compat::*;
+
+use futures::io::AsyncReadExt;
use async_trait::async_trait;
use kuska_handshake::async_std::{
- handshake_client, handshake_server, BoxStream, TokioCompatExt, TokioCompatExtRead,
- TokioCompatExtWrite,
+ handshake_client, handshake_server, BoxStream
};
use crate::error::*;
@@ -35,9 +36,11 @@ pub(crate) struct ServerConn {
impl ServerConn {
pub(crate) async fn run(netapp: Arc<NetApp>, socket: TcpStream) -> Result<(), Error> {
- let mut asyncstd_socket = TokioCompatExt::wrap(socket);
+ let remote_addr = socket.peer_addr()?;
+ let mut socket = socket.compat();
+
let handshake = handshake_server(
- &mut asyncstd_socket,
+ &mut socket,
netapp.netid.clone(),
netapp.id,
netapp.privkey.clone(),
@@ -45,19 +48,13 @@ impl ServerConn {
.await?;
let peer_id = handshake.peer_pk;
- let tokio_socket = asyncstd_socket.into_inner();
- let remote_addr = tokio_socket.peer_addr()?;
-
debug!(
"Handshake complete (server) with {}@{}",
hex::encode(&peer_id),
remote_addr
);
- let (read, write) = split(tokio_socket);
-
- let read = TokioCompatExtRead::wrap(read);
- let write = TokioCompatExtWrite::wrap(write);
+ let (read, write) = socket.split();
let (read, write) =
BoxStream::from_handshake(read, write, handshake, 0x8000).split_read_write();
@@ -148,10 +145,11 @@ impl ClientConn {
socket: TcpStream,
peer_id: NodeID,
) -> Result<(), Error> {
- let mut asyncstd_socket = TokioCompatExt::wrap(socket);
+ let remote_addr = socket.peer_addr()?;
+ let mut socket = socket.compat();
let handshake = handshake_client(
- &mut asyncstd_socket,
+ &mut socket,
netapp.netid.clone(),
netapp.id,
netapp.privkey.clone(),
@@ -159,19 +157,13 @@ impl ClientConn {
)
.await?;
- let tokio_socket = asyncstd_socket.into_inner();
- let remote_addr = tokio_socket.peer_addr()?;
-
debug!(
"Handshake complete (client) with {}@{}",
hex::encode(&peer_id),
remote_addr
);
- let (read, write) = split(tokio_socket);
-
- let read = TokioCompatExtRead::wrap(read);
- let write = TokioCompatExtWrite::wrap(write);
+ let (read, write) = socket.split();
let (read, write) =
BoxStream::from_handshake(read, write, handshake, 0x8000).split_read_write();