Fix subscribers limit check (#127)

src/app/endpoint/rtc.rs

@@ -291,7 +291,7 @@ impl RequestHandler for ConnectHandler {
                 .rtc_id(payload.id)
                 .execute(&conn)?;
 
-            match maybe_rtc_stream {
+            let backend = match maybe_rtc_stream {
                 Some(ref stream) => db::janus_backend::FindQuery::new()
                     .id(stream.backend_id().to_owned())
                     .execute(&conn)?
@@ -300,7 +300,22 @@ impl RequestHandler for ConnectHandler {
                 None => db::janus_backend::least_loaded(room.id(), &conn)?
                     .ok_or("no available backends")
                     .status(ResponseStatus::SERVICE_UNAVAILABLE)?,
+            };
+
+            // Check that the backend's subscribers limit is not reached.
+            if let Some(limit) = backend.subscribers_limit() {
+                let agents_count = db::agent::CountQuery::new()
+                    .room_id(room.id())
+                    .status(db::agent::Status::Ready)
+                    .execute(&conn)?;
+
+                if agents_count >= limit.into() {
+                    return Err("subscribers limit reached")
+                        .status(ResponseStatus::SERVICE_UNAVAILABLE);
+                }
             }
+
+            backend
         };
 
         // Send janus handle creation request.

src/db/agent.rs

@@ -126,6 +126,53 @@ impl<'a> ListQuery<'a> {
 
 ////////////////////////////////////////////////////////////////////////////////
 
+pub(crate) struct CountQuery {
+    room_id: Option<Uuid>,
+    status: Option<Status>,
+}
+
+impl CountQuery {
+    pub(crate) fn new() -> Self {
+        Self {
+            room_id: None,
+            status: None,
+        }
+    }
+
+    pub(crate) fn room_id(self, room_id: Uuid) -> Self {
+        Self {
+            room_id: Some(room_id),
+            ..self
+        }
+    }
+
+    pub(crate) fn status(self, status: Status) -> Self {
+        Self {
+            status: Some(status),
+            ..self
+        }
+    }
+
+    pub(crate) fn execute(&self, conn: &PgConnection) -> Result<i64, Error> {
+        use diesel::dsl::count;
+        use diesel::prelude::*;
+
+        let mut q = agent::table.into_boxed();
+
+        if let Some(room_id) = self.room_id {
+            q = q.filter(agent::room_id.eq(room_id));
+        }
+
+        if let Some(status) = self.status {
+            q = q.filter(agent::status.eq(status));
+        }
+
+        q.select(count(agent::agent_id)).first(conn)
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
 #[derive(Debug, Insertable)]
 #[table_name = "agent"]
 pub(crate) struct InsertQuery<'a> {

src/db/janus_backend.rs

@@ -30,6 +30,10 @@ impl Object {
     pub(crate) fn session_id(&self) -> i64 {
         self.session_id
     }
+
+    pub(crate) fn subscribers_limit(&self) -> Option<i32> {
+        self.subscribers_limit
+    }
 }
 
 ////////////////////////////////////////////////////////////////////////////////