Add Rooms, Room, Publisher and Subscriber actors and basic rooms logic (71e64df3) · Commits · Orm Finnendahl / WASP

server/src/main.rs

+3 −0

Original line number	Diff line number	Diff line
		@@ -3,7 +3,10 @@
		// Licensed under the MIT license, see the LICENSE file or <http://opensource.org/licenses/MIT>

		mod config;
		mod publisher;
		mod rooms;
		mod server;
		mod subscriber;

		use anyhow::Error;

server/src/publisher.rs

0 → 100644

+96 −0

Original line number	Diff line number	Diff line
		// Copyright (C) 2020 Sebastian Dröge <sebastian@centricular.com>
		//
		// Licensed under the MIT license, see the LICENSE file or <http://opensource.org/licenses/MIT>

		use crate::config::Config;
		use crate::rooms::{Room, Rooms};
		use crate::subscriber::Subscriber;

		use std::sync::{Arc, Mutex};

		use anyhow::Error;

		use actix::{Actor, Addr, Handler, Message, StreamHandler};

		use actix_web_actors::ws;

		use log::debug;

		/// Actor that represents a WebRTC publisher.
		#[derive(Debug)]
		pub struct Publisher {
		cfg: Arc<Config>,
		rooms: Addr<Rooms>,
		room: Mutex<Option<Addr<Room>>>,
		}

		impl Publisher {
		/// Create a new `Publisher` actor.
		pub fn new(cfg: Arc<Config>, rooms: Addr<Rooms>) -> Self {
		Publisher {
		cfg,
		rooms,
		room: Mutex::new(None),
		}
		}
		}

		impl Actor for Publisher {
		type Context = ws::WebsocketContext<Self>;
		}

		impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for Publisher {
		fn handle(&mut self, _msg: Result<ws::Message, ws::ProtocolError>, _ctx: &mut Self::Context) {
		// TODO
		}
		}

		/// New `Subscriber` joining the `Room` of this `Publisher`.
		#[derive(Debug)]
		pub struct NewSubscriberMessage {
		pub subscriber: Addr<Subscriber>,
		}

		impl Message for NewSubscriberMessage {
		type Result = Result<(), Error>;
		}

		impl Handler<NewSubscriberMessage> for Publisher {
		type Result = Result<(), Error>;

		fn handle(
		&mut self,
		msg: NewSubscriberMessage,
		_ctx: &mut ws::WebsocketContext<Self>,
		) -> Self::Result {
		debug!("New subscriber {:?} joining", msg.subscriber);

		// TODO
		Ok(())
		}
		}

		/// Existing `Subscriber` leaving the `Room` of this `Publisher`.
		#[derive(Debug)]
		pub struct LeavingSubscriberMessage {
		pub subscriber: Addr<Subscriber>,
		}

		impl Message for LeavingSubscriberMessage {
		type Result = Result<(), Error>;
		}

		impl Handler<LeavingSubscriberMessage> for Publisher {
		type Result = Result<(), Error>;

		fn handle(
		&mut self,
		msg: LeavingSubscriberMessage,
		_ctx: &mut ws::WebsocketContext<Self>,
		) -> Self::Result {
		debug!("Subscriber {:?} leaving", msg.subscriber);

		// TODO
		Ok(())
		}
		}

server/src/rooms.rs

0 → 100644

+302 −0

