[refactor] #1981, #4195: Refactor the event filters

client/benches/tps/utils.rs

@@ -172,9 +172,8 @@ impl MeasurerUnit {
         let listener = self.client.clone();
         let (init_sender, init_receiver) = mpsc::channel();
         let event_filter = PipelineEventFilter::new()
-            .entity_kind(PipelineEntityKind::Block)
-            .status_kind(PipelineStatusKind::Committed)
-            .into();
+            .for_entity(PipelineEntityKind::Block)
+            .for_status(PipelineStatusKind::Committed);
         let blocks_expected = self.config.blocks as usize;
         let name = self.name;
         let handle = thread::spawn(move || -> Result<()> {

client/src/client.rs

@@ -323,7 +323,7 @@ impl_query_output! {
     crate::data_model::metadata::MetadataValueBox,
     crate::data_model::query::TransactionQueryOutput,
     crate::data_model::permission::PermissionTokenSchema,
-    crate::data_model::trigger::Trigger<crate::data_model::events::TriggeringFilterBox>,
+    crate::data_model::trigger::Trigger<crate::data_model::events::TriggeringEventFilterBox>,
     crate::data_model::prelude::Numeric,
 }
 
@@ -605,9 +605,7 @@ impl Client {
             let mut event_iterator = {
                 let event_iterator_result = tokio::time::timeout_at(
                     deadline,
-                    self.listen_for_events_async(
-                        PipelineEventFilter::new().hash(hash.into()).into(),
-                    ),
+                    self.listen_for_events_async(PipelineEventFilter::new().for_hash(hash.into())),
                 )
                 .await
                 .map_err(Into::into)
@@ -905,8 +903,9 @@ impl Client {
     /// - Forwards from [`events_api::EventIterator::new`]
     pub fn listen_for_events(
         &self,
-        event_filter: FilterBox,
+        event_filter: impl Into<EventFilterBox>,
     ) -> Result<impl Iterator<Item = Result<Event>>> {
+        let event_filter = event_filter.into();
         iroha_logger::trace!(?event_filter);
         events_api::EventIterator::new(self.events_handler(event_filter)?)
     }
@@ -918,8 +917,9 @@ impl Client {
     /// - Forwards from [`events_api::AsyncEventStream::new`]
     pub async fn listen_for_events_async(
         &self,
-        event_filter: FilterBox,
+        event_filter: impl Into<EventFilterBox> + Send,
     ) -> Result<AsyncEventStream> {
+        let event_filter = event_filter.into();
         iroha_logger::trace!(?event_filter, "Async listening with");
         events_api::AsyncEventStream::new(self.events_handler(event_filter)?).await
     }
@@ -929,9 +929,12 @@ impl Client {
     /// # Errors
     /// Fails if handler construction fails
     #[inline]
-    pub fn events_handler(&self, event_filter: FilterBox) -> Result<events_api::flow::Init> {
+    pub fn events_handler(
+        &self,
+        event_filter: impl Into<EventFilterBox>,
+    ) -> Result<events_api::flow::Init> {
         events_api::flow::Init::new(
-            event_filter,
+            event_filter.into(),
             self.headers.clone(),
             self.torii_url
                 .join(torii_uri::SUBSCRIPTION)
@@ -1235,7 +1238,7 @@ pub mod events_api {
         /// Initialization struct for Events API flow.
         pub struct Init {
             /// Event filter
-            filter: FilterBox,
+            filter: EventFilterBox,
             /// HTTP request headers
             headers: HashMap<String, String>,
             /// TORII URL
@@ -1249,7 +1252,7 @@ pub mod events_api {
             /// Fails if [`transform_ws_url`] fails.
             #[inline]
             pub(in super::super) fn new(
-                filter: FilterBox,
+                filter: EventFilterBox,
                 headers: HashMap<String, String>,
                 url: Url,
             ) -> Result<Self> {

client/tests/integration/domain_owner_permissions.rs

@@ -309,11 +309,9 @@ fn domain_owner_trigger_permissions() -> Result<()> {
             trigger_instructions,
             Repeats::from(2_u32),
             bob_id,
-            // FIXME: due to restriction in `ExecuteTriggerEventFilter` it's impossible to execute trigger on behalf of multiple users
-            TriggeringFilterBox::ExecuteTrigger(ExecuteTriggerEventFilter::new(
-                trigger_id.clone(),
-                alice_id,
-            )),
+            TriggeringEventFilterBox::ExecuteTrigger(
+                ExecuteTriggerEventFilter::new().for_trigger(trigger_id.clone()),
+            ),
         ),
     ));
     test_client.submit_blocking(register_trigger)?;

client/tests/integration/events/data.rs

@@ -138,7 +138,7 @@ fn transaction_execution_should_produce_events(
     let listener = client.clone();
     let (init_sender, init_receiver) = mpsc::channel();
     let (event_sender, event_receiver) = mpsc::channel();
-    let event_filter = DataEventFilter::AcceptAll.into();
+    let event_filter = DataEventFilter::Any;
     thread::spawn(move || -> Result<()> {
         let event_iterator = listener.listen_for_events(event_filter)?;
         init_sender.send(())?;
@@ -182,7 +182,7 @@ fn produce_multiple_events() -> Result<()> {
     let listener = client.clone();
     let (init_sender, init_receiver) = mpsc::channel();
     let (event_sender, event_receiver) = mpsc::channel();
-    let event_filter = DataEventFilter::AcceptAll.into();
+    let event_filter = DataEventFilter::Any;
     thread::spawn(move || -> Result<()> {
         let event_iterator = listener.listen_for_events(event_filter)?;
         init_sender.send(())?;

client/tests/integration/events/notification.rs

@@ -21,10 +21,11 @@ fn trigger_completion_success_should_produce_event() -> Result<()> {
             vec![InstructionBox::from(instruction)],
             Repeats::Indefinitely,
             asset_id.account_id.clone(),
-            TriggeringFilterBox::ExecuteTrigger(ExecuteTriggerEventFilter::new(
-                trigger_id.clone(),
-                asset_id.account_id,
-            )),
+            TriggeringEventFilterBox::ExecuteTrigger(
+                ExecuteTriggerEventFilter::new()
+                    .for_trigger(trigger_id.clone())
+                    .under_authority(asset_id.account_id),
+            ),
         ),
     ));
     test_client.submit_blocking(register_trigger)?;
@@ -35,11 +36,9 @@ fn trigger_completion_success_should_produce_event() -> Result<()> {
     let (sender, receiver) = mpsc::channel();
     let _handle = thread::spawn(move || -> Result<()> {
         let mut event_it = thread_client.listen_for_events(
-            NotificationEventFilter::TriggerCompleted(TriggerCompletedEventFilter::new(
-                Some(trigger_id),
-                Some(TriggerCompletedOutcomeType::Success),
-            ))
-            .into(),
+            TriggerCompletedEventFilter::new()
+                .for_trigger(trigger_id)
+                .for_outcome(TriggerCompletedOutcomeType::Success),
         )?;
         if event_it.next().is_some() {
             sender.send(())?;
@@ -70,10 +69,11 @@ fn trigger_completion_failure_should_produce_event() -> Result<()> {
             vec![InstructionBox::from(instruction)],
             Repeats::Indefinitely,
             account_id.clone(),
-            TriggeringFilterBox::ExecuteTrigger(ExecuteTriggerEventFilter::new(
-                trigger_id.clone(),
-                account_id,
-            )),
+            TriggeringEventFilterBox::ExecuteTrigger(
+                ExecuteTriggerEventFilter::new()
+                    .for_trigger(trigger_id.clone())
+                    .under_authority(account_id),
+            ),
         ),
     ));
     test_client.submit_blocking(register_trigger)?;
@@ -84,11 +84,9 @@ fn trigger_completion_failure_should_produce_event() -> Result<()> {
     let (sender, receiver) = mpsc::channel();
     let _handle = thread::spawn(move || -> Result<()> {
         let mut event_it = thread_client.listen_for_events(
-            NotificationEventFilter::TriggerCompleted(TriggerCompletedEventFilter::new(
-                Some(trigger_id),
-                Some(TriggerCompletedOutcomeType::Failure),
-            ))
-            .into(),
+            TriggerCompletedEventFilter::new()
+                .for_trigger(trigger_id)
+                .for_outcome(TriggerCompletedOutcomeType::Failure),
         )?;
         if event_it.next().is_some() {
             sender.send(())?;

client/tests/integration/events/pipeline.rs

@@ -82,12 +82,12 @@ impl Checker {
         thread::spawn(move || {
             let mut event_iterator = self
                 .listener
-                .listen_for_events(FilterBox::Pipeline(
+                .listen_for_events(
                     PipelineEventFilter::new()
-                        .entity_kind(PipelineEntityKind::Transaction)
-                        .status_kind(status_kind)
-                        .hash(*self.hash),
-                ))
+                        .for_entity(PipelineEntityKind::Transaction)
+                        .for_status(status_kind)
+                        .for_hash(*self.hash),
+                )
                 .expect("Failed to create event iterator.");
             let event_result = event_iterator.next().expect("Stream closed");
             let _event = event_result.expect("Must be valid");
@@ -101,9 +101,8 @@ fn committed_block_must_be_available_in_kura() {
     wait_for_genesis_committed(&[client.clone()], 0);
 
     let event_filter = PipelineEventFilter::new()
-        .entity_kind(PipelineEntityKind::Block)
-        .status_kind(PipelineStatusKind::Committed)
-        .into();
+        .for_entity(PipelineEntityKind::Block)
+        .for_status(PipelineStatusKind::Committed);
     let mut event_iter = client
         .listen_for_events(event_filter)
         .expect("Failed to subscribe for events");