eisterman/bonknet
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Implement opening of the DataStream. Just the broker copy task/manager is missing

Browse Source
This commit is contained in:
Federico Pasqua
2024-02-21 16:40:49 +01:00
parent 69a37ae89a
commit 83c7a95414
9 changed files with 328 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bonknet_broker/src/dataconnmanager.rs Normal file
View File

@@ -0,0 +1,2 @@
use actix::prelude::*;

22
bonknet_broker/src/main.rs
View File

@@ -1,14 +1,13 @@
mod servercertdb;
mod pendingdataconndb;
mod servermanager;
mod dataconnmanager;
use servercertdb::*;
use pendingdataconndb::*;
use servermanager::*;
use actix::prelude::*;
use std::collections::HashMap;
use std::sync::{Arc};
use std::time::{Instant, Duration};
use std::sync::Arc;
use libbonknet::*;
use rustls::{RootCertStore, ServerConfig};
use rustls::server::WebPkiClientVerifier;
@@ -17,16 +16,10 @@ use actix_server::Server;
use actix_rt::net::TcpStream;
use actix_service::{ServiceFactoryExt as _};
use futures::{StreamExt, SinkExt};
use rand::random;
use tokio_util::codec::{Framed, FramedRead, FramedWrite, LengthDelimitedCodec};
use tracing::{info, error, warn};
use rcgen::{Certificate, CertificateParams, DnType, KeyPair};
use thiserror::Error;
use tokio::io::{ReadHalf, WriteHalf};
use tokio_util::bytes::{Bytes, BytesMut};
use tokio::io::Error;
use tokio::sync::{oneshot, Mutex};
use uuid::Uuid;
type TransportStream = Framed<TlsStream<TcpStream>, LengthDelimitedCodec>;
type TransportStreamTx = FramedWrite<WriteHalf<TlsStream<TcpStream>>, LengthDelimitedCodec>;
@@ -160,7 +153,8 @@ async fn main() {
}
OpenDataStream(conn_id) => {
info!("OpenDataStream with {:?}", conn_id);
// TODO: OpenDataStream
let msg = RegisterStream::server(conn_id, transport);
pdcm_addr.send(msg).await.unwrap().unwrap();
}
}
}
@@ -180,7 +174,7 @@ async fn main() {
info!("Client connection");
let codec = LengthDelimitedCodec::new();
let transport = Framed::new(stream, codec);
client_handler(transport, sm_addr).await;
client_handler(transport, sm_addr, pdcm_addr).await;
} else {
error!("Unknown Root Certificate");
}
@@ -312,7 +306,7 @@ async fn guestserver_handler(mut transport: TransportStream, server_db_addr: Add
}
}
async fn client_handler(mut transport: TransportStream, sm_addr: Addr<ServerManager>) {
async fn client_handler(mut transport: TransportStream, sm_addr: Addr<ServerManager>, pdcm_addr: Addr<PendingDataConnManager>) {
loop {
match transport.next().await {
None => {
@@ -347,7 +341,9 @@ async fn client_handler(mut transport: TransportStream, sm_addr: Addr<ServerMana
}
FromClientCommand::UpgradeToDataStream(conn_id) => {
info!("Upgrade to DataStream with conn_id {:?}", conn_id);
// TODO: Upgrade to DataStream
let msg = RegisterStream::client(conn_id, transport);
pdcm_addr.send(msg).await.unwrap().unwrap();
break;
}
}
}

135
bonknet_broker/src/pendingdataconndb.rs
View File

@@ -1,11 +1,12 @@
use actix::prelude::*;
use actix_tls::accept::rustls_0_22::TlsStream;
use futures::SinkExt;
use thiserror::Error;
use tokio::io::{ReadHalf, WriteHalf};
use tokio::net::TcpStream;
use tokio_util::codec::{Framed, FramedRead, FramedWrite, LengthDelimitedCodec};
use tokio_util::codec::{Framed, LengthDelimitedCodec};
use tracing::{error, info};
use uuid::Uuid;
use libbonknet::*;
type TransportStream = Framed<TlsStream<TcpStream>, LengthDelimitedCodec>;
@@ -73,8 +74,8 @@ struct Record {
impl Record {
fn new(server_conn_id: Uuid, client_conn_id: Uuid) -> Self {
let server = SideRecord { conn_id: client_conn_id, transport: None };
let client = SideRecord { conn_id: server_conn_id, transport: None };
let server = SideRecord { conn_id: server_conn_id, transport: None };
let client = SideRecord { conn_id: client_conn_id, transport: None };
Record { server, client }
}
}
@@ -88,6 +89,21 @@ impl PendingDataConnManager {
pub fn new() -> Self {
PendingDataConnManager { db: Vec::new() }
}
fn retrieve_siderecord(&mut self, kind: &RegisterKind, conn_id: &Uuid) -> Option<&mut SideRecord> {
use RegisterKind::*;
let record = match match kind {
Server => self.db.iter_mut().find(|x| x.server.conn_id == *conn_id),
Client => self.db.iter_mut().find(|x| x.client.conn_id == *conn_id),
} {
None => return None,
Some(item) => item,
};
Some(match kind {
Server => &mut record.server,
Client => &mut record.client,
})
}
}
impl Actor for PendingDataConnManager {
@@ -113,37 +129,104 @@ impl Handler<NewPendingConn> for PendingDataConnManager {
}
}
impl Handler<RegisterStream> for PendingDataConnManager {
/*
Esegui tutti i test normali in Sync.
Quando devi inviare il Accepted, notificati la cosa come AcceptStream con lo stream in suo possesso
ma stavolta con un ResponseActFuture.
Se c'e un fallimento, sposta il transport in un ctx::spawn che invii un FAILED.
Se tutto OK, checka DI NUOVO tutto, e se i check sono positivi, registra lo stream nell'Actor.
Per gestire lo Spawn di una connection, l'unica risposta e' gestire lo spawn connection come un
Message Handler a sua volta. Quando uno stream completa l'invio del suo Accepted, esso appare nel Record.
Quando il secondo stream arriva e completa il suo accepted, anch'esso viene registrato nel Record, quindi
siamo nella condizione di spawn, perche ci sono entrambi i transport nel Record.
Quindi se alla fine di un check and register ci sono entrambi gli stream, spostali entrambi fuori, droppa
il record e invia i due transport ad un terzo actor, su un altro Arbiter, che esegua il tokio::io::copy e
gestisca le connessioni aperte.
*/
#[derive(Message)]
#[rtype(result = "Result<(),PendingDataConnError>")]
struct TryStartDataStream {
kind: RegisterKind,
conn_id: Uuid,
}
impl Handler<TryStartDataStream> for PendingDataConnManager {
type Result = Result<(), PendingDataConnError>;
fn handle(&mut self, msg: RegisterStream, _ctx: &mut Self::Context) -> Self::Result {
fn handle(&mut self, msg: TryStartDataStream, _ctx: &mut Self::Context) -> Self::Result {
use RegisterKind::*;
let conn_id = msg.conn_id;
let record = match match msg.kind {
Server => self.db.iter_mut().find(|x| x.server.conn_id == conn_id),
Client => self.db.iter_mut().find(|x| x.client.conn_id == conn_id),
} {
None => {
error!("Found no connection with {:?} conn_id {:?}", msg.kind, conn_id);
return Err(PendingDataConnError::GenericFailure);
},
Some(item) => item,
let idx = match msg.kind {
Server => self.db.iter().enumerate().find(|(i, x)| x.server.conn_id == msg.conn_id),
Client => self.db.iter().enumerate().find(|(i, x)| x.client.conn_id == msg.conn_id),
};
let side_record = match msg.kind {
Server => &mut record.server,
Client => &mut record.client,
if let Some((_idx, record)) = idx {
if record.client.transport.is_some() && record.server.transport.is_some() {
// TODO: Launch the "thing" that will manage the data mirroring
info!("LAUNCHING DATA MIRRORING");
// TODO: we can drop record and use idx to remove the record itself from the vector
// and then send it to another manager for the spawn of the real connection
}
Ok(())
} else {
Err(PendingDataConnError::GenericFailure)
}
}
}
impl Handler<RegisterStream> for PendingDataConnManager {
type Result = ResponseActFuture<Self, Result<(), PendingDataConnError>>;
fn handle(&mut self, msg: RegisterStream, _ctx: &mut Self::Context) -> Self::Result {
let side_record = match self.retrieve_siderecord(&msg.kind, &msg.conn_id) {
None => {
error!("Found no connection with {:?} conn_id {:?}", msg.kind, msg.conn_id);
return Box::pin(fut::err(PendingDataConnError::GenericFailure));
}
Some(item) => item,
};
if side_record.transport.is_some() {
// TODO: It can be good to check if the connection is still open, if not, drop and use the new one.
error!("Connection already with a socket!");
return Err(PendingDataConnError::GenericFailure);
Box::pin(fut::err(PendingDataConnError::GenericFailure))
} else {
side_record.transport = Some(msg.transport);
// This Fut will send the Accepted and only then, register the transport stream
// in the Manager. If during the registration there are all the transport in places,
// you can start the datapiping
Box::pin(async move {
let mut transport = msg.transport;
let reply = ToPeerDataStream::OkDataStreamRequestAccepted;
let res = transport.send(rmp_serde::to_vec(&reply).unwrap().into()).await;
(transport, res)
}.into_actor(self).map(move |(transport, res), a, c| {
match res {
Ok(_) => {
// TODO: to not do the check twice I can put a "lock variable" inside the record
// to prevent the put in Accept of another stream while we are waiting to send the
// accept message
let side_record = match a.retrieve_siderecord(&msg.kind, &msg.conn_id) {
None => {
error!("Found no connection with {:?} conn_id {:?}", msg.kind, msg.conn_id);
return Err(PendingDataConnError::GenericFailure);
}
Some(item) => item,
};
if side_record.transport.is_some() {
// TODO: It can be good to check if the connection is still open, if not, drop and use the new one.
error!("Connection already with a socket!");
return Err(PendingDataConnError::GenericFailure);
}
side_record.transport = Some(transport);
c.notify(TryStartDataStream { kind: msg.kind, conn_id: msg.conn_id });
Ok(())
}
Err(e) => {
error!("Error during OkDataStreamRequestAccepted sending: {:?}", e);
Err(PendingDataConnError::GenericFailure)
}
}
}))
}
if record.client.transport.is_some() && record.server.transport.is_some() {
// TODO: Launch the "thing" that will manage the data mirroring
info!("LAUNCHING DATA MIRRORING");
}
Ok(())
}
}

3
bonknet_broker/src/servermanager.rs
View File

@@ -3,10 +3,9 @@ use std::io::Error;
use std::sync::{Arc};
use std::time::{Duration, Instant};
use actix::prelude::*;
use actix_server::Server;
use rand::random;
use thiserror::Error;
use futures::{StreamExt, SinkExt};
use futures::{SinkExt};
use tokio::sync::{Mutex, oneshot};
use tokio_util::bytes::{Bytes, BytesMut};
use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec};