Merge branch 'master' into chore/add-code-owner

src/dx/subscribe/builders/subscription.rs

@@ -505,7 +505,9 @@ impl Subscription {
 
     fn subscribed_for_update(&self, update: &Update) -> bool {
         self.channels.contains(&update.channel())
-            || self.channel_groups.contains(&update.channel_group())
+            || update
+                .channel_group()
+                .is_some_and(|g| self.channel_groups.contains(&g))
     }
 }
 

src/dx/subscribe/event_engine/effect_handler.rs

@@ -61,20 +61,24 @@ impl EffectHandler<SubscribeEffectInvocation, SubscribeEffect> for SubscribeEffe
             SubscribeEffectInvocation::Handshake {
                 channels,
                 channel_groups,
+                cursor,
             } => Some(SubscribeEffect::Handshake {
                 channels: channels.clone(),
                 channel_groups: channel_groups.clone(),
+                cursor: cursor.clone(),
                 executor: self.subscribe_call.clone(),
                 cancellation_channel: self.cancellation_channel.clone(),
             }),
             SubscribeEffectInvocation::HandshakeReconnect {
                 channels,
                 channel_groups,
+                cursor,
                 attempts,
                 reason,
             } => Some(SubscribeEffect::HandshakeReconnect {
                 channels: channels.clone(),
                 channel_groups: channel_groups.clone(),
+                cursor: cursor.clone(),
                 attempts: *attempts,
                 reason: reason.clone(),
                 retry_policy: self.retry_policy.clone(),

src/dx/subscribe/event_engine/effects/emit_messages.rs

@@ -29,7 +29,7 @@ mod should {
             sender: Some("test-user".into()),
             timestamp: 1234567890,
             channel: "test".to_string(),
-            subscription: "test-group".to_string(),
+            subscription: Some("test-group".to_string()),
             data: vec![],
             r#type: None,
             space_id: None,

src/dx/subscribe/event_engine/effects/handshake.rs

@@ -53,8 +53,8 @@ mod should {
         let mock_handshake_function: Arc<SubscribeEffectExecutor> = Arc::new(move |params| {
             assert_eq!(params.channels, &Some(vec!["ch1".to_string()]));
             assert_eq!(params.channel_groups, &Some(vec!["cg1".to_string()]));
-            assert_eq!(params.attempt, 0);
             assert_eq!(params.cursor, None);
+            assert_eq!(params.attempt, 0);
             assert_eq!(params.reason, None);
             assert_eq!(params.effect_id, "id");
 

src/dx/subscribe/event_engine/effects/handshake_reconnection.rs

@@ -64,8 +64,8 @@ mod should {
         let mock_handshake_function: Arc<SubscribeEffectExecutor> = Arc::new(move |params| {
             assert_eq!(params.channels, &Some(vec!["ch1".to_string()]));
             assert_eq!(params.channel_groups, &Some(vec!["cg1".to_string()]));
-            assert_eq!(params.attempt, 1);
             assert_eq!(params.cursor, None);
+            assert_eq!(params.attempt, 1);
             assert_eq!(
                 params.reason.unwrap(),
                 PubNubError::Transport {

src/dx/subscribe/event_engine/effects/mod.rs

@@ -45,6 +45,12 @@ pub(crate) enum SubscribeEffect {
         /// after initial subscription completion.
         channel_groups: Option<Vec<String>>,
 
+        /// Time cursor.
+        ///
+        /// Cursor used by subscription loop to identify point in time after
+        /// which updates will be delivered.
+        cursor: Option<SubscribeCursor>,
+
         /// Executor function.
         ///
         /// Function which will be used to execute initial subscription.
@@ -70,6 +76,12 @@ pub(crate) enum SubscribeEffect {
         /// initial subscription.
         channel_groups: Option<Vec<String>>,
 
+        /// Time cursor.
+        ///
+        /// Cursor used by subscription loop to identify point in time after
+        /// which updates will be delivered.
+        cursor: Option<SubscribeCursor>,
+
         /// Current initial subscribe retry attempt.
         ///
         /// Used to track overall number of initial subscription retry attempts.
@@ -280,7 +292,8 @@ impl Effect for SubscribeEffect {
                     channels,
                     channel_groups,
                     *attempts,
-                    reason.clone(), // TODO: Does run function need to borrow self? Or we can consume it?
+                    reason.clone(), /* TODO: Does run function need to borrow self? Or we can
+                                     * consume it? */
                     &self.id(),
                     retry_policy,
                     executor,
@@ -309,7 +322,8 @@ impl Effect for SubscribeEffect {
                     channel_groups,
                     cursor,
                     *attempts,
-                    reason.clone(), // TODO: Does run function need to borrow self? Or we can consume it?
+                    reason.clone(), /* TODO: Does run function need to borrow self? Or we can
+                                     * consume it? */
                     &self.id(),
                     retry_policy,
                     executor,
@@ -357,12 +371,13 @@ mod should {
     use super::*;
 
     #[tokio::test]
-    async fn send_cancelation_notification() {
+    async fn send_cancellation_notification() {
         let (tx, rx) = async_channel::bounded(1);
 
         let effect = SubscribeEffect::Handshake {
             channels: None,
             channel_groups: None,
+            cursor: None,
             executor: Arc::new(|_| {
                 Box::pin(async move {
                     Ok(SubscribeResult {

src/dx/subscribe/event_engine/invocation.rs

@@ -31,6 +31,12 @@ pub(crate) enum SubscribeEffectInvocation {
         /// List of channel groups which will be source of real-time updates
         /// after initial subscription completion.
         channel_groups: Option<Vec<String>>,
+
+        /// Time cursor.
+        ///
+        /// Cursor used by subscription loop to identify point in time after
+        /// which updates will be delivered.
+        cursor: Option<SubscribeCursor>,
     },
 
     /// Cancel initial subscribe effect invocation.
@@ -50,6 +56,12 @@ pub(crate) enum SubscribeEffectInvocation {
         /// initial subscription.
         channel_groups: Option<Vec<String>>,
 
+        /// Time cursor.
+        ///
+        /// Cursor used by subscription loop to identify point in time after
+        /// which updates will be delivered.
+        cursor: Option<SubscribeCursor>,
+
         /// Current initial subscribe retry attempt.
         ///
         /// Used to track overall number of initial subscription retry attempts.