Original line number	Diff line number	Diff line
		// Copyright (C) 2020 Sebastian Dröge <sebastian@centricular.com>
		//
		// Licensed under the MIT license, see the LICENSE file or <http://opensource.org/licenses/MIT>

		use std::collections::{HashMap, HashSet};
		use std::sync::Mutex;

		use actix::{Actor, ActorContext, Addr, AsyncContext, Context, Handler, Message, MessageResult};
		use anyhow::{bail, Error};
		use log::{debug, error, info};

		use crate::publisher::{self, Publisher};
		use crate::subscriber::{self, Subscriber};

		/// Unique identifier for a `Room`.
		#[derive(Debug, PartialEq, Eq, Copy, Clone, Hash)]
		pub struct RoomId(u64);

		/// Message to create a new `Room` for a `Publisher`.
		#[derive(Debug)]
		pub struct CreateRoomMessage {
		pub publisher: Addr<Publisher>,
		pub name: String,
		pub description: Option<String>,
		}

		impl Message for CreateRoomMessage {
		type Result = Result<Addr<Room>, Error>;
		}

		/// Message to find an existing `Room`.
		#[derive(Debug)]
		pub struct FindRoomMessage {
		pub room_id: RoomId,
		}

		impl Message for FindRoomMessage {
		type Result = Option<Addr<Room>>;
		}

		/// Message to list all currently existing `Room`s.
		#[derive(Debug)]
		pub struct ListRoomsMessage;

		impl Message for ListRoomsMessage {
		type Result = Vec<Addr<Room>>;
		}

		/// Sent from a `Room` to remove itself from the list of `Rooms` after the `Publisher` deleted it.
		#[derive(Debug)]
		struct RoomDeletedMessage {
		room_id: RoomId,
		}

		impl Message for RoomDeletedMessage {
		type Result = ();
		}

		/// Actor that keeps track of all currently existing `Room`s.
		#[derive(Debug)]
		pub struct Rooms {
		rooms: Mutex<HashMap<RoomId, Addr<Room>>>,
		}

		impl Rooms {
		/// Create a new `Rooms` instance.
		pub fn new() -> Self {
		Self {
		rooms: Mutex::new(HashMap::new()),
		}
		}
		}

		impl Actor for Rooms {
		type Context = Context<Self>;
		}

		impl Handler<CreateRoomMessage> for Rooms {
		type Result = Result<Addr<Room>, Error>;

		fn handle(&mut self, msg: CreateRoomMessage, ctx: &mut Context<Self>) -> Self::Result {
		info!("Creating new room for message {:?}", msg);

		let mut rooms = self.rooms.lock().unwrap();

		let room = Room::new(ctx.address(), msg.name, msg.description, msg.publisher);
		let room_id = room.id;

		info!("Created new room {:?}", room_id);

		let room_addr = room.start();
		rooms.insert(room_id, room_addr.clone());

		Ok(room_addr)
		}
		}

		impl Handler<FindRoomMessage> for Rooms {
		type Result = Option<Addr<Room>>;

		fn handle(&mut self, msg: FindRoomMessage, _ctx: &mut Context<Self>) -> Self::Result {
		debug!("Finding room {:?}", msg.room_id);

		let rooms = self.rooms.lock().unwrap();
		rooms.get(&msg.room_id).cloned()
		}
		}

		impl Handler<ListRoomsMessage> for Rooms {
		type Result = MessageResult<ListRoomsMessage>;

		fn handle(&mut self, _msg: ListRoomsMessage, _ctx: &mut Context<Self>) -> Self::Result {
		debug!("Listing all current rooms");

		MessageResult(self.rooms.lock().unwrap().values().cloned().collect())
		}
		}

		impl Handler<RoomDeletedMessage> for Rooms {
		type Result = ();

		fn handle(&mut self, msg: RoomDeletedMessage, _ctx: &mut Context<Self>) -> Self::Result {
		let mut rooms = self.rooms.lock().unwrap();

		info!("Room {:?} destroyed", msg.room_id);

		rooms.remove(&msg.room_id).expect("Room not found");
		}
		}

		/// Request `RoomInformation` from a given `Room`.
		#[derive(Debug)]
		pub struct RoomInformationMessage;

		/// Room information returned from the `RoomInformationMessage`.
		#[derive(Debug)]
		pub struct RoomInformation {
		pub id: RoomId,
		pub name: String,
		pub description: Option<String>,
		}

		impl Message for RoomInformationMessage {
		type Result = RoomInformation;
		}

		/// Delete a `Room` from a `Publisher`.
		#[derive(Debug)]
		pub struct DeleteRoomMessage {
		pub publisher: Addr<Publisher>,
		}

		impl Message for DeleteRoomMessage {
		type Result = Result<(), Error>;
		}

		/// Join a `Room` by a `Subscriber`.
		#[derive(Debug)]
		pub struct JoinRoomMessage {
		pub subscriber: Addr<Subscriber>,
		}

		impl Message for JoinRoomMessage {
		type Result = Result<(), Error>;
		}

		/// Leave a `Room` by a `Subscriber`.
		#[derive(Debug)]
		pub struct LeaveRoomMessage {
		pub subscriber: Addr<Subscriber>,
		}

		impl Message for LeaveRoomMessage {
		type Result = Result<(), Error>;
		}

		/// Actor that manages a `Room` with its `Publisher` and `Subscriber`s.
		#[derive(Debug)]
		pub struct Room {
		id: RoomId,
		name: String,
		description: Option<String>,

		rooms: Addr<Rooms>,

		publisher: Addr<Publisher>,
		subscribers: Mutex<HashSet<Addr<Subscriber>>>,
		}

		impl Room {
		/// Returns a next valid `RoomId`.
		fn next_room_id() -> RoomId {
		// XXX: We assume that no more than 2**64 rooms are ever created
		use std::sync::atomic::{self, AtomicU64};

		static ROOM_ID: AtomicU64 = AtomicU64::new(0);

		RoomId(ROOM_ID.fetch_add(1, atomic::Ordering::Relaxed))
		}

		/// Create a new `Room`.
		fn new(
		rooms: Addr<Rooms>,
		name: String,
		description: Option<String>,
		publisher: Addr<Publisher>,
		) -> Self {
		Room {
		rooms,
		id: Self::next_room_id(),
		name,
		description,
		publisher,
		subscribers: Mutex::new(HashSet::new()),
		}
		}
		}

		impl Actor for Room {
		type Context = Context<Self>;
		}

		impl Handler<DeleteRoomMessage> for Room {
		type Result = Result<(), Error>;

		fn handle(&mut self, msg: DeleteRoomMessage, ctx: &mut Context<Self>) -> Self::Result {
		if self.publisher == msg.publisher {
		info!("Deleting room {:?}", self.id);

		{
		let subscribers = self.subscribers.lock().unwrap();
		for subscriber in &*subscribers {
		subscriber.do_send(subscriber::RoomDeletedMessage);
		}
		}
		self.rooms.do_send(RoomDeletedMessage { room_id: self.id });

		ctx.stop();
		Ok(())
		} else {
		error!(
		"Tried to delete room {:?} from wrong publisher {:?}",
		self.id, msg.publisher
		);
		bail!("Deleting room {:?} not permitted", self.id)
		}
		}
		}

		impl Handler<JoinRoomMessage> for Room {
		type Result = Result<(), Error>;

		fn handle(&mut self, msg: JoinRoomMessage, _ctx: &mut Context<Self>) -> Self::Result {
		info!(
		"Joining room {:?} by subscriber {:?}",
		self.id, msg.subscriber
		);

		self.publisher.do_send(publisher::NewSubscriberMessage {
		subscriber: msg.subscriber.clone(),
		});

		let mut subscribers = self.subscribers.lock().unwrap();
		subscribers.insert(msg.subscriber);

		Ok(())
		}
		}

		impl Handler<LeaveRoomMessage> for Room {
		type Result = Result<(), Error>;

		fn handle(&mut self, msg: LeaveRoomMessage, _ctx: &mut Context<Self>) -> Self::Result {
		info!(
		"Leaving room {:?} by subscriber {:?}",
		self.id, msg.subscriber
		);

		self.publisher.do_send(publisher::LeavingSubscriberMessage {
		subscriber: msg.subscriber.clone(),
		});

		let mut subscribers = self.subscribers.lock().unwrap();
		subscribers.remove(&msg.subscriber);

		Ok(())
		}
		}

		impl Handler<RoomInformationMessage> for Room {
		type Result = MessageResult<RoomInformationMessage>;

		fn handle(&mut self, _msg: RoomInformationMessage, _ctx: &mut Context<Self>) -> Self::Result {
		debug!("Returning room information for room {:?}", self.id);

		MessageResult(RoomInformation {
		id: self.id,
		name: self.name.clone(),
		description: self.description.clone(),
		})
		}
		}

