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
118
119
120
121
122
123
124
125
126
127
128

use crate::encoding;
use crate::io::provider::{
    PackedAction, PackedEvent, PackedState, ProviderReqId, StreamType, Timestamp,
};
use anyhow::Error;
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use std::cmp::Ordering;
use std::fmt;

// TODO: Move to the separate module
/// Requirements for a data fraction in a data flow.
pub trait DataFraction:
    DeserializeOwned + Serialize + Clone + fmt::Debug + Sync + Send + 'static
{
}

impl<T> DataFraction for T where
    T: DeserializeOwned + Serialize + Clone + fmt::Debug + Sync + Send + 'static
{
}

/// Immutable state of a data flow.
pub trait Flow: DataFraction {
    /// `ControlEvent` - that send from a client to a server
    type Action: DataFraction;

    /// `UpdateEvent` - that sent from a server to a client
    type Event: DataFraction;

    fn stream_type() -> StreamType;

    fn apply(&mut self, event: Self::Event);

    fn pack_state(&self) -> Result<PackedState, Error> {
        encoding::pack(self)
    }

    fn unpack_state(data: &PackedState) -> Result<Self, Error> {
        encoding::unpack(data)
    }

    fn pack_event(delta: &Self::Event) -> Result<PackedEvent, Error> {
        encoding::pack(delta)
    }

    fn unpack_event(data: &PackedEvent) -> Result<Self::Event, Error> {
        encoding::unpack(data)
    }

    fn pack_action(action: &Self::Action) -> Result<PackedAction, Error> {
        encoding::pack(action)
    }

    fn unpack_action(data: &PackedAction) -> Result<Self::Action, Error> {
        encoding::unpack(data)
    }
}

#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct TimedEvent<T> {
    pub timestamp: Timestamp,
    pub event: T,
}

impl<T> TimedEvent<T> {
    pub fn into_inner(self) -> T {
        self.event
    }
}

impl<T> Ord for TimedEvent<T> {
    fn cmp(&self, other: &Self) -> Ordering {
        self.timestamp.cmp(&other.timestamp)
    }
}

impl<T> PartialOrd for TimedEvent<T> {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl<T> PartialEq for TimedEvent<T> {
    fn eq(&self, other: &Self) -> bool {
        self.timestamp == other.timestamp
    }
}

impl<T> Eq for TimedEvent<T> {}

/// Envelope for incoming actions that contains routing information.
#[derive(Debug, Clone)]
pub struct ActionEnvelope<T: Flow> {
    /// Direction to a client.
    pub origin: ProviderReqId,
    /// The reason of sending an envelope.
    pub activity: Activity,
    /// An action sent to a clinet.
    /// It's detached from activity to make it suitable for
    /// third-party languages.
    pub action: Option<T::Action>,
}

/// Variant of activity that send to tracers.
///
/// It doesn't include `Action` value to make this type
/// compatible with languages that have no ADTs.
#[derive(Debug, Clone)]
pub enum Activity {
    /// No one connected client
    Suspend = 0, // 0b0000
    /// At least one client connected
    Awake = 1, // 0b0001

    /// Listener disconnected
    Disconnected = 2, // 0b0010
    /// Listener connected
    Connected = 3, // 0b0011

    /// Forwards an action
    Action = 4, // 0b010
}

impl Activity {
    pub fn is_action(&self) -> bool {
        matches!(self, Self::Action)
    }
}