impl Connection for TcpConnection {
type Error = tokio::io::Error;
fn new(config: ConnectionConfiguration) -> Self {
return Self {
listener: None,
task_aborters: Vec::new(),
config: config,
};
}
async fn bind(&mut self) -> Result<(), Self::Error> {
// Bind a socket and set the field with it:
self.listener = Some(TcpListener::bind((self.config.addr, self.config.port)).await?);
return Ok(());
}
async fn accept(&mut self) -> Result<(Sender<Vec<u8>>, Receiver<Vec<u8>>), Self::Error> {
// Check if we're supposed to be listening:
if let Some(listener) = &self.listener {
// If we are, accept one connection:
let (stream, address) = listener.accept().await?;
eprintln!("New connection from {}.", &address);
// And split the stream into a sender and a receiver:
let (mut rx, mut tx) = stream.into_split();
// Now we need to make two key exchange objects:
let mut i_kex = self.config.kex.generate();
let mut o_kex = self.config.kex.generate();
// We'll send `i_kex`'s public key first, then `o_kex`'s:
tx.write_all(&i_kex.get_local_pubkey()).await?;
tx.write_all(&o_kex.get_local_pubkey()).await?;
// Then, we read out the remote public keys:
let mut i_pubkey_buf: Vec<u8> = vec![0_u8; i_kex.get_public_key_length()];
let mut o_pubkey_buf: Vec<u8> = vec![0_u8; o_kex.get_public_key_length()];
rx.read_exact(&mut o_pubkey_buf).await?; // The other's output should be my input,
rx.read_exact(&mut i_pubkey_buf).await?; // and vis-versa.
// And set the remote public key on the key exchangers:
if let Err(e) = i_kex.set_remote_pubkey(&i_pubkey_buf) {
return Err(Error::other(e));
}
if let Err(e) = o_kex.set_remote_pubkey(&o_pubkey_buf) {
return Err(Error::other(e));
}
// Now we need to actually initiate the key exchange, starting with the client init step:
tx.write_all(&o_kex.client_init().map_err(|e| { Error::other(e.to_string()) })?).await?; // Send a client init.
let mut i_remote_client_init_buf: Vec<u8> = vec![0_u8; i_kex.get_client_init_length()]; // For holding the client's client init.
rx.read_exact(&mut i_remote_client_init_buf).await?; // Receive it.
// Do the server init step:
tx.write_all(&i_kex.server_init(&i_remote_client_init_buf).map_err(|e| { Error::other(e.to_string()) })?).await?; // Send a server init.
let mut o_remote_server_init_buf: Vec<u8> = vec![0_u8; o_kex.get_server_init_length()]; // For holding the client's server init.
rx.read_exact(&mut o_remote_server_init_buf).await?;
// Do the client confirm step:
o_kex.client_confirm(&o_remote_server_init_buf).map_err(|e| { Error::other(e.to_string()) })?; // Done with key exchange!
// Make the keys/crypto thingies:
let (encryptor, decryptor) = self.config.crypto.generate(i_kex, o_kex);
// Make the channels:
let (send_sender, mut send_receiver) = mpsc::channel::<Vec<u8>>(Self::CHANNEL_BUFFER_SIZE);
let (mut recv_sender, recv_receiver) = mpsc::channel::<Vec<u8>>(Self::CHANNEL_BUFFER_SIZE);
// Spawn the tasks, and add their abort handles to the internal vector:
self.task_aborters.push(task::spawn(async move {
// Read byte vectors out of the channel, until the other end is dropped:
while let Some(mut byte_vec) = send_receiver.recv().await {
// Attempt to encrypt the data:
if let Err(e) = encryptor.encrypt(&mut byte_vec, b"") {
// If unsuccessful:
eprintln!("error in sender task on connection to {}: {:?}", &address, e);
break;
} else {
// If successful, try to send it:
let mut message: Vec<u8> = byte_vec.len().to_le_bytes().to_vec(); // First, get the length.
message.append(&mut byte_vec); // Now append the rest of the message.
if let Err(e) = tx.write_all(&byte_vec).await {
// If sending fails:
eprintln!("error in sender task on connection to {}: {}", &address, e);
break;
}
}
}
}).abort_handle()); // Send task.
self.task_aborters.push(task::spawn(async move {}).abort_handle()); // Receive task.
// Return the channels:
return Ok((send_sender, recv_receiver));
} else {
// If this `struct` _shouldn't_ be listening:
return Err(Error::other("this `TcpConnection` should not be listening!"));
}
}
async fn connect(&mut self, addr: Ipv6Addr, port: u16) -> Result<(Sender<Vec<u8>>, Receiver<Vec<u8>>), Self::Error> {
todo!()
}
}
I'm getting this error message from the first call to task::spawn(), and from what I've seen online, it's not that uncommon an error, but every time it pops up on these forums it's quite different, so finding a solution has been difficult for me. Note that I will be cleaning up this messy code soon. Here is the full error message:
error[E0521]: borrowed data escapes outside of method
--> src/qshd/connection/qsh_tcp.rs:105:28
|
50 | async fn accept(&mut self) -> Result<(Sender<Vec<u8>>, Receiver<Vec<u8>>), Self::Error> {
| ---------
| |
| `self` is a reference that is only valid in the method body
| let's call the lifetime of this reference `'1`
...
105 | self.task_aborters.push(task::spawn(async move {
| _____________________________________^
106 | | // Read byte vectors out of the channel, until the other end is dropped:
107 | | while let Some(mut byte_vec) = send_receiver.recv().await {
... |
124 | | }).abort_handle()); // Send task.
| | ^
| | |
| |______________`self` escapes the method body here
| argument requires that `'1` must outlive `'static`