chore: reformat the code

crates/arkflow-plugin/src/buffer/common_window.rs

@@ -0,0 +1,193 @@
+/*
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+//! Common window infrastructure
+//!
+//! This module provides shared utilities for window implementations
+//! to reduce code duplication across different window types.
+
+use crate::buffer::join::JoinConfig;
+use crate::buffer::window::BaseWindow;
+use arkflow_core::{Error, Resource};
+use std::sync::Arc;
+use std::time::Instant;
+use tokio::sync::Notify;
+use tokio::time::sleep;
+use tokio_util::sync::CancellationToken;
+
+/// Common window context
+///
+/// Provides shared infrastructure for window implementations including
+/// timer management, notification, and cancellation handling.
+pub struct CommonWindowContext {
+    /// Notification mechanism for signaling between threads
+    pub notify: Arc<Notify>,
+    /// Token for cancellation of background tasks
+    pub close_token: CancellationToken,
+    /// Last time the window was triggered
+    pub last_trigger: Arc<std::sync::RwLock<Instant>>,
+}
+
+impl CommonWindowContext {
+    /// Create a new common window context
+    ///
+    /// # Returns
+    ///
+    /// A new CommonWindowContext instance
+    pub fn new() -> Self {
+        Self {
+            notify: Arc::new(Notify::new()),
+            close_token: CancellationToken::new(),
+            last_trigger: Arc::new(std::sync::RwLock::new(Instant::now())),
+        }
+    }
+
+    /// Start the background timer task
+    ///
+    /// This spawns a background task that periodically notifies waiters
+    /// based on the check interval. The task runs until cancelled.
+    ///
+    /// # Arguments
+    ///
+    /// * `check_interval` - How often to check if window should trigger
+    pub fn start_timer(&self, check_interval: std::time::Duration) {
+        let notify = Arc::clone(&self.notify);
+        let close = self.close_token.clone();
+
+        tokio::spawn(async move {
+            loop {
+                let timer = sleep(check_interval);
+                tokio::select! {
+                    _ = timer => {
+                        notify.notify_waiters();
+                    }
+                    _ = close.cancelled() => {
+                        notify.notify_waiters();
+                        break;
+                    }
+                    _ = notify.notified() => {
+                        if close.is_cancelled(){
+                            break;
+                        }
+                    }
+                }
+            }
+        });
+    }
+
+    /// Update the last trigger time
+    ///
+    /// # Arguments
+    ///
+    /// * `time` - The new last trigger time
+    pub fn update_last_trigger(&self, time: Instant) {
+        if let Ok(mut last) = self.last_trigger.write() {
+            *last = time;
+        }
+    }
+
+    /// Get the last trigger time
+    ///
+    /// # Returns
+    ///
+    /// The last trigger time
+    pub fn get_last_trigger(&self) -> Instant {
+        self.last_trigger
+            .read()
+            .map(|t| *t)
+            .unwrap_or_else(|_| Instant::now())
+    }
+
+    /// Check if the window is closed
+    ///
+    /// # Returns
+    ///
+    /// * `bool` - true if closed, false otherwise
+    pub fn is_closed(&self) -> bool {
+        self.close_token.is_cancelled()
+    }
+
+    /// Close the window context
+    ///
+    /// Cancels the background timer task
+    pub fn close(&self) {
+        self.close_token.cancel();
+    }
+
+    /// Create a BaseWindow with join support
+    ///
+    /// # Arguments
+    ///
+    /// * `join_config` - Optional join configuration
+    /// * `gap` - Time interval for the timer
+    /// * `resource` - Resource reference
+    ///
+    /// # Returns
+    ///
+    /// A BaseWindow instance or an error
+    pub fn create_base_window(
+        &self,
+        join_config: Option<JoinConfig>,
+        gap: std::time::Duration,
+        resource: &Resource,
+    ) -> Result<BaseWindow, Error> {
+        BaseWindow::new(
+            join_config,
+            Arc::clone(&self.notify),
+            self.close_token.clone(),
+            gap,
+            resource,
+        )
+    }
+}
+
+impl Default for CommonWindowContext {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_common_window_context_creation() {
+        let ctx = CommonWindowContext::new();
+        assert!(!ctx.is_closed());
+        assert!(ctx.get_last_trigger() <= Instant::now());
+    }
+
+    #[test]
+    fn test_common_window_context_close() {
+        let ctx = CommonWindowContext::new();
+        assert!(!ctx.is_closed());
+        ctx.close();
+        assert!(ctx.is_closed());
+    }
+
+    #[test]
+    fn test_common_window_context_update_trigger() {
+        let ctx = CommonWindowContext::new();
+        let now = Instant::now();
+        ctx.update_last_trigger(now);
+        assert!(ctx.get_last_trigger() >= now);
+    }
+
+    #[test]
+    fn test_common_window_context_default() {
+        let ctx = CommonWindowContext::default();
+        assert!(!ctx.is_closed());
+    }
+}

crates/arkflow-plugin/src/buffer/mod.rs

@@ -11,12 +11,14 @@
  *    See the License for the specific language governing permissions and
  *    limitations under the License.
  */
