arcon 0.2.1

A runtime for writing streaming applications
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

use crate::{
    error::ArconResult,
    index::{IndexOps, IndexValue, ValueIndex},
    table::ImmutableTable,
};
use arcon_state::{
    backend::{
        handles::{ActiveHandle, Handle},
        Backend, MapState,
    },
    error::*,
};
use std::{borrow::Cow, sync::Arc};

pub struct EagerValue<V, B>
where
    V: IndexValue,
    B: Backend,
{
    /// A handle to the ValueState
    handle: ActiveHandle<B, MapState<u64, V>>,
    current_key: u64,
}

impl<V, B> EagerValue<V, B>
where
    V: IndexValue,
    B: Backend,
{
    /// Creates an EagerValue
    pub fn new(id: impl Into<String>, backend: Arc<B>) -> Self {
        let mut handle = Handle::map(id.into());
        backend.register_map_handle(&mut handle);

        let handle: ActiveHandle<B, MapState<u64, V>> = handle.activate(backend);

        EagerValue {
            handle,
            current_key: 0,
        }
    }
}

impl<V, B> ValueIndex<V> for EagerValue<V, B>
where
    V: IndexValue,
    B: Backend,
{
    fn put(&mut self, value: V) -> Result<()> {
        self.handle.fast_insert(self.current_key, value)
    }
    fn get(&self) -> Result<Option<Cow<V>>> {
        let value = self.handle.get(&self.current_key)?;
        Ok(value.map(Cow::Owned))
    }
    fn take(&mut self) -> Result<Option<V>> {
        self.handle.remove(&self.current_key)
    }
    fn clear(&mut self) -> Result<()> {
        let _ = self.take()?;
        Ok(())
    }
    fn rmw<F>(&mut self, mut f: F) -> Result<()>
    where
        F: FnMut(&mut V) + Sized,
    {
        let value = self.get()?;
        if let Some(v) = value {
            let mut owned = v.into_owned();
            f(&mut owned);
            self.put(owned)
        } else {
            self.put(V::default())
        }
    }
}

impl<V, B> IndexOps for EagerValue<V, B>
where
    V: IndexValue,
    B: Backend,
{
    fn persist(&mut self) -> ArconResult<()> {
        Ok(())
    }
    fn set_key(&mut self, key: u64) {
        self.current_key = key;
    }
    fn table(&mut self) -> ArconResult<Option<ImmutableTable>> {
        let mut table = V::table();
        let values = self.handle.values()?;
        table
            .load(values.filter_map(|v| v.ok()))
            .map_err(|e| ArconStateError::Unknown { msg: e.to_string() })?;
        let imut = table
            .immutable()
            .map_err(|e| ArconStateError::Unknown { msg: e.to_string() })?;
        Ok(Some(imut))
    }
}