server/src/subscriber.rs

0 → 100644

+65 −0

Original line number	Diff line number	Diff line
		// Copyright (C) 2020 Sebastian Dröge <sebastian@centricular.com>
		//
		// Licensed under the MIT license, see the LICENSE file or <http://opensource.org/licenses/MIT>

		use crate::config::Config;
		use crate::rooms::{Room, Rooms};

		use std::sync::{Arc, Mutex};

		use actix::{Actor, Addr, Handler, Message, StreamHandler};

		use actix_web_actors::ws;

		use log::debug;

		/// Actor that represents a WebRTC subscriber.
		#[derive(Debug)]
		pub struct Subscriber {
		cfg: Arc<Config>,
		rooms: Addr<Rooms>,
		room: Mutex<Option<Addr<Room>>>,
		}

		impl Subscriber {
		/// Create a new `Subscriber` actor.
		pub fn new(cfg: Arc<Config>, rooms: Addr<Rooms>) -> Self {
		Subscriber {
		cfg,
		rooms,
		room: Mutex::new(None),
		}
		}
		}

		impl Actor for Subscriber {
		type Context = ws::WebsocketContext<Self>;
		}

		impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for Subscriber {
		fn handle(&mut self, _msg: Result<ws::Message, ws::ProtocolError>, _ctx: &mut Self::Context) {
		// TODO
		}
		}

		/// Leave a `Room` by a `Subscriber`.
		#[derive(Debug)]
		pub struct RoomDeletedMessage;

		impl Message for RoomDeletedMessage {
		type Result = ();
		}

		impl Handler<RoomDeletedMessage> for Subscriber {
		type Result = ();

		fn handle(
		&mut self,
		_msg: RoomDeletedMessage,
		_ctx: &mut ws::WebsocketContext<Self>,
		) -> Self::Result {
		debug!("Room deleted");

		// TODO
		}
		}