greentic-runner-host 0.5.15

Host runtime shim for Greentic runner: config, pack loading, activity handling
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107

use super::super::error::GResult;
use super::super::host::{SessionHost, SessionKey, SessionSnapshot};
use async_trait::async_trait;
use parking_lot::RwLock;
use std::collections::HashMap;
use std::time::{Duration, Instant};

#[derive(Clone)]
pub struct SessionEntry {
    pub snapshot: SessionSnapshot,
    pub cas: u64,
    pub expires_at: Option<Instant>,
}

impl SessionEntry {
    fn new(snapshot: SessionSnapshot) -> Self {
        let ttl = snapshot.ttl;
        Self {
            snapshot,
            cas: 0,
            expires_at: Some(Instant::now() + ttl),
        }
    }
}

#[derive(Default)]
pub struct InMemorySessionHost {
    store: RwLock<HashMap<SessionKey, SessionEntry>>,
}

impl InMemorySessionHost {
    pub fn new() -> Self {
        Self {
            store: RwLock::new(HashMap::new()),
        }
    }

    fn apply_ttl(entry: &mut SessionEntry) {
        let ttl = entry.snapshot.ttl;
        entry.expires_at = Some(Instant::now() + ttl);
    }

    fn is_expired(entry: &SessionEntry) -> bool {
        entry
            .expires_at
            .map(|exp| Instant::now() > exp)
            .unwrap_or(false)
    }
}

#[async_trait]
impl SessionHost for InMemorySessionHost {
    async fn get(&self, key: &SessionKey) -> GResult<Option<SessionSnapshot>> {
        let mut guard = self.store.write();
        if let Some(entry) = guard.get_mut(key) {
            if Self::is_expired(entry) {
                guard.remove(key);
                return Ok(None);
            }
            let mut snapshot = entry.snapshot.clone();
            snapshot.revision = entry.cas;
            return Ok(Some(snapshot));
        }
        Ok(None)
    }

    async fn put(&self, mut snapshot: SessionSnapshot) -> GResult<()> {
        snapshot.revision = 0;
        let mut entry = SessionEntry::new(snapshot);
        Self::apply_ttl(&mut entry);
        self.store.write().insert(entry.snapshot.key.clone(), entry);
        Ok(())
    }

    async fn update_cas(
        &self,
        mut snapshot: SessionSnapshot,
        expected_revision: u64,
    ) -> GResult<bool> {
        let mut guard = self.store.write();
        if let Some(entry) = guard.get_mut(&snapshot.key) {
            if entry.cas != expected_revision {
                return Ok(false);
            }
            entry.cas = expected_revision.saturating_add(1);
            snapshot.revision = entry.cas;
            entry.snapshot = snapshot;
            Self::apply_ttl(entry);
            return Ok(true);
        }
        Ok(false)
    }

    async fn delete(&self, key: &SessionKey) -> GResult<()> {
        self.store.write().remove(key);
        Ok(())
    }

    async fn touch(&self, key: &SessionKey, ttl: Duration) -> GResult<()> {
        let mut guard = self.store.write();
        if let Some(entry) = guard.get_mut(key) {
            entry.snapshot.ttl = ttl;
            entry.expires_at = Some(Instant::now() + ttl);
        }
        Ok(())
    }
}