Resolves PR comments

Author: SirCipher

example_apps/time_series/src/agent.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/count.rs

@@ -21,21 +21,29 @@ use swimos::{
     server::{Server, ServerBuilder},
 };
 
-use crate::agent::{ExampleAgent, ExampleLifecycle, RetentionPolicy};
+use crate::count::{CountAgent, CountLifecycle};
+use crate::time::{TimeAgent, TimeLifecycle};
 
-mod agent;
+mod count;
+mod time;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     example_logging()?;
-    let route = RoutePattern::parse_str("/example/:name}")?;
 
-    let agent = AgentModel::from_fn(ExampleAgent::default, move || {
-        ExampleLifecycle::new(RetentionPolicy::count(5)).into_lifecycle()
+    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(route, agent)
+        .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);
         })

example_apps/time_series/src/main.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()
+    }
+}