diff options
Diffstat (limited to 'client-native-lib/src')
| -rw-r--r-- | client-native-lib/src/lib.rs | 27 | ||||
| -rw-r--r-- | client-native-lib/src/peer.rs | 69 | ||||
| -rw-r--r-- | client-native-lib/src/state.rs | 36 |
3 files changed, 65 insertions, 67 deletions
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<State> { +pub async fn connect<I, P>(config: Config, sup: Arc<I>) -> Arc<State<P, I>> +where + I: PeerInit<P> + 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<State> { .with_interceptor_registry(registry) .build(); + let (relay_tx, mut relay_rx) = unbounded_channel(); let state = Arc::new(State { peers: Default::default(), key, @@ -45,12 +52,24 @@ pub async fn connect(config: Config) -> Arc<State> { 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 { + debug!("receiving packets now"); while let Some(packet) = recv.recv().await { debug!("{packet:?}"); let state = state.clone(); 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<State>, // 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<State>, id: usize) -> Arc<Self> { + pub async fn create<P: HasPeer, I: PeerInit<P>>( + state: Arc<State<P, I>>, + signal: UnboundedSender<(usize, RelayMessage)>, + id: usize, + ) -> Arc<Self> { 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<Peer>; +} +pub trait PeerInit<P> { + fn add_peer( + &self, + p: Arc<Peer>, + ) -> Pin<Box<dyn Future<Output = Arc<P>> + 'static + Send + Sync>>; +} + +pub struct State<P: HasPeer, I: PeerInit<P>> { + pub sup: Arc<I>, pub config: Config, pub api: API, pub key: Key, pub my_id: RwLock<Option<usize>>, pub sender: UnboundedSender<ServerboundPacket>, - pub peers: RwLock<HashMap<usize, Arc<Peer>>>, + pub peers: RwLock<HashMap<usize, Arc<P>>>, + pub relay_tx: UnboundedSender<(usize, RelayMessage)>, } -impl State { +impl<P: HasPeer, I: PeerInit<P>> State<P, I> { 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") } |