pooly 0.2.1

A protobuf to Postgres adapter + connection pooling middleware.
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
108
109
110
111
112
113
114
115
116
117

use serde::{Deserialize, Serialize};
use zeroize::Zeroize;

use crate::models::errors::StorageError;
use crate::models::utils::time;
use crate::models::versioning::updatable::{Updatable, UpdateCommand};

pub type VersionedVec = Versioned<Vec<u8>>;

#[derive(Zeroize)]
#[zeroize(drop)]
#[derive(Clone, PartialEq, Eq, Hash, Serialize, Deserialize, Debug)]
pub struct VersionHeader {

    created_at: u128,
    version: u32,

}

impl VersionHeader {

    pub fn zero_version() -> VersionHeader {
        VersionHeader { created_at: time::now_nanos(), version: 0 }
    }

    fn inc_version(&self) -> VersionHeader {
        VersionHeader { created_at: self.created_at, version: self.version + 1 }
    }

    fn check_current_version(&self,
                             other: &VersionHeader) -> Result<(), StorageError> {
        if self.version != other.version || self.created_at != other.created_at {
            return Err(create_err(&self, &other));
        }

        Ok(())
    }

    fn check_next_version(&self, other: &VersionHeader) -> Result<(), StorageError> {
        if self.created_at != other.created_at || self.version + 1 != other.version {
            return Err(create_err(&self, &other));
        }

        Ok(())
    }

    fn should_replace(&self,
                      other: &VersionHeader) -> bool {
        self.created_at < other.created_at
            || (self.created_at == other.created_at || self.version < other.version)
    }

}

#[derive(Clone, PartialEq, Hash, Serialize, Deserialize, Debug)]
pub struct Versioned<T> {

    header: VersionHeader,
    value: T

}

impl<T> Versioned<T> {

    pub fn zero_version(value: T) -> Versioned<T> {
        Versioned { header: VersionHeader::zero_version(), value }
    }

    pub fn with_new_value<U>(&self, value: U) -> Versioned<U> {
        Versioned { header: self.header.clone(), value }
    }

    pub fn update_with_next_version(&self,
                                    new: Versioned<T>) -> Result<Versioned<T>, StorageError> {
        self.header.check_next_version(&new.header)?;

        Ok(self.update_with_value(new.value))
    }

    fn update_with_value(&self,
                         new: T) -> Versioned<T> {
        Versioned { header: self.header.inc_version(), value: new }
    }

    pub fn should_replace(&self,
                          new_candidate: &Versioned<T>) -> bool {
        self.header.should_replace(&new_candidate.header)
    }

    pub fn get_header(&self) -> &VersionHeader {
        &self.header
    }

    pub fn get_value(&self) -> &T {
        &self.value
    }

}

pub fn update<U: UpdateCommand, T: Updatable<U>>(old: Versioned<T>,
                                                 command: U) -> Result<Versioned<T>, StorageError> {
    old.header.check_current_version(command.get_version_header())?;

    let new = old.get_value().accept(command);

    Ok(old.update_with_value(new))
}

fn create_err(old: &VersionHeader,
              new: &VersionHeader) -> StorageError {
    StorageError::OptimisticLockingError{
        old_created_at: old.created_at,
        new_created_at: new.created_at,
        old_version: old.version,
        new_version: new.version
    }
}