diff --git a/Cargo.lock b/Cargo.lock
index fc90dbea5..11bd95c70 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -795,6 +795,7 @@ dependencies = [
"slog-scope 4.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
"slog-stdlog 3.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
"smart-default 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio 0.1.14 (registry+https://github.com/rust-lang/crates.io-index)",
"toml 0.4.10 (registry+https://github.com/rust-lang/crates.io-index)",
]
diff --git a/Cargo.toml b/Cargo.toml
index d18c3581c..75f7af630 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -44,3 +44,4 @@ toml = "0.4"
[dev-dependencies]
serial_test = "0.2"
serial_test_derive = "0.2"
+tokio = "0.1"
diff --git a/README.md b/README.md
index 311fc6997..b9f8434f3 100644
--- a/README.md
+++ b/README.md
@@ -6,3 +6,5 @@ Medea
Medea media server
__DEVELOPMENT IN PROGRESS__
+
+{"MakeSdpOffer":{"peer":0,"sdp_offer":"caller_offer"}}
diff --git a/signaling_test.html b/signaling_test.html
new file mode 100644
index 000000000..0fb718e37
--- /dev/null
+++ b/signaling_test.html
@@ -0,0 +1,161 @@
+
+
+
+ Chat
+
+
+
+
+
+
+
+
diff --git a/src/api/client/mod.rs b/src/api/client/mod.rs
index 0f3537eab..81bf83f8e 100644
--- a/src/api/client/mod.rs
+++ b/src/api/client/mod.rs
@@ -1,7 +1,6 @@
//! Implementation of Client API.
-pub mod room;
-pub mod server;
-pub mod session;
+mod session;
-pub use self::{room::*, server::*, session::*};
+pub mod rpc_connection;
+pub mod server;
diff --git a/src/api/client/room.rs b/src/api/client/room.rs
deleted file mode 100644
index 0023db16d..000000000
--- a/src/api/client/room.rs
+++ /dev/null
@@ -1,180 +0,0 @@
-//! Room definitions and implementations.
-
-use std::{
- fmt,
- sync::{Arc, Mutex},
-};
-
-use actix::{
- fut::wrap_future, Actor, ActorFuture, Addr, Context, Handler, Message,
-};
-use futures::{
- future::{self, Either},
- Future,
-};
-use hashbrown::HashMap;
-
-use crate::{
- api::control::{Id as MemberId, Member},
- log::prelude::*,
-};
-
-/// ID of [`Room`].
-pub type Id = u64;
-
-/// Media server room with its [`Member`]s.
-#[derive(Debug)]
-pub struct Room {
- /// ID of this [`Room`].
- pub id: Id,
- /// [`Member`]s which currently are present in this [`Room`].
- pub members: HashMap,
- /// Established [`WsSession`]s of [`Member`]s in this [`Room`].
- pub connections: HashMap>,
- /* TODO: Replace Box> with enum,
- * as the set of all possible RpcConnection types is not closed. */
-}
-
-/// [`Actor`] implementation that provides an ergonomic way
-/// to interact with [`Room`].
-impl Actor for Room {
- type Context = Context;
-}
-
-/// Established RPC connection with some remote [`Member`].
-pub trait RpcConnection: fmt::Debug + Send {
- /// Closes [`RpcConnection`].
- /// No [`RpcConnectionClosed`] signals should be emitted.
- fn close(&mut self) -> Box>;
-}
-
-/// Signal for authorizing new [`RpcConnection`] before establishing.
-#[derive(Debug, Message)]
-#[rtype(result = "Result<(), RpcConnectionAuthorizationError>")]
-pub struct AuthorizeRpcConnection {
- /// ID of [`Member`] to authorize [`RpcConnection`] for.
- pub member_id: MemberId,
- /// Credentials to authorize [`RpcConnection`] with.
- pub credentials: String, // TODO: &str when futures will allow references
-}
-
-/// Error of authorization [`RpcConnection`] in [`Room`].
-#[derive(Debug)]
-pub enum RpcConnectionAuthorizationError {
- /// Authorizing [`Member`] does not exists in the [`Room`].
- MemberNotExists,
- /// Provided credentials are invalid.
- InvalidCredentials,
-}
-
-impl Handler for Room {
- type Result = Result<(), RpcConnectionAuthorizationError>;
-
- /// Responses with `Ok` if `RpcConnection` is authorized, otherwise `Err`s.
- fn handle(
- &mut self,
- msg: AuthorizeRpcConnection,
- _ctx: &mut Self::Context,
- ) -> Self::Result {
- use RpcConnectionAuthorizationError::*;
- if let Some(ref member) = self.members.get(&msg.member_id) {
- if member.credentials.eq(&msg.credentials) {
- return Ok(());
- }
- return Err(InvalidCredentials);
- }
- Err(MemberNotExists)
- }
-}
-
-/// Signal of new [`RpcConnection`] being established with specified [`Member`].
-#[derive(Debug, Message)]
-#[rtype(result = "Result<(), ()>")]
-pub struct RpcConnectionEstablished {
- /// ID of [`Member`] that establishes [`RpcConnection`].
- pub member_id: MemberId,
- /// Established [`RpcConnection`].
- pub connection: Box,
-}
-
-/// Ergonomic type alias for using [`ActorFuture`] for [`Room`].
-type ActFuture = Box>;
-
-impl Handler for Room {
- type Result = ActFuture<(), ()>;
-
- /// Stores provided [`RpcConnection`] for given [`Member`] in the [`Room`].
- ///
- /// If [`Member`] already has any other [`RpcConnection`],
- /// then it will be closed.
- fn handle(
- &mut self,
- msg: RpcConnectionEstablished,
- _: &mut Self::Context,
- ) -> Self::Result {
- info!("RpcConnectionEstablished for member {}", msg.member_id);
-
- let mut fut = Either::A(future::ok(()));
-
- if let Some(mut old_conn) = self.connections.remove(&msg.member_id) {
- debug!("Closing old RpcConnection for member {}", msg.member_id);
- fut = Either::B(old_conn.close());
- }
-
- self.connections.insert(msg.member_id, msg.connection);
-
- Box::new(wrap_future(fut))
- }
-}
-
-/// Signal of existing [`RpcConnection`] of specified [`Member`] being closed.
-#[derive(Debug, Message)]
-pub struct RpcConnectionClosed {
- /// ID of [`Member`] which [`RpcConnection`] is closed.
- pub member_id: MemberId,
- /// Reason of why [`RpcConnection`] is closed.
- pub reason: RpcConnectionClosedReason,
-}
-
-/// Reasons of why [`RpcConnection`] may be closed.
-#[derive(Debug)]
-pub enum RpcConnectionClosedReason {
- /// [`RpcConnection`] is disconnect by server itself.
- Disconnected,
- /// [`RpcConnection`] has become idle and is disconnected by idle timeout.
- Idle,
-}
-
-impl Handler for Room {
- type Result = ();
-
- /// Removes [`RpcConnection`] of specified [`Member`] from the [`Room`].
- fn handle(&mut self, msg: RpcConnectionClosed, _: &mut Self::Context) {
- info!(
- "RpcConnectionClosed for member {}, reason {:?}",
- msg.member_id, msg.reason
- );
- self.connections.remove(&msg.member_id);
- }
-}
-
-/// Repository that stores [`Room`]s.
-#[derive(Clone, Default)]
-pub struct RoomsRepository {
- rooms: Arc>>>,
-}
-
-impl RoomsRepository {
- /// Creates new [`Room`]s repository with passed-in [`Room`]s.
- pub fn new(rooms: HashMap>) -> Self {
- Self {
- rooms: Arc::new(Mutex::new(rooms)),
- }
- }
-
- /// Returns [`Room`] by its ID.
- pub fn get(&self, id: Id) -> Option> {
- let rooms = self.rooms.lock().unwrap();
- rooms.get(&id).cloned()
- }
-}
diff --git a/src/api/client/rpc_connection.rs b/src/api/client/rpc_connection.rs
new file mode 100644
index 000000000..bd6026933
--- /dev/null
+++ b/src/api/client/rpc_connection.rs
@@ -0,0 +1,196 @@
+//! [`RpcConnection`] with related messages.
+use actix::Message;
+use futures::Future;
+
+use crate::api::{control::MemberId, protocol::Event};
+
+use std::fmt;
+
+/// Abstraction over RPC connection with some remote [`Member`].
+pub trait RpcConnection: fmt::Debug + Send {
+ /// Closes [`RpcConnection`].
+ /// No [`RpcConnectionClosed`] signals should be emitted.
+ /// Always returns success.
+ fn close(&mut self) -> Box>;
+
+ /// Sends [`Event`] to remote [`Member`].
+ fn send_event(
+ &self,
+ event: Event,
+ ) -> Box>;
+}
+
+/// Signal for authorizing new [`RpcConnection`] before establishing.
+#[derive(Debug, Message)]
+#[rtype(result = "Result<(), AuthorizationError>")]
+pub struct Authorize {
+ /// ID of [`Member`] to authorize [`RpcConnection`] for.
+ pub member_id: MemberId,
+ /// Credentials to authorize [`RpcConnection`] with.
+ pub credentials: String, // TODO: &str when futures will allow references
+}
+
+/// Error of authorization [`RpcConnection`] in [`Room`].
+#[derive(Debug)]
+pub enum AuthorizationError {
+ /// Authorizing [`Member`] does not exists in the [`Room`].
+ MemberNotExists,
+ /// Provided credentials are invalid.
+ InvalidCredentials,
+}
+
+/// Signal of new [`RpcConnection`] being established with specified [`Member`].
+/// Transport should consider dropping connection if message result is err.
+#[derive(Debug, Message)]
+#[rtype(result = "Result<(), ()>")]
+#[allow(clippy::module_name_repetitions)]
+pub struct RpcConnectionEstablished {
+ /// ID of [`Member`] that establishes [`RpcConnection`].
+ pub member_id: MemberId,
+ /// Established [`RpcConnection`].
+ pub connection: Box,
+}
+/// Signal of existing [`RpcConnection`] of specified [`Member`] being closed.
+#[derive(Debug, Message)]
+#[allow(clippy::module_name_repetitions)]
+pub struct RpcConnectionClosed {
+ /// ID of [`Member`] which [`RpcConnection`] is closed.
+ pub member_id: MemberId,
+ /// Reason of why [`RpcConnection`] is closed.
+ pub reason: ClosedReason,
+}
+
+/// Reasons of why [`RpcConnection`] may be closed.
+#[derive(Debug)]
+pub enum ClosedReason {
+ /// [`RpcConnection`] was irrevocably closed.
+ Closed,
+ /// [`RpcConnection`] was lost, but may be reestablished.
+ Lost,
+}
+
+#[cfg(test)]
+pub mod test {
+ use std::sync::{
+ atomic::{AtomicUsize, Ordering},
+ Arc, Mutex,
+ };
+
+ use actix::{
+ Actor, ActorContext, Addr, AsyncContext, Context, Handler, Message,
+ System,
+ };
+ use futures::future::Future;
+
+ use crate::{
+ api::{
+ client::rpc_connection::{
+ ClosedReason, RpcConnection, RpcConnectionClosed,
+ RpcConnectionEstablished,
+ },
+ control::MemberId,
+ protocol::{Command, Event},
+ },
+ signalling::Room,
+ };
+
+ /// [`RpcConnection`] impl convenient for testing.
+ #[derive(Debug, Clone)]
+ pub struct TestConnection {
+ pub member_id: MemberId,
+ pub room: Addr,
+ pub events: Arc>>,
+ pub stopped: Arc,
+ }
+
+ impl Actor for TestConnection {
+ type Context = Context;
+
+ fn started(&mut self, ctx: &mut Self::Context) {
+ self.room
+ .try_send(RpcConnectionEstablished {
+ member_id: self.member_id,
+ connection: Box::new(ctx.address()),
+ })
+ .unwrap();
+ }
+
+ fn stopped(&mut self, _ctx: &mut Self::Context) {
+ self.stopped.fetch_add(1, Ordering::Relaxed);
+ if self.stopped.load(Ordering::Relaxed) > 1 {
+ System::current().stop()
+ }
+ }
+ }
+
+ #[derive(Message)]
+ struct Close;
+
+ impl Handler for TestConnection {
+ type Result = ();
+
+ fn handle(&mut self, _: Close, ctx: &mut Self::Context) {
+ ctx.stop()
+ }
+ }
+
+ impl Handler for TestConnection {
+ type Result = ();
+
+ fn handle(&mut self, event: Event, _ctx: &mut Self::Context) {
+ let mut events = self.events.lock().unwrap();
+ events.push(serde_json::to_string(&event).unwrap());
+ match event {
+ Event::PeerCreated {
+ peer_id,
+ sdp_offer,
+ tracks: _,
+ } => {
+ match sdp_offer {
+ Some(_) => self.room.do_send(Command::MakeSdpAnswer {
+ peer_id,
+ sdp_answer: "responder_answer".into(),
+ }),
+ None => self.room.do_send(Command::MakeSdpOffer {
+ peer_id,
+ sdp_offer: "caller_offer".into(),
+ }),
+ }
+ self.room.do_send(Command::SetIceCandidate {
+ peer_id,
+ candidate: "ice_candidate".into(),
+ })
+ }
+ Event::IceCandidateDiscovered {
+ peer_id: _,
+ candidate: _,
+ } => {
+ self.room.do_send(RpcConnectionClosed {
+ member_id: self.member_id,
+ reason: ClosedReason::Closed,
+ });
+ }
+ Event::PeersRemoved { peer_ids: _ } => {}
+ Event::SdpAnswerMade {
+ peer_id: _,
+ sdp_answer: _,
+ } => {}
+ }
+ }
+ }
+
+ impl RpcConnection for Addr {
+ fn close(&mut self) -> Box> {
+ let fut = self.send(Close {}).map_err(|_| ());
+ Box::new(fut)
+ }
+
+ fn send_event(
+ &self,
+ event: Event,
+ ) -> Box> {
+ let fut = self.send(event).map_err(|_| ());
+ Box::new(fut)
+ }
+ }
+}
diff --git a/src/api/client/server.rs b/src/api/client/server.rs
index 989021e6e..604976eb7 100644
--- a/src/api/client/server.rs
+++ b/src/api/client/server.rs
@@ -10,13 +10,14 @@ use serde::Deserialize;
use crate::{
api::{
client::{
- AuthorizeRpcConnection, Id as RoomId, RoomsRepository,
- RpcConnectionAuthorizationError, WsSession,
+ rpc_connection::{AuthorizationError, Authorize},
+ session::WsSession,
},
- control::Id as MemberId,
+ control::MemberId,
},
conf::{Conf, Rpc},
log::prelude::*,
+ signalling::{RoomId, RoomsRepository},
};
/// Parameters of new WebSocket connection creation HTTP request.
@@ -39,13 +40,11 @@ fn ws_index(
State,
),
) -> FutureResponse {
- use RpcConnectionAuthorizationError::*;
-
debug!("Request params: {:?}", info);
match state.rooms.get(info.room_id) {
Some(room) => room
- .send(AuthorizeRpcConnection {
+ .send(Authorize {
member_id: info.member_id,
credentials: info.credentials.clone(),
})
@@ -59,8 +58,12 @@ fn ws_index(
state.config.idle_timeout,
),
),
- Err(MemberNotExists) => Ok(HttpResponse::NotFound().into()),
- Err(InvalidCredentials) => Ok(HttpResponse::Forbidden().into()),
+ Err(AuthorizationError::MemberNotExists) => {
+ Ok(HttpResponse::NotFound().into())
+ }
+ Err(AuthorizationError::InvalidCredentials) => {
+ Ok(HttpResponse::Forbidden().into())
+ }
})
.responder(),
None => future::ok(HttpResponse::NotFound().into()).responder(),
@@ -104,25 +107,24 @@ mod test {
use actix::Arbiter;
use actix_web::{http, test, App};
use futures::Stream;
- use hashbrown::HashMap;
use crate::{
- api::{client::Room, control::Member},
+ api::control::Member,
conf::{Conf, Server},
+ media::create_peers,
+ signalling::Room,
};
use super::*;
/// Creates [`RoomsRepository`] for tests filled with a single [`Room`].
- fn room() -> RoomsRepository {
+ fn room(conf: Rpc) -> RoomsRepository {
let members = hashmap! {
1 => Member{id: 1, credentials: "caller_credentials".into()},
2 => Member{id: 2, credentials: "responder_credentials".into()},
};
- let room = Arbiter::start(move |_| Room {
- id: 1,
- members,
- connections: HashMap::new(),
+ let room = Arbiter::start(move |_| {
+ Room::new(1, members, create_peers(1, 2), conf.reconnect_timeout)
});
let rooms = hashmap! {1 => room};
RoomsRepository::new(rooms)
@@ -132,7 +134,7 @@ mod test {
fn ws_server(conf: Conf) -> test::TestServer {
test::TestServer::with_factory(move || {
App::with_state(Context {
- rooms: room(),
+ rooms: room(conf.rpc.clone()),
config: conf.rpc.clone(),
})
.resource("/ws/{room_id}/{member_id}/{credentials}", |r| {
@@ -149,14 +151,15 @@ mod test {
write.text(r#"{"ping":33}"#);
let (item, _) = server.execute(read.into_future()).unwrap();
- assert_eq!(item, Some(ws::Message::Text(r#"{"pong":33}"#.into())));
+ assert_eq!(Some(ws::Message::Text(r#"{"pong":33}"#.into())), item);
}
#[test]
fn disconnects_on_idle() {
let conf = Conf {
rpc: Rpc {
- idle_timeout: Duration::new(1, 0),
+ idle_timeout: Duration::new(2, 0),
+ reconnect_timeout: Default::default(),
},
server: Server::default(),
};
@@ -167,14 +170,14 @@ mod test {
write.text(r#"{"ping":33}"#);
let (item, read) = server.execute(read.into_future()).unwrap();
- assert_eq!(item, Some(ws::Message::Text(r#"{"pong":33}"#.into())));
+ assert_eq!(Some(ws::Message::Text(r#"{"pong":33}"#.into())), item);
thread::sleep(conf.rpc.idle_timeout.add(Duration::from_secs(1)));
let (item, _) = server.execute(read.into_future()).unwrap();
assert_eq!(
- item,
- Some(ws::Message::Close(Some(ws::CloseCode::Normal.into())))
+ Some(ws::Message::Close(Some(ws::CloseCode::Normal.into()))),
+ item
);
}
}
diff --git a/src/api/client/session.rs b/src/api/client/session.rs
index 6698439c2..40e17eccd 100644
--- a/src/api/client/session.rs
+++ b/src/api/client/session.rs
@@ -1,22 +1,25 @@
//! WebSocket session.
-use std::time::Duration;
+use std::time::{Duration, Instant};
use actix::{
- fut::wrap_future, Actor, ActorContext, Addr, AsyncContext, Handler,
- Message, SpawnHandle, StreamHandler,
+ fut::wrap_future, Actor, ActorContext, ActorFuture, Addr, AsyncContext,
+ Handler, Message, StreamHandler,
};
-use actix_web::ws::{self, CloseReason};
-use futures::Future;
-use serde::{Deserialize, Serialize};
+use actix_web::ws::{self, CloseReason, WebsocketContext};
+use futures::future::Future;
use crate::{
- api::client::room::{
- Room, RpcConnection, RpcConnectionClosed, RpcConnectionClosedReason,
- RpcConnectionEstablished,
+ api::{
+ client::rpc_connection::{
+ ClosedReason, RpcConnection, RpcConnectionClosed,
+ RpcConnectionEstablished,
+ },
+ control::MemberId,
+ protocol::{ClientMsg, Event, ServerMsg},
},
- api::control::member::Id as MemberId,
log::prelude::*,
+ signalling::Room,
};
/// Long-running WebSocket connection of Client API.
@@ -25,17 +28,20 @@ use crate::{
pub struct WsSession {
/// ID of [`Member`] that WebSocket connection is associated with.
member_id: MemberId,
+
/// [`Room`] that [`Member`] is associated with.
room: Addr,
- /// Handle for watchdog which checks whether WebSocket client became
- /// idle (no `ping` messages received during [`idle_timeout`]).
- ///
- /// This one should be renewed on received `ping` message from client.
- idle_handler: Option,
- /// Timeout of receiving `ping` messages from client.
+ /// Timeout of receiving any messages from client.
idle_timeout: Duration,
+ /// Timestamp for watchdog which checks whether WebSocket client became
+ /// idle (no messages received during [`idle_timeout`]).
+ ///
+ /// This one should be renewed on any received WebSocket message
+ /// from client.
+ last_activity: Instant,
+
/// Indicates whether WebSocket connection is closed by server ot by
/// client.
closed_by_server: bool,
@@ -51,42 +57,39 @@ impl WsSession {
Self {
member_id,
room,
- idle_handler: None,
idle_timeout,
+ last_activity: Instant::now(),
closed_by_server: false,
}
}
- /// Resets idle handler watchdog.
- fn reset_idle_timeout(&mut self, ctx: &mut ::Context) {
- if let Some(handler) = self.idle_handler {
- ctx.cancel_future(handler);
- }
-
- self.idle_handler =
- Some(ctx.run_later(self.idle_timeout, |sess, ctx| {
- info!("WsConnection with member {} is idle", sess.member_id);
-
- let member_id = sess.member_id;
- ctx.wait(wrap_future(
- sess.room
- .send(RpcConnectionClosed {
- member_id,
- reason: RpcConnectionClosedReason::Idle,
- })
- .map_err(move |err| {
- error!(
- "WsSession of member {} failed to remove from \
- Room, because: {:?}",
- member_id, err,
- )
- }),
- ));
+ fn close_normal(&self, ctx: &mut WebsocketContext) {
+ ctx.notify(Close {
+ reason: Some(ws::CloseCode::Normal.into()),
+ });
+ }
- ctx.notify(Close {
- reason: Some(ws::CloseCode::Normal.into()),
- });
- }));
+ /// Start watchdog which will drop connection if now-last_activity >
+ /// idle_timeout.
+ fn start_watchdog(&mut self, ctx: &mut ::Context) {
+ ctx.run_interval(Duration::new(1, 0), |session, ctx| {
+ if Instant::now().duration_since(session.last_activity)
+ > session.idle_timeout
+ {
+ info!("WsSession of member {} is idle", session.member_id);
+ if let Err(err) = session.room.try_send(RpcConnectionClosed {
+ member_id: session.member_id,
+ reason: ClosedReason::Lost,
+ }) {
+ error!(
+ "WsSession of member {} failed to remove from Room, \
+ because: {:?}",
+ session.member_id, err,
+ )
+ }
+ session.close_normal(ctx);
+ }
+ });
}
}
@@ -100,24 +103,41 @@ impl Actor for WsSession {
fn started(&mut self, ctx: &mut Self::Context) {
debug!("Started WsSession for member {}", self.member_id);
- self.reset_idle_timeout(ctx);
+ self.start_watchdog(ctx);
let member_id = self.member_id;
- ctx.wait(wrap_future(
- self.room
- .send(RpcConnectionEstablished {
- member_id: self.member_id,
- connection: Box::new(ctx.address()),
- })
- .map(|_| ())
- .map_err(move |err| {
+ ctx.wait(
+ wrap_future(self.room.send(RpcConnectionEstablished {
+ member_id: self.member_id,
+ connection: Box::new(ctx.address()),
+ }))
+ .map(
+ move |auth_result,
+ session: &mut Self,
+ ctx: &mut ws::WebsocketContext| {
+ if let Err(e) = auth_result {
+ error!(
+ "Room rejected Established for member {}, cause \
+ {:?}",
+ member_id, e
+ );
+ session.close_normal(ctx);
+ }
+ },
+ )
+ .map_err(
+ move |send_err,
+ session: &mut Self,
+ ctx: &mut ws::WebsocketContext| {
error!(
"WsSession of member {} failed to join Room, because: \
{:?}",
- member_id, err,
- )
- }),
- ));
+ member_id, send_err,
+ );
+ session.close_normal(ctx);
+ },
+ ),
+ );
}
fn stopped(&mut self, _ctx: &mut Self::Context) {
@@ -127,12 +147,25 @@ impl Actor for WsSession {
impl RpcConnection for Addr {
/// Closes [`WsSession`] by sending itself "normal closure" close message.
+ ///
+ /// Never returns error.
fn close(&mut self) -> Box> {
let fut = self
.send(Close {
reason: Some(ws::CloseCode::Normal.into()),
})
- .map_err(|_| ());
+ .or_else(|_| Ok(()));
+ Box::new(fut)
+ }
+
+ /// Sends [`Event`] to Web Client.
+ fn send_event(
+ &self,
+ event: Event,
+ ) -> Box> {
+ let fut = self
+ .send(ServerMsg::Event(event))
+ .map_err(|err| error!("Failed send event {:?} ", err));
Box::new(fut)
}
}
@@ -155,29 +188,13 @@ impl Handler for WsSession {
}
}
-/// Message for keeping client WebSocket connection alive.
-#[derive(Debug, Deserialize, Message, Serialize)]
-pub enum Heartbeat {
- /// `ping` message that WebSocket client is expected to send to the server
- /// periodically.
- #[serde(rename = "ping")]
- Ping(usize),
- /// `pong` message that server answers with to WebSocket client in response
- /// to received `ping` message.
- #[serde(rename = "pong")]
- Pong(usize),
-}
-
-impl Handler for WsSession {
+impl Handler for WsSession {
type Result = ();
- /// Answers with `Heartbeat::Pong` message to WebSocket client in response
- /// to the received `Heartbeat::Ping` message.
- fn handle(&mut self, msg: Heartbeat, ctx: &mut Self::Context) {
- if let Heartbeat::Ping(n) = msg {
- trace!("Received ping: {}", n);
- ctx.text(serde_json::to_string(&Heartbeat::Pong(n)).unwrap())
- }
+ /// Sends [`Event`] to Web Client.
+ fn handle(&mut self, msg: ServerMsg, ctx: &mut Self::Context) {
+ debug!("Event {:?} for member {}", msg, self.member_id);
+ ctx.text(serde_json::to_string(&msg).unwrap())
}
}
@@ -190,32 +207,39 @@ impl StreamHandler for WsSession {
);
match msg {
ws::Message::Text(text) => {
- self.reset_idle_timeout(ctx);
- if let Ok(msg) = serde_json::from_str::(&text) {
- ctx.notify(msg);
+ self.last_activity = Instant::now();
+ match serde_json::from_str::(&text) {
+ Ok(ClientMsg::Ping(n)) => {
+ trace!("Received ping: {}", n);
+ // Answer with Heartbeat::Pong.
+ ctx.notify(ServerMsg::Pong(n));
+ }
+ Ok(ClientMsg::Command(command)) => {
+ if let Err(err) = self.room.try_send(command) {
+ error!(
+ "Cannot send Command to Room {}, because {}",
+ self.member_id, err
+ )
+ }
+ }
+ Err(err) => error!(
+ "Error [{:?}] parsing client message [{}]",
+ err, &text
+ ),
}
}
ws::Message::Close(reason) => {
if !self.closed_by_server {
- debug!(
- "Send close frame with reason {:?} for member {}",
- reason, self.member_id
- );
- let member_id = self.member_id;
- ctx.wait(wrap_future(
- self.room
- .send(RpcConnectionClosed {
- member_id: self.member_id,
- reason: RpcConnectionClosedReason::Disconnected,
- })
- .map_err(move |err| {
- error!(
- "WsSession of member {} failed to remove \
- from Room, because: {:?}",
- member_id, err,
- )
- }),
- ));
+ if let Err(err) = self.room.try_send(RpcConnectionClosed {
+ member_id: self.member_id,
+ reason: ClosedReason::Closed,
+ }) {
+ error!(
+ "WsSession of member {} failed to remove from \
+ Room, because: {:?}",
+ self.member_id, err,
+ )
+ };
ctx.close(reason);
ctx.stop();
}
diff --git a/src/api/control/member.rs b/src/api/control/member.rs
index 57734dfde..880e666aa 100644
--- a/src/api/control/member.rs
+++ b/src/api/control/member.rs
@@ -8,6 +8,7 @@ pub type Id = u64;
pub struct Member {
/// ID of [`Member`].
pub id: Id,
+
/// Credentials to authorize [`Member`] with.
pub credentials: String,
}
diff --git a/src/api/control/mod.rs b/src/api/control/mod.rs
index b29bf968d..db3c38fcd 100644
--- a/src/api/control/mod.rs
+++ b/src/api/control/mod.rs
@@ -1,5 +1,5 @@
//! Implementation of Control API.
-pub mod member;
+mod member;
-pub use self::member::*;
+pub use self::member::{Id as MemberId, Member};
diff --git a/src/api/mod.rs b/src/api/mod.rs
index 1c66e73e5..f12d438ca 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -2,3 +2,4 @@
pub mod client;
pub mod control;
+pub mod protocol;
diff --git a/src/api/protocol.rs b/src/api/protocol.rs
new file mode 100644
index 000000000..34848e063
--- /dev/null
+++ b/src/api/protocol.rs
@@ -0,0 +1,283 @@
+use actix::Message;
+use serde::{de::Deserializer, ser::Serializer, Deserialize, Serialize};
+
+// TODO: should be properly shared between medea and jason
+#[cfg_attr(test, derive(PartialEq))]
+#[derive(Message, Debug)]
+#[allow(dead_code)]
+/// Message sent by `Media Server` to `Client`.
+pub enum ServerMsg {
+ /// `pong` message that server answers with to WebSocket client in response
+ /// to received `ping` message.
+ Pong(u64),
+ /// `Media Server` notifies `Client` about happened facts and it reacts on
+ /// them to reach the proper state.
+ Event(Event),
+}
+
+#[cfg_attr(test, derive(PartialEq, Debug))]
+#[allow(dead_code)]
+/// Message from 'Client' to 'Media Server'.
+pub enum ClientMsg {
+ /// `ping` message that WebSocket client is expected to send to the server
+ /// periodically.
+ Ping(u64),
+ /// Request of `Web Client` to change the state on `Media Server`.
+ Command(Command),
+}
+
+/// WebSocket message from Web Client to Media Server.
+#[derive(Deserialize, Serialize, Message)]
+#[cfg_attr(test, derive(PartialEq, Debug))]
+#[serde(tag = "command", content = "data")]
+#[allow(dead_code)]
+#[rtype(result = "Result<(), ()>")]
+pub enum Command {
+ /// Web Client sends SDP Offer.
+ MakeSdpOffer { peer_id: u64, sdp_offer: String },
+ /// Web Client sends SDP Answer.
+ MakeSdpAnswer { peer_id: u64, sdp_answer: String },
+ /// Web Client sends Ice Candidate.
+ SetIceCandidate { peer_id: u64, candidate: String },
+}
+
+/// WebSocket message from Medea to Jason.
+#[derive(Deserialize, Serialize, Message, Debug)]
+#[cfg_attr(test, derive(PartialEq))]
+#[allow(dead_code)]
+#[serde(tag = "event", content = "data")]
+pub enum Event {
+ /// Media Server notifies Web Client about necessity of RTCPeerConnection
+ /// creation.
+ PeerCreated {
+ peer_id: u64,
+ sdp_offer: Option,
+ tracks: Vec,
+ },
+ /// Media Server notifies Web Client about necessity to apply specified SDP
+ /// Answer to Web Client's RTCPeerConnection.
+ SdpAnswerMade { peer_id: u64, sdp_answer: String },
+
+ /// Media Server notifies Web Client about necessity to apply specified
+ /// ICE Candidate.
+ IceCandidateDiscovered { peer_id: u64, candidate: String },
+
+ /// Media Server notifies Web Client about necessity of RTCPeerConnection
+ /// close.
+ PeersRemoved { peer_ids: Vec },
+}
+
+/// [`Track] with specified direction.
+#[derive(Deserialize, Serialize, Debug)]
+#[cfg_attr(test, derive(PartialEq))]
+pub struct Directional {
+ pub id: u64,
+ pub direction: Direction,
+ pub media_type: MediaType,
+}
+
+/// Direction of [`Track`].
+#[derive(Deserialize, Serialize, Debug)]
+#[cfg_attr(test, derive(PartialEq))]
+pub enum Direction {
+ Send { receivers: Vec },
+ Recv { sender: u64 },
+}
+
+/// Type of [`Track`].
+#[derive(Deserialize, Serialize, Debug, Clone)]
+#[cfg_attr(test, derive(PartialEq))]
+pub enum MediaType {
+ Audio(AudioSettings),
+ Video(VideoSettings),
+}
+
+#[derive(Deserialize, Serialize, Debug, Clone)]
+#[cfg_attr(test, derive(PartialEq))]
+pub struct AudioSettings {}
+
+#[derive(Deserialize, Serialize, Debug, Clone)]
+#[cfg_attr(test, derive(PartialEq))]
+pub struct VideoSettings {}
+
+impl Serialize for ClientMsg {
+ fn serialize(&self, serializer: S) -> Result
+ where
+ S: Serializer,
+ {
+ use serde::ser::SerializeStruct;
+
+ match self {
+ ClientMsg::Ping(n) => {
+ let mut ping = serializer.serialize_struct("ping", 1)?;
+ ping.serialize_field("ping", n)?;
+ ping.end()
+ }
+ ClientMsg::Command(command) => command.serialize(serializer),
+ }
+ }
+}
+
+impl<'de> Deserialize<'de> for ClientMsg {
+ fn deserialize(deserializer: D) -> Result
+ where
+ D: Deserializer<'de>,
+ {
+ use serde::de::Error;
+
+ let ev = serde_json::Value::deserialize(deserializer)?;
+ let map = ev.as_object().ok_or_else(|| {
+ Error::custom(format!("unable to deser ClientMsg [{:?}]", &ev))
+ })?;
+
+ if let Some(v) = map.get("ping") {
+ let n = v.as_u64().ok_or_else(|| {
+ Error::custom(format!(
+ "unable to deser ClientMsg::Ping [{:?}]",
+ &ev
+ ))
+ })?;
+
+ Ok(ClientMsg::Ping(n))
+ } else {
+ let command =
+ serde_json::from_value::(ev).map_err(|e| {
+ Error::custom(format!(
+ "unable to deser ClientMsg::Command [{:?}]",
+ e
+ ))
+ })?;
+ Ok(ClientMsg::Command(command))
+ }
+ }
+}
+
+impl Serialize for ServerMsg {
+ fn serialize(&self, serializer: S) -> Result
+ where
+ S: Serializer,
+ {
+ use serde::ser::SerializeStruct;
+
+ match self {
+ ServerMsg::Pong(n) => {
+ let mut ping = serializer.serialize_struct("pong", 1)?;
+ ping.serialize_field("pong", n)?;
+ ping.end()
+ }
+ ServerMsg::Event(command) => command.serialize(serializer),
+ }
+ }
+}
+
+impl<'de> Deserialize<'de> for ServerMsg {
+ fn deserialize(deserializer: D) -> Result
+ where
+ D: Deserializer<'de>,
+ {
+ use serde::de::Error;
+
+ let ev = serde_json::Value::deserialize(deserializer)?;
+ let map = ev.as_object().ok_or_else(|| {
+ Error::custom(format!("unable to deser ServerMsg [{:?}]", &ev))
+ })?;
+
+ if let Some(v) = map.get("pong") {
+ let n = v.as_u64().ok_or_else(|| {
+ Error::custom(format!(
+ "unable to deser ServerMsg::Pong [{:?}]",
+ &ev
+ ))
+ })?;
+
+ Ok(ServerMsg::Pong(n))
+ } else {
+ let event = serde_json::from_value::(ev).map_err(|e| {
+ Error::custom(format!(
+ "unable to deser ServerMsg::Event [{:?}]",
+ e
+ ))
+ })?;
+ Ok(ServerMsg::Event(event))
+ }
+ }
+}
+
+#[cfg(test)]
+mod test {
+ use crate::api::protocol::{ClientMsg, Command, Event, ServerMsg};
+
+ #[test]
+ fn command() {
+ let command = ClientMsg::Command(Command::MakeSdpOffer {
+ peer_id: 77,
+ sdp_offer: "offer".to_owned(),
+ });
+ #[cfg_attr(nightly, rustfmt::skip)]
+ let command_str =
+ "{\
+ \"command\":\"MakeSdpOffer\",\
+ \"data\":{\
+ \"peer_id\":77,\
+ \"sdp_offer\":\"offer\"\
+ }\
+ }";
+
+ assert_eq!(command_str, serde_json::to_string(&command).unwrap());
+ assert_eq!(
+ command,
+ serde_json::from_str(&serde_json::to_string(&command).unwrap())
+ .unwrap()
+ );
+ }
+
+ #[test]
+ fn ping() {
+ let ping = ClientMsg::Ping(15);
+ let ping_str = "{\"ping\":15}";
+
+ assert_eq!(ping_str, serde_json::to_string(&ping).unwrap());
+ assert_eq!(
+ ping,
+ serde_json::from_str(&serde_json::to_string(&ping).unwrap())
+ .unwrap()
+ )
+ }
+
+ #[test]
+ fn event() {
+ let event = ServerMsg::Event(Event::SdpAnswerMade {
+ peer_id: 45,
+ sdp_answer: "answer".to_owned(),
+ });
+ #[cfg_attr(nightly, rustfmt::skip)]
+ let event_str =
+ "{\
+ \"event\":\"SdpAnswerMade\",\
+ \"data\":{\
+ \"peer_id\":45,\
+ \"sdp_answer\":\"answer\"\
+ }\
+ }";
+
+ assert_eq!(event_str, serde_json::to_string(&event).unwrap());
+ assert_eq!(
+ event,
+ serde_json::from_str(&serde_json::to_string(&event).unwrap())
+ .unwrap()
+ );
+ }
+
+ #[test]
+ fn pong() {
+ let pong = ServerMsg::Pong(5);
+ let pong_str = "{\"pong\":5}";
+
+ assert_eq!(pong_str, serde_json::to_string(&pong).unwrap());
+ assert_eq!(
+ pong,
+ serde_json::from_str(&serde_json::to_string(&pong).unwrap())
+ .unwrap()
+ )
+ }
+}
diff --git a/src/conf/rpc.rs b/src/conf/rpc.rs
index 0327e2225..e3fdda999 100644
--- a/src/conf/rpc.rs
+++ b/src/conf/rpc.rs
@@ -12,4 +12,10 @@ pub struct Rpc {
#[default(Duration::from_secs(10))]
#[serde(with = "serde_humantime")]
pub idle_timeout: Duration,
+
+ /// Duration, after which the server deletes the client session if
+ /// the remote RPC client does not reconnect after it is idle.
+ #[default(Duration::from_secs(10))]
+ #[serde(with = "serde_humantime")]
+ pub reconnect_timeout: Duration,
}
diff --git a/src/main.rs b/src/main.rs
index fd42507b1..aab1c2a74 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,23 +1,22 @@
//! Medea media server application.
#[macro_use]
-mod utils;
-
+pub mod utils;
pub mod api;
pub mod conf;
pub mod log;
+pub mod media;
+pub mod signalling;
use actix::prelude::*;
use dotenv::dotenv;
-use hashbrown::HashMap;
use log::prelude::*;
use crate::{
- api::{
- client::{server, Room, RoomsRepository},
- control::Member,
- },
+ api::{client::server, control::Member},
conf::Conf,
+ media::create_peers,
+ signalling::{Room, RoomsRepository},
};
fn main() {
@@ -28,22 +27,20 @@ fn main() {
let sys = System::new("medea");
+ let config = Conf::parse().unwrap();
+
+ info!("{:?}", config);
+
let members = hashmap! {
1 => Member{id: 1, credentials: "caller_credentials".to_owned()},
2 => Member{id: 2, credentials: "responder_credentials".to_owned()},
};
- let room = Arbiter::start(move |_| Room {
- id: 1,
- members,
- connections: HashMap::new(),
- });
+ let peers = create_peers(1, 2);
+ let room = Room::new(1, members, peers, config.rpc.reconnect_timeout);
+ let room = Arbiter::start(move |_| room);
let rooms = hashmap! {1 => room};
let rooms_repo = RoomsRepository::new(rooms);
- let config = Conf::parse().unwrap();
-
- info!("{:?}", config);
-
server::run(rooms_repo, config);
let _ = sys.run();
}
diff --git a/src/media/mod.rs b/src/media/mod.rs
new file mode 100644
index 000000000..ed48bf622
--- /dev/null
+++ b/src/media/mod.rs
@@ -0,0 +1,11 @@
+//! Representations of media and media connection establishment objects.
+pub mod peer;
+pub mod track;
+
+pub use self::{
+ peer::{
+ create_peers, Id as PeerId, New, Peer, PeerStateError,
+ PeerStateMachine, WaitLocalHaveRemote, WaitLocalSdp, WaitRemoteSdp,
+ },
+ track::{Id as TrackId, Track},
+};
diff --git a/src/media/peer.rs b/src/media/peer.rs
new file mode 100644
index 000000000..a01b7bb2c
--- /dev/null
+++ b/src/media/peer.rs
@@ -0,0 +1,389 @@
+//! Remote [`RTCPeerConnection`][1] representation.
+//!
+//! [1]: https://www.w3.org/TR/webrtc/#rtcpeerconnection-interface
+
+#![allow(clippy::use_self)]
+use failure::Fail;
+use hashbrown::HashMap;
+
+use std::{convert::TryFrom, fmt::Display, sync::Arc};
+
+use crate::{
+ api::{
+ control::MemberId,
+ protocol::{
+ AudioSettings, Direction, Directional, MediaType, VideoSettings,
+ },
+ },
+ media::{Track, TrackId},
+};
+
+/// Newly initialized [`Peer`] ready to signalling.
+#[derive(Debug, PartialEq)]
+pub struct New {}
+
+/// [`Peer`] doesnt have remote SDP and is waiting for local SDP.
+#[derive(Debug, PartialEq)]
+pub struct WaitLocalSdp {}
+
+/// [`Peer`] has remote SDP and is waiting for local SDP.
+#[derive(Debug, PartialEq)]
+pub struct WaitLocalHaveRemote {}
+
+/// [`Peer`] has local SDP and is waiting for remote SDP.
+#[derive(Debug, PartialEq)]
+pub struct WaitRemoteSdp {}
+
+/// SDP exchange ended.
+#[derive(Debug, PartialEq)]
+pub struct Stable {}
+
+/// Produced when unwrapping [`PeerStateMachine`] to [`Peer`] with wrong state.
+#[derive(Fail, Debug)]
+#[allow(clippy::module_name_repetitions)]
+pub enum PeerStateError {
+ #[fail(
+ display = "Cannot unwrap Peer from PeerStateMachine [id = {}]. \
+ Expected state {} was {}",
+ _0, _1, _2
+ )]
+ WrongState(Id, &'static str, String),
+}
+
+impl PeerStateError {
+ pub fn new_wrong_state(
+ peer: &PeerStateMachine,
+ expected: &'static str,
+ ) -> Self {
+ PeerStateError::WrongState(peer.id(), expected, format!("{}", peer))
+ }
+}
+
+/// Implementation of ['Peer'] state machine.
+#[derive(Debug)]
+#[allow(clippy::module_name_repetitions)]
+pub enum PeerStateMachine {
+ New(Peer),
+ WaitLocalSdp(Peer),
+ WaitLocalHaveRemote(Peer),
+ WaitRemoteSdp(Peer),
+ Stable(Peer),
+}
+
+// TODO: macro to remove boilerplate
+impl PeerStateMachine {
+ /// Returns ID of [`Peer`].
+ pub fn id(&self) -> Id {
+ match self {
+ PeerStateMachine::New(peer) => peer.id(),
+ PeerStateMachine::WaitLocalSdp(peer) => peer.id(),
+ PeerStateMachine::WaitLocalHaveRemote(peer) => peer.id(),
+ PeerStateMachine::WaitRemoteSdp(peer) => peer.id(),
+ PeerStateMachine::Stable(peer) => peer.id(),
+ }
+ }
+
+ /// Returns ID of [`Member`] associated with this [`Peer`].
+ pub fn member_id(&self) -> MemberId {
+ match self {
+ PeerStateMachine::New(peer) => peer.member_id(),
+ PeerStateMachine::WaitLocalSdp(peer) => peer.member_id(),
+ PeerStateMachine::WaitLocalHaveRemote(peer) => peer.member_id(),
+ PeerStateMachine::WaitRemoteSdp(peer) => peer.member_id(),
+ PeerStateMachine::Stable(peer) => peer.member_id(),
+ }
+ }
+
+ /// Returns ID of interconnected [`Peer`].
+ pub fn partner_peer_id(&self) -> Id {
+ match self {
+ PeerStateMachine::New(peer) => peer.partner_peer_id(),
+ PeerStateMachine::WaitLocalSdp(peer) => peer.partner_peer_id(),
+ PeerStateMachine::WaitLocalHaveRemote(peer) => {
+ peer.partner_peer_id()
+ }
+ PeerStateMachine::WaitRemoteSdp(peer) => peer.partner_peer_id(),
+ PeerStateMachine::Stable(peer) => peer.partner_peer_id(),
+ }
+ }
+
+ /// Returns ID of interconnected [`Member`].
+ pub fn partner_member_id(&self) -> Id {
+ match self {
+ PeerStateMachine::New(peer) => peer.partner_peer_id(),
+ PeerStateMachine::WaitLocalSdp(peer) => peer.partner_peer_id(),
+ PeerStateMachine::WaitLocalHaveRemote(peer) => {
+ peer.partner_peer_id()
+ }
+ PeerStateMachine::WaitRemoteSdp(peer) => peer.partner_peer_id(),
+ PeerStateMachine::Stable(peer) => peer.partner_peer_id(),
+ }
+ }
+}
+
+impl Display for PeerStateMachine {
+ fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+ match self {
+ PeerStateMachine::WaitRemoteSdp(_) => write!(f, "WaitRemoteSdp"),
+ PeerStateMachine::New(_) => write!(f, "New"),
+ PeerStateMachine::WaitLocalSdp(_) => write!(f, "WaitLocalSdp"),
+ PeerStateMachine::WaitLocalHaveRemote(_) => {
+ write!(f, "WaitLocalHaveRemote")
+ }
+ PeerStateMachine::Stable(_) => write!(f, "Stable"),
+ }
+ }
+}
+
+macro_rules! impl_peer_converts {
+ ($peer_type:tt) => {
+ impl<'a> TryFrom<&'a PeerStateMachine> for &'a Peer<$peer_type> {
+ type Error = PeerStateError;
+
+ fn try_from(
+ peer: &'a PeerStateMachine,
+ ) -> Result {
+ match peer {
+ PeerStateMachine::$peer_type(peer) => Ok(peer),
+ _ => Err(PeerStateError::WrongState(
+ 1,
+ stringify!($peer_type),
+ format!("{}", peer),
+ )),
+ }
+ }
+ }
+
+ impl TryFrom for Peer<$peer_type> {
+ type Error = PeerStateError;
+
+ fn try_from(peer: PeerStateMachine) -> Result {
+ match peer {
+ PeerStateMachine::$peer_type(peer) => Ok(peer),
+ _ => Err(PeerStateError::WrongState(
+ 1,
+ stringify!($peer_type),
+ format!("{}", peer),
+ )),
+ }
+ }
+ }
+
+ impl From> for PeerStateMachine {
+ fn from(peer: Peer<$peer_type>) -> Self {
+ PeerStateMachine::$peer_type(peer)
+ }
+ }
+ };
+}
+
+impl_peer_converts!(New);
+impl_peer_converts!(WaitLocalSdp);
+impl_peer_converts!(WaitLocalHaveRemote);
+impl_peer_converts!(WaitRemoteSdp);
+impl_peer_converts!(Stable);
+
+/// ID of [`Peer`].
+pub type Id = u64;
+
+#[derive(Debug)]
+pub struct Context {
+ id: Id,
+ member_id: MemberId,
+ partner_peer: Id,
+ partner_member: MemberId,
+ sdp_offer: Option,
+ sdp_answer: Option,
+ receivers: HashMap>,
+ senders: HashMap>,
+}
+
+/// [`RTCPeerConnection`] representation.
+#[derive(Debug)]
+pub struct Peer {
+ context: Context,
+ state: S,
+}
+
+impl Peer {
+ /// Returns ID of [`Member`] associated with this [`Peer`].
+ pub fn member_id(&self) -> MemberId {
+ self.context.member_id
+ }
+
+ /// Returns ID of [`Peer`].
+ pub fn id(&self) -> Id {
+ self.context.id
+ }
+
+ /// Returns ID of interconnected [`Peer`].
+ pub fn partner_peer_id(&self) -> Id {
+ self.context.partner_peer
+ }
+
+ /// Returns ID of interconnected [`Member`].
+ pub fn partner_member_id(&self) -> Id {
+ self.context.partner_member
+ }
+
+ /// Returns [`Track`]'s of [`Peer`].
+ pub fn tracks(&self) -> Vec {
+ let tracks = self.context.senders.iter().fold(
+ vec![],
+ |mut tracks, (_, track)| {
+ tracks.push(Directional {
+ id: track.id,
+ media_type: track.media_type.clone(),
+ direction: Direction::Send {
+ receivers: vec![self.context.partner_peer],
+ },
+ });
+ tracks
+ },
+ );
+ self.context
+ .receivers
+ .iter()
+ .fold(tracks, |mut tracks, (_, track)| {
+ tracks.push(Directional {
+ id: track.id,
+ media_type: track.media_type.clone(),
+ direction: Direction::Recv {
+ sender: self.context.partner_peer,
+ },
+ });
+ tracks
+ })
+ }
+
+ pub fn is_sender(&self) -> bool {
+ !self.context.senders.is_empty()
+ }
+}
+
+impl Peer {
+ /// Creates new [`Peer`] for [`Member`].
+ pub fn new(
+ id: Id,
+ member_id: MemberId,
+ partner_peer: Id,
+ partner_member: MemberId,
+ ) -> Self {
+ let context = Context {
+ id,
+ member_id,
+ partner_peer,
+ partner_member,
+ sdp_offer: None,
+ sdp_answer: None,
+ receivers: HashMap::new(),
+ senders: HashMap::new(),
+ };
+ Self {
+ context,
+ state: New {},
+ }
+ }
+
+ /// Transition new [`Peer`] into state of waiting for local description.
+ pub fn start(self) -> Peer {
+ Peer {
+ context: self.context,
+ state: WaitLocalSdp {},
+ }
+ }
+
+ /// Transition new [`Peer`] into state of waiting for remote description.
+ pub fn set_remote_sdp(
+ self,
+ sdp_offer: String,
+ ) -> Peer {
+ let mut context = self.context;
+ context.sdp_offer = Some(sdp_offer);
+ Peer {
+ context,
+ state: WaitLocalHaveRemote {},
+ }
+ }
+
+ /// Add [`Track`] to [`Peer`] for send.
+ pub fn add_sender(&mut self, track: Arc