+pub mod common_window;
 mod join;
 pub mod memory;
 pub mod session_window;
 pub mod sliding_window;
 pub mod tumbling_window;
 pub(crate) mod window;
+pub mod window_strategy;
 
 use arkflow_core::Error;
 

crates/arkflow-plugin/src/buffer/session_window.rs

@@ -19,8 +19,10 @@
 //! if they arrive within the gap duration of each other. When the gap duration elapses
 //! without new messages, the session is closed and all accumulated messages are emitted.
 
+use crate::buffer::common_window::CommonWindowContext;
 use crate::buffer::join::JoinConfig;
 use crate::buffer::window::BaseWindow;
+use crate::buffer::window_strategy::{SessionWindowStrategy, WindowStrategy};
 use crate::time::deserialize_duration;
 use arkflow_core::buffer::{register_buffer_builder, Buffer, BufferBuilder};
 use arkflow_core::input::Ack;
@@ -30,9 +32,8 @@ use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use std::sync::Arc;
 use std::time;
-use tokio::sync::{Notify, RwLock};
+use tokio::sync::RwLock;
 use tokio::time::Instant;
-use tokio_util::sync::CancellationToken;
 
 /// Configuration for the session window buffer
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -52,10 +53,8 @@ struct SessionWindow {
     /// Configuration parameters for the session window
     config: SessionWindowConfig,
     base_window: BaseWindow,
-    /// Notification mechanism for signaling between threads
-    notify: Arc<Notify>,
-    /// Token for cancellation of background tasks
-    close: CancellationToken,
+    /// Common window context for timer and notification management
+    context: CommonWindowContext,
     /// Timestamp of the last received message, used to determine session boundaries
     last_message_time: Arc<RwLock<Instant>>,
 }
@@ -69,26 +68,19 @@ impl SessionWindow {
     /// # Returns
     /// * `Result<Self, Error>` - A new session window instance or an error
     fn new(config: SessionWindowConfig, resource: &Resource) -> Result<Self, Error> {
-        let notify = Arc::new(Notify::new());
-        let notify_clone = Arc::clone(&notify);
-        let gap = config.gap;
-        let close = CancellationToken::new();
-        let close_clone = close.clone();
-        let last_message_time = Arc::new(RwLock::new(Instant::now()));
-        let base_window = BaseWindow::new(
-            config.join.clone(),
-            notify_clone,
-            close_clone,
-            gap,
-            resource,
-        )?;
+        let strategy = SessionWindowStrategy::new(config.gap);
+        let context = CommonWindowContext::new();
+
+        // Start the background timer
+        context.start_timer(strategy.check_interval());
+
+        let base_window = context.create_base_window(config.join.clone(), config.gap, resource)?;
 
         Ok(Self {
-            close,
-            notify,
             config,
             base_window,
-            last_message_time,
+            context,
+            last_message_time: Arc::new(RwLock::new(Instant::now())),
         })
     }
 }
@@ -117,7 +109,7 @@ impl Buffer for SessionWindow {
     /// * `Result<Option<(MessageBatchRef, Arc<dyn Ack>)>, Error>` - The merged message batch and combined acknowledgment,
     ///   or None if the buffer is closed and empty
     async fn read(&self) -> Result<Option<(MessageBatchRef, Arc<dyn Ack>)>, Error> {
-        if self.close.is_cancelled() {
+        if self.context.is_closed() {
             return Ok(None);
         }
 
@@ -134,7 +126,7 @@ impl Buffer for SessionWindow {
             }
 
             // Wait for notification from timer or write operation
-            let notify = Arc::clone(&self.notify);
+            let notify = Arc::clone(&self.context.notify);
             notify.notified().await;
         }
         // Process and return the current session
@@ -146,15 +138,21 @@ impl Buffer for SessionWindow {
     /// # Returns
     /// * `Result<(), Error>` - Success or an error
     async fn flush(&self) -> Result<(), Error> {
-        self.base_window.flush().await
+        self.context.close();
+        if !self.base_window.queue_is_empty().await {
+            let notify = Arc::clone(&self.context.notify);
+            notify.notify_waiters();
+        }
+        Ok(())
     }
 
     /// Closes the buffer by cancelling the background task
     ///
     /// # Returns
     /// * `Result<(), Error>` - Success or an error
     async fn close(&self) -> Result<(), Error> {
-        self.base_window.close().await
+        self.context.close();
+        Ok(())
     }
 }