use clause in subscriptions and proto

Author: Larkooo

Cargo.lock

@@ -1526,10 +1526,10 @@ impl proto::world::world_server::World for DojoWorld {
         &self,
         request: Request<SubscribeEntitiesRequest>,
     ) -> ServiceResult<Self::SubscribeEntitiesStream> {
-        let SubscribeEntitiesRequest { clauses } = request.into_inner();
+        let SubscribeEntitiesRequest { clause } = request.into_inner();
         let rx = self
             .entity_manager
-            .add_subscriber(clauses.into_iter().map(|keys| keys.into()).collect())
+            .add_subscriber(clause.into())
             .await
             .map_err(|e| Status::internal(e.to_string()))?;
 
@@ -1544,12 +1544,12 @@ impl proto::world::world_server::World for DojoWorld {
     ) -> ServiceResult<()> {
         let UpdateEntitiesSubscriptionRequest {
             subscription_id,
-            clauses,
+            clause,
         } = request.into_inner();
         self.entity_manager
             .update_subscriber(
                 subscription_id,
-                clauses.into_iter().map(|keys| keys.into()).collect(),
+                clause.into(),
             )
             .await;
 
@@ -1626,10 +1626,10 @@ impl proto::world::world_server::World for DojoWorld {
         &self,
         request: Request<SubscribeEventMessagesRequest>,
     ) -> ServiceResult<Self::SubscribeEntitiesStream> {
-        let SubscribeEventMessagesRequest { clauses } = request.into_inner();
+        let SubscribeEventMessagesRequest { clause } = request.into_inner();
         let rx = self
             .event_message_manager
-            .add_subscriber(clauses.into_iter().map(|keys| keys.into()).collect())
+            .add_subscriber(clause.into())
             .await
             .map_err(|e| Status::internal(e.to_string()))?;
 
@@ -1644,12 +1644,12 @@ impl proto::world::world_server::World for DojoWorld {
     ) -> ServiceResult<()> {
         let UpdateEventMessagesSubscriptionRequest {
             subscription_id,
-            clauses,
+            clause,
         } = request.into_inner();
         self.event_message_manager
             .update_subscriber(
                 subscription_id,
-                clauses.into_iter().map(|keys| keys.into()).collect(),
+                clause.into(),
             )
             .await;
 

crates/grpc/server/src/lib.rs

@@ -19,16 +19,16 @@ use torii_sqlite::simple_broker::SimpleBroker;
 use torii_sqlite::types::OptimisticEntity;
 use tracing::{error, trace};
 
-use super::match_entity_keys;
+use super::{match_entity};
 use torii_proto::proto::world::SubscribeEntityResponse;
-use torii_proto::EntityKeysClause;
+use torii_proto::Clause;
 
 pub(crate) const LOG_TARGET: &str = "torii::grpc::server::subscriptions::entity";
 
 #[derive(Debug)]
 pub struct EntitiesSubscriber {
-    /// Entity ids that the subscriber is interested in
-    pub(crate) clauses: Vec<EntityKeysClause>,
+    /// The clause that the subscriber is interested in
+    pub(crate) clause: Clause,
     /// The channel to send the response back to the subscriber.
     pub(crate) sender: Sender<Result<SubscribeEntityResponse, tonic::Status>>,
 }
@@ -40,7 +40,7 @@ pub struct EntityManager {
 impl EntityManager {
     pub async fn add_subscriber(
         &self,
-        clauses: Vec<EntityKeysClause>,
+        clause: Clause,
     ) -> Result<Receiver<Result<SubscribeEntityResponse, tonic::Status>>, Error> {
         let subscription_id = rand::thread_rng().gen::<u64>();
         let (sender, receiver) = channel(1);
@@ -58,12 +58,12 @@ impl EntityManager {
         self.subscribers
             .write()
             .await
-            .insert(subscription_id, EntitiesSubscriber { clauses, sender });
+            .insert(subscription_id, EntitiesSubscriber { clause, sender });
 
         Ok(receiver)
     }
 
-    pub async fn update_subscriber(&self, id: u64, clauses: Vec<EntityKeysClause>) {
+    pub async fn update_subscriber(&self, id: u64, clause: Clause) {
         let sender = {
             let subscribers = self.subscribers.read().await;
             if let Some(subscriber) = subscribers.get(&id) {
@@ -76,7 +76,7 @@ impl EntityManager {
         self.subscribers
             .write()
             .await
-            .insert(id, EntitiesSubscriber { clauses, sender });
+            .insert(id, EntitiesSubscriber { clause, sender });
     }
 
     pub(super) async fn remove_subscriber(&self, id: u64) {
@@ -144,7 +144,7 @@ impl Service {
 
             // If we have a clause of keys, then check that the key pattern of the entity
             // matches the key pattern of the subscriber.
-            if !match_entity_keys(hashed, &keys, &entity.updated_model, &sub.clauses) {
+            if !match_entity(hashed, &keys, &entity.updated_model, &sub.clause) {
                 continue;
             }
 

crates/grpc/server/src/subscriptions/entity.rs

@@ -13,6 +13,7 @@ use tokio::sync::mpsc::{
     channel, unbounded_channel, Receiver, Sender, UnboundedReceiver, UnboundedSender,
 };
 use tokio::sync::RwLock;
+use torii_proto::KeysClause;
 use torii_sqlite::constants::SQL_FELT_DELIMITER;
 use torii_sqlite::error::{Error, ParseError};
 use torii_sqlite::simple_broker::SimpleBroker;
@@ -22,14 +23,13 @@ use tracing::{error, trace};
 use super::match_keys;
 use torii_proto::proto::types::Event as ProtoEvent;
 use torii_proto::proto::world::SubscribeEventsResponse;
-use torii_proto::EntityKeysClause;
 
 pub(crate) const LOG_TARGET: &str = "torii::grpc::server::subscriptions::event";
 
 #[derive(Debug)]
 pub struct EventSubscriber {
     /// Event keys that the subscriber is interested in
-    keys: Vec<EntityKeysClause>,
+    keys: Vec<KeysClause>,
     /// The channel to send the response back to the subscriber.
     sender: Sender<Result<SubscribeEventsResponse, tonic::Status>>,
 }
@@ -42,7 +42,7 @@ pub struct EventManager {
 impl EventManager {
     pub async fn add_subscriber(
         &self,
-        keys: Vec<EntityKeysClause>,
+        keys: Vec<KeysClause>,
     ) -> Result<Receiver<Result<SubscribeEventsResponse, tonic::Status>>, Error> {
         let id = rand::thread_rng().gen::<usize>();
         let (sender, receiver) = channel(1);

crates/grpc/server/src/subscriptions/event.rs

@@ -14,22 +14,23 @@ use tokio::sync::mpsc::{
 };
 use tokio::sync::RwLock;
 use torii_proto::proto::types::Entity;
-use torii_proto::EntityKeysClause;
+use torii_proto::Clause;
 use torii_sqlite::constants::SQL_FELT_DELIMITER;
 use torii_sqlite::error::{Error, ParseError};
 use torii_sqlite::simple_broker::SimpleBroker;
 use torii_sqlite::types::OptimisticEventMessage;
 use tracing::{error, trace};
 
-use super::match_entity_keys;
 use torii_proto::proto::world::SubscribeEntityResponse;
 
+use super::match_entity;
+
 pub(crate) const LOG_TARGET: &str = "torii::grpc::server::subscriptions::event_message";
 
 #[derive(Debug)]
 pub struct EventMessageSubscriber {
-    /// Entity ids that the subscriber is interested in
-    pub(crate) clauses: Vec<EntityKeysClause>,
+    /// The clause that the subscriber is interested in
+    pub(crate) clause: Clause,
     /// The channel to send the response back to the subscriber.
     pub(crate) sender: Sender<Result<SubscribeEntityResponse, tonic::Status>>,
 }
@@ -42,7 +43,7 @@ pub struct EventMessageManager {
 impl EventMessageManager {
     pub async fn add_subscriber(
         &self,
-        clauses: Vec<EntityKeysClause>,
+        clause: Clause,
     ) -> Result<Receiver<Result<SubscribeEntityResponse, tonic::Status>>, Error> {
         let subscription_id = rand::thread_rng().gen::<u64>();
         let (sender, receiver) = channel(1);
@@ -60,12 +61,12 @@ impl EventMessageManager {
         self.subscribers
             .write()
             .await
-            .insert(subscription_id, EventMessageSubscriber { clauses, sender });
+            .insert(subscription_id, EventMessageSubscriber { clause, sender });
 
         Ok(receiver)
     }
 
-    pub async fn update_subscriber(&self, id: u64, clauses: Vec<EntityKeysClause>) {
+    pub async fn update_subscriber(&self, id: u64, clause: Clause) {
         let sender = {
             let subscribers = self.subscribers.read().await;
             if let Some(subscriber) = subscribers.get(&id) {
@@ -78,7 +79,7 @@ impl EventMessageManager {
         self.subscribers
             .write()
             .await
-            .insert(id, EventMessageSubscriber { clauses, sender });
+            .insert(id, EventMessageSubscriber { clause, sender });
     }
 
     pub(super) async fn remove_subscriber(&self, id: u64) {
@@ -138,7 +139,7 @@ impl Service {
 
             // If we have a clause of keys, then check that the key pattern of the entity
             // matches the key pattern of the subscriber.
-            if !match_entity_keys(hashed, &keys, &entity.updated_model, &sub.clauses) {
+            if !match_entity(hashed, &keys, &entity.updated_model, &sub.clause) {
                 continue;
             }