Merge pull request #660 from swimos/time_series

Data windowing example application

Author: SirCipher

Cargo.toml

@@ -39,6 +39,7 @@ members = [
     "example_apps/join_map",
     "example_apps/join_value",
     "example_apps/aggregations",
+    "example_apps/time_series",
 ]
 
 exclude = [

example_apps/time_series/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "time_series"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+swimos = { path = "../../swimos", features = ["server", "agent"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+example-util = { path = "../example_util" }
+futures-util = { workspace = true }
+swimos_form = { path = "../../api/swimos_form" }

example_apps/time_series/src/count.rs

@@ -0,0 +1,95 @@
+// Copyright 2015-2023 Swim Inc.
+//
+// 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.
+
+use std::cell::RefCell;
+use std::collections::{HashMap, VecDeque};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use swimos::{
+    agent::event_handler::Sequentially,
+    agent::lanes::CommandLane,
+    agent::lanes::MapLane,
+    agent::{
+        agent_lifecycle::HandlerContext,
+        event_handler::{EventHandler, HandlerActionExt},
+        lifecycle, projections, AgentLaneModel,
+    },
+};
+
+#[derive(AgentLaneModel)]
+#[projections]
+pub struct CountAgent {
+    history: MapLane<u64, String>,
+    add: CommandLane<String>,
+}
+
+pub struct CountLifecycle {
+    max: usize,
+    keys: RefCell<VecDeque<u64>>,
+}
+
+impl CountLifecycle {
+    pub fn new(max: usize) -> CountLifecycle {
+        CountLifecycle {
+            max,
+            keys: Default::default(),
+        }
+    }
+}
+
+#[lifecycle(CountAgent, no_clone)]
+impl CountLifecycle {
+    #[on_command(add)]
+    pub fn add(
+        &self,
+        context: HandlerContext<CountAgent>,
+        cmd: &str,
+    ) -> impl EventHandler<CountAgent> {
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+        context.update(CountAgent::HISTORY, now, cmd.to_string())
+    }
+
+    #[on_update(history)]
+    pub fn on_update(
+        &self,
+        context: HandlerContext<CountAgent>,
+        _map: &HashMap<u64, String>,
+        key: u64,
+        _prev: Option<String>,
+        _new_value: &str,
+    ) -> impl EventHandler<CountAgent> {
+        let CountLifecycle { max, keys } = self;
+        let timestamps = &mut *keys.borrow_mut();
+        timestamps.push_front(key);
+
+        let len = timestamps.len();
+        let to_drop = if len > *max { len - *max } else { 0 };
+
+        let handler = if to_drop > 0 {
+            let keys = timestamps
+                .split_off(to_drop)
+                .into_iter()
+                .take(to_drop)
+                .map(move |key| context.remove(CountAgent::HISTORY, key));
+            Some(Sequentially::new(keys))
+        } else {
+            None
+        };
+
+        handler.discard()
+    }
+}

example_apps/time_series/src/main.rs

@@ -0,0 +1,62 @@
+// Copyright 2015-2023 Swim Inc.
+//
+// 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.
+
+use std::{error::Error, time::Duration};
+
+use example_util::{example_logging, manage_handle};
+use swimos::{
+    agent::agent_model::AgentModel,
+    route::RoutePattern,
+    server::{Server, ServerBuilder},
+};
+
+use crate::count::{CountAgent, CountLifecycle};
+use crate::time::{TimeAgent, TimeLifecycle};
+
+mod count;
+mod time;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    example_logging()?;
+
+    let count_route = RoutePattern::parse_str("/count/:name}")?;
+    let count_agent = AgentModel::from_fn(CountAgent::default, move || {
+        CountLifecycle::new(5).into_lifecycle()
+    });
+
+    let time_route = RoutePattern::parse_str("/time/:name}")?;
+    let time_agent = AgentModel::from_fn(TimeAgent::default, move || {
+        TimeLifecycle::new(13).into_lifecycle()
+    });
+
+    let server = ServerBuilder::with_plane_name("Example Plane")
+        .add_route(count_route, count_agent)
+        .add_route(time_route, time_agent)
+        .update_config(|config| {
+            config.agent_runtime.inactive_timeout = Duration::from_secs(5 * 60);
+        })
+        .build()
+        .await?;
+
+    let (task, handle) = server.run();
+
+    let shutdown = manage_handle(handle);
+
+    let (_, result) = tokio::join!(shutdown, task);
+
+    result?;
+    println!("Server stopped successfully.");
+    Ok(())
+}

example_apps/time_series/src/time.rs

@@ -0,0 +1,101 @@
+// Copyright 2015-2023 Swim Inc.
+//
+// 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.
+
+use std::cell::RefCell;
+use std::collections::{HashMap, VecDeque};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use swimos::{
+    agent::event_handler::Sequentially,
+    agent::lanes::CommandLane,
+    agent::lanes::MapLane,
+    agent::{
+        agent_lifecycle::HandlerContext,
+        event_handler::{EventHandler, HandlerActionExt},
+        lifecycle, projections, AgentLaneModel,
+    },
+};
+
+#[derive(AgentLaneModel)]
+#[projections]
+pub struct TimeAgent {
+    history: MapLane<u64, String>,
+    add: CommandLane<String>,
+}
+
+pub struct TimeLifecycle {
+    interval: u64,
+    keys: RefCell<VecDeque<u64>>,
+}
+
+impl TimeLifecycle {
+    pub fn new(interval: u64) -> TimeLifecycle {
+        TimeLifecycle {
+            interval,
+            keys: Default::default(),
+        }
+    }
+}
+
+#[lifecycle(TimeAgent, no_clone)]
+impl TimeLifecycle {
+    #[on_command(add)]
+    pub fn add(
+        &self,
+        context: HandlerContext<TimeAgent>,
+        cmd: &str,
+    ) -> impl EventHandler<TimeAgent> {
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+        context.update(TimeAgent::HISTORY, now, cmd.to_string())
+    }
+
+    #[on_update(history)]
+    pub fn on_update(
+        &self,
+        context: HandlerContext<TimeAgent>,
+        _map: &HashMap<u64, String>,
+        key: u64,
+        _prev: Option<String>,
+        _new_value: &str,
+    ) -> impl EventHandler<TimeAgent> {
+        let TimeLifecycle { interval, keys } = self;
+        let timestamps = &mut *keys.borrow_mut();
+        timestamps.push_back(key);
+        let mut to_remove = Vec::new();
+
+        let start = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_millis() as u64;
+
+        timestamps.retain(|timestamp| {
+            if start - *timestamp > *interval {
+                to_remove.push(context.remove(TimeAgent::HISTORY, *timestamp));
+                false
+            } else {
+                true
+            }
+        });
+        let handler = if to_remove.is_empty() {
+            None
+        } else {
+            Some(Sequentially::new(to_remove))
+        };
+
+        handler.discard()
+    }
+}