diff options
| -rw-r--r-- | client-native-lib/src/instance.rs (renamed from client-native-lib/src/state.rs) | 8 | ||||
| -rw-r--r-- | client-native-lib/src/lib.rs | 8 | ||||
| -rw-r--r-- | client-native-lib/src/peer.rs | 48 | ||||
| -rw-r--r-- | client-native-rift/src/main.rs | 17 |
4 files changed, 55 insertions, 26 deletions
diff --git a/client-native-lib/src/state.rs b/client-native-lib/src/instance.rs index d129e32..b3688c3 100644 --- a/client-native-lib/src/state.rs +++ b/client-native-lib/src/instance.rs @@ -17,7 +17,7 @@ use std::{collections::HashMap, sync::Arc}; use tokio::sync::RwLock; use webrtc::api::API; -pub struct State { +pub struct Instance { pub event_handler: Box<dyn EventHandler>, pub conn: SignalingConnection, pub config: Config, @@ -27,7 +27,7 @@ pub struct State { my_id: RwLock<Option<usize>>, pub peers: RwLock<HashMap<usize, Arc<Peer>>>, } -impl State { +impl Instance { pub async fn new(config: Config, event_handler: Box<dyn EventHandler>) -> Arc<Self> { let conn = signaling::SignalingConnection::new(&config.signaling_uri, &config.secret).await; let key = crypto::Key::derive(&config.secret); @@ -51,8 +51,8 @@ impl State { pub async fn receive_loop(self: Arc<Self>) { while let Some(packet) = self.conn.recv.write().await.next().await { debug!("{packet:?}"); - let state = self.clone(); - state.on_message(packet).await + let inst = self.clone(); + inst.on_message(packet).await } } diff --git a/client-native-lib/src/lib.rs b/client-native-lib/src/lib.rs index bb88b9f..27c0595 100644 --- a/client-native-lib/src/lib.rs +++ b/client-native-lib/src/lib.rs @@ -12,7 +12,7 @@ use std::{pin::Pin, sync::Arc}; use futures_util::Future; use peer::Peer; use protocol::ProvideInfo; -use state::State; +use instance::Instance; use tokio::sync::RwLock; use webrtc::{ api::{ @@ -25,7 +25,7 @@ pub mod crypto; pub mod peer; pub mod protocol; pub mod signaling; -pub mod state; +pub mod instance; pub use webrtc; @@ -54,9 +54,9 @@ pub trait LocalResource: Send + Sync + 'static { pub trait EventHandler: Send + Sync + 'static { fn remote_resource_added( &self, - peer: &Peer, + peer: Arc<Peer>, info: ProvideInfo, ) -> Pin<Box<dyn Future<Output = ()>>>; - fn remote_resource_removed(&self, peer: &Peer, id: String) + fn remote_resource_removed(&self, peer: Arc<Peer>, id: String) -> Pin<Box<dyn Future<Output = ()>>>; } diff --git a/client-native-lib/src/peer.rs b/client-native-lib/src/peer.rs index 5e672e6..b6ecf55 100644 --- a/client-native-lib/src/peer.rs +++ b/client-native-lib/src/peer.rs @@ -4,13 +4,14 @@ Copyright (C) 2022 metamuffin <metamuffin@disroot.org> */ use crate::{ + instance::Instance, protocol::{self, ProvideInfo, RelayMessage, Sdp}, - state::State, }; use log::info; use std::{collections::HashMap, sync::Arc}; use tokio::sync::RwLock; use webrtc::{ + data::data_channel::DataChannel, ice_transport::{ ice_candidate::{RTCIceCandidate, RTCIceCandidateInit}, ice_server::RTCIceServer, @@ -19,17 +20,34 @@ use webrtc::{ configuration::RTCConfiguration, peer_connection_state::RTCPeerConnectionState, sdp::session_description::RTCSessionDescription, RTCPeerConnection, }, + track::track_remote::TrackRemote, }; pub struct Peer { - pub state: Arc<State>, + pub inst: Arc<Instance>, pub peer_connection: RTCPeerConnection, pub resources_provided: RwLock<HashMap<String, ProvideInfo>>, pub id: usize, } +// pub struct RemoteResource { +// info: ProvideInfo, +// state: RemoteResourceInner, +// } +// // (Box<dyn FnOnce(Arc<TransportChannel>) -> Pin<Box<dyn Future<Output = ()>>>>) +// pub enum RemoteResourceInner { +// Disconnected, +// AwaitConnect, +// Connected(Arc<TransportChannel>), +// AwaitDisconnect, +// } +// pub enum TransportChannel { +// Track(TrackRemote), +// DataChannel(DataChannel), +// } + impl Peer { - pub async fn create(state: Arc<State>, id: usize) -> Arc<Self> { + pub async fn create(inst: Arc<Instance>, id: usize) -> Arc<Self> { info!("({id}) peer joined"); let config = RTCConfiguration { ice_servers: vec![RTCIceServer { @@ -39,10 +57,10 @@ impl Peer { ..Default::default() }; - let peer_connection = state.api.new_peer_connection(config).await.unwrap(); + let peer_connection = inst.api.new_peer_connection(config).await.unwrap(); let peer = Arc::new(Self { resources_provided: Default::default(), - state: state.clone(), + inst: inst.clone(), peer_connection, id, }); @@ -80,6 +98,7 @@ impl Peer { { peer.peer_connection .on_data_channel(box move |dc| { + info!("got a data channel"); Box::pin(async move { dc.on_message(box move |message| { Box::pin(async move { println!("{:?}", message.data) }) @@ -92,11 +111,18 @@ impl Peer { peer } + pub async fn request_resource(&self, id: String) { + self.send_relay(RelayMessage::Request { id }).await; + } + pub async fn request_stop_resource(&self, id: String) { + self.send_relay(RelayMessage::RequestStop { id }).await; + } + pub async fn send_relay(&self, inner: RelayMessage) { - self.state.send_relay(self.id, inner).await + self.inst.send_relay(self.id, inner).await } - pub async fn on_relay(&self, p: RelayMessage) { + pub async fn on_relay(self: Arc<Self>, p: RelayMessage) { match p { RelayMessage::Offer(o) => self.on_offer(o).await, RelayMessage::Answer(a) => self.on_answer(a).await, @@ -110,17 +136,17 @@ impl Peer { .write() .await .insert(info.id.clone(), info.clone()); - self.state + self.inst .event_handler - .remote_resource_added(&self, info) + .remote_resource_added(self.clone(), info) .await; } RelayMessage::ProvideStop { id } => { info!("remote resource removed: ({:?}) ", id); self.resources_provided.write().await.remove(&id); - self.state + self.inst .event_handler - .remote_resource_removed(&self, id) + .remote_resource_removed(self.clone(), id) .await; } _ => (), diff --git a/client-native-rift/src/main.rs b/client-native-rift/src/main.rs index 9d97b8e..613a4e6 100644 --- a/client-native-rift/src/main.rs +++ b/client-native-rift/src/main.rs @@ -8,7 +8,7 @@ use bytes::Bytes; use clap::{Parser, Subcommand}; use client_native_lib::{ - peer::Peer, state::State, webrtc::data_channel::RTCDataChannel, Config, EventHandler, + instance::Instance, peer::Peer, webrtc::data_channel::RTCDataChannel, Config, EventHandler, }; use log::{error, info}; use std::{future::Future, pin::Pin, sync::Arc}; @@ -48,7 +48,7 @@ pub struct Args { async fn run() { let args = Args::parse(); - let state = State::new( + let inst = Instance::new( Config { secret: args.secret.clone(), signaling_uri: args.signaling_uri.clone(), @@ -58,7 +58,7 @@ async fn run() { ) .await; - state.receive_loop().await; + inst.receive_loop().await; tokio::signal::ctrl_c().await.unwrap(); error!("interrupt received, exiting"); @@ -69,18 +69,21 @@ struct Handler {} impl EventHandler for Handler { fn remote_resource_added( &self, - peer: &Peer, + peer: Arc<Peer>, info: client_native_lib::protocol::ProvideInfo, ) -> Pin<Box<dyn Future<Output = ()>>> { - todo!() + let id = info.id.clone(); + Box::pin(async move { + peer.request_resource(id).await; + }) } fn remote_resource_removed( &self, - peer: &Peer, + peer: Arc<Peer>, id: String, ) -> Pin<Box<dyn Future<Output = ()>>> { - todo!() + Box::pin(async {}) } } |