From 7c68bdb983c2f52fb09e1a5418e5bc44d3b44b30 Mon Sep 17 00:00:00 2001 From: metamuffin Date: Wed, 14 Sep 2022 23:03:52 +0200 Subject: rift works ™™ MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- client-native-lib/src/lib.rs | 27 ++++++++++++++--- client-native-lib/src/peer.rs | 69 ++++++++++-------------------------------- client-native-lib/src/state.rs | 36 ++++++++++++++++------ 3 files changed, 65 insertions(+), 67 deletions(-) (limited to 'client-native-lib/src') diff --git a/client-native-lib/src/lib.rs b/client-native-lib/src/lib.rs index 118d962..68dcecb 100644 --- a/client-native-lib/src/lib.rs +++ b/client-native-lib/src/lib.rs @@ -3,9 +3,9 @@ use log::debug; use signaling::signaling_connect; -use state::State; -use std::sync::Arc; -use tokio::sync::RwLock; +use state::{HasPeer, PeerInit, State}; +use std::{marker::Sync, sync::Arc}; +use tokio::sync::{mpsc::unbounded_channel, RwLock}; use webrtc::{ api::{ interceptor_registry::register_default_interceptors, media_engine::MediaEngine, APIBuilder, @@ -19,12 +19,18 @@ pub mod protocol; pub mod signaling; pub mod state; +pub use webrtc; + pub struct Config { pub signaling_host: String, pub secret: String, } -pub async fn connect(config: Config) -> Arc { +pub async fn connect(config: Config, sup: Arc) -> Arc> +where + I: PeerInit

+ Sync + std::marker::Send + 'static, + P: HasPeer + Sync + std::marker::Send + 'static, +{ let (sender, mut recv) = signaling_connect(&config.signaling_host, &config.secret).await; let key = crypto::Key::derive(&config.secret); @@ -38,6 +44,7 @@ pub async fn connect(config: Config) -> Arc { .with_interceptor_registry(registry) .build(); + let (relay_tx, mut relay_rx) = unbounded_channel(); let state = Arc::new(State { peers: Default::default(), key, @@ -45,8 +52,20 @@ pub async fn connect(config: Config) -> Arc { my_id: RwLock::new(None), sender, config, + relay_tx, + sup, }); + { + let state = state.clone(); + tokio::spawn(async move { + debug!("receiving packets now"); + while let Some((r, p)) = relay_rx.recv().await { + let state = state.clone(); + state.send_relay(r, p).await + } + }); + } { let state = state.clone(); tokio::spawn(async move { diff --git a/client-native-lib/src/peer.rs b/client-native-lib/src/peer.rs index 64eb641..280cc06 100644 --- a/client-native-lib/src/peer.rs +++ b/client-native-lib/src/peer.rs @@ -1,7 +1,11 @@ +use crate::{ + protocol::{self, RTCSessionDescriptionInit, RelayMessage}, + state::{HasPeer, PeerInit, State}, +}; use log::info; use std::sync::Arc; +use tokio::sync::mpsc::UnboundedSender; use webrtc::{ - data_channel::data_channel_message::DataChannelMessage, ice_transport::{ice_candidate::RTCIceCandidate, ice_server::RTCIceServer}, peer_connection::{ configuration::RTCConfiguration, peer_connection_state::RTCPeerConnectionState, @@ -9,19 +13,18 @@ use webrtc::{ }, }; -use crate::{ - protocol::{self, RTCSessionDescriptionInit, RelayMessage}, - state::State, -}; - pub struct Peer { - state: Arc, // maybe use Weak later - peer_connection: RTCPeerConnection, - id: usize, + pub signal: UnboundedSender<(usize, RelayMessage)>, + pub peer_connection: RTCPeerConnection, + pub id: usize, } impl Peer { - pub async fn create(state: Arc, id: usize) -> Arc { + pub async fn create>( + state: Arc>, + signal: UnboundedSender<(usize, RelayMessage)>, + id: usize, + ) -> Arc { info!("({id}) peer joined"); let config = RTCConfiguration { ice_servers: vec![RTCIceServer { @@ -32,15 +35,14 @@ impl Peer { }; let peer_connection = state.api.new_peer_connection(config).await.unwrap(); - let peer = Arc::new(Self { + signal, peer_connection, id, - state: state.clone(), }); peer.peer_connection .on_peer_connection_state_change(Box::new(move |s: RTCPeerConnectionState| { - println!("conn state: {s}"); + info!("connection state changed: {s}"); Box::pin(async {}) })) .await; @@ -81,50 +83,11 @@ impl Peer { }) .await; } - - // if let Action::Send { .. } = &peer.state.args.action { - // peer.start_transfer().await - // } - peer } pub async fn send_relay(&self, inner: RelayMessage) { - self.state.send_relay(self.id, inner).await - } - - pub async fn start_transfer(&self) { - info!("starting data channel"); - let data_channel = self - .peer_connection - .create_data_channel("file-transfer", None) - .await - .unwrap(); - - data_channel - .on_message(Box::new(move |msg: DataChannelMessage| { - let msg_str = String::from_utf8(msg.data.to_vec()).unwrap(); - println!("message! '{}'", msg_str); - Box::pin(async {}) - })) - .await; - - { - let dc2 = data_channel.clone(); - data_channel - .on_open(box move || { - let data_channel = dc2.clone(); - Box::pin(async move { - loop { - data_channel - .send(&bytes::Bytes::from_static(b"test\n")) - .await - .unwrap(); - } - }) - }) - .await; - } + self.signal.send((self.id, inner)).unwrap() } pub async fn on_relay(&self, p: RelayMessage) { diff --git a/client-native-lib/src/state.rs b/client-native-lib/src/state.rs index 57c4b29..c5e9365 100644 --- a/client-native-lib/src/state.rs +++ b/client-native-lib/src/state.rs @@ -1,5 +1,6 @@ -use std::{collections::HashMap, sync::Arc}; +use std::{collections::HashMap, pin::Pin, sync::Arc}; +use futures_util::Future; use log::warn; use tokio::sync::{mpsc::UnboundedSender, RwLock}; use webrtc::api::API; @@ -7,18 +8,31 @@ use webrtc::api::API; use crate::{ crypto::Key, peer::Peer, - protocol::{self, ClientboundPacket, RelayMessage, RelayMessageWrapper, ServerboundPacket}, Config, + protocol::{self, ClientboundPacket, RelayMessage, RelayMessageWrapper, ServerboundPacket}, + Config, }; -pub struct State { +pub trait HasPeer { + fn peer(&self) -> &Arc; +} +pub trait PeerInit

{ + fn add_peer( + &self, + p: Arc, + ) -> Pin> + 'static + Send + Sync>>; +} + +pub struct State> { + pub sup: Arc, pub config: Config, pub api: API, pub key: Key, pub my_id: RwLock>, pub sender: UnboundedSender, - pub peers: RwLock>>, + pub peers: RwLock>>, + pub relay_tx: UnboundedSender<(usize, RelayMessage)>, } -impl State { +impl> State { pub async fn my_id(&self) -> usize { self.my_id.read().await.expect("not initialized yet") } @@ -35,10 +49,12 @@ impl State { if id == self.my_id().await { // we joined - YAY! } else { - self.peers - .write() - .await - .insert(id, Peer::create(self.clone(), id).await); + self.peers.write().await.insert( + id, + self.sup + .add_peer(Peer::create(self.clone(), self.relay_tx.clone(), id).await) + .await, + ); } } protocol::ClientboundPacket::ClientLeave { id: _ } => {} @@ -52,7 +68,7 @@ impl State { pub async fn on_relay(&self, sender: usize, p: RelayMessage) { if let Some(peer) = self.peers.read().await.get(&sender).cloned() { - peer.on_relay(p).await + peer.peer().on_relay(p).await } else { warn!("got a packet from a non-existent peer") } -- cgit v1.2.3-70-g09d2