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
101
102
103

use crate::{
    data::{
        flight_serde::{reliable_remote::ReliableSerde, FlightSerde},
        ArconEvent, ArconMessage, ArconType,
    },
    stream::channel::Channel,
};
use kompact::prelude::{ComponentDefinition, SerError};
use std::sync::Arc;

#[allow(dead_code)]
pub mod broadcast;
pub mod forward;
pub mod keyed;

/// A `ChannelStrategy` defines a strategy of how messages are sent downstream
///
/// Common strategies include (one-to-one)[forward::Forward] and (one-to-many)[broadcast::Broadcast]
pub enum ChannelStrategy<A>
where
    A: ArconType,
{
    /// Send messages to a single Component
    Forward(forward::Forward<A>),
    /// Broadcasts the message to a Vec of `Channels`
    #[allow(dead_code)]
    Broadcast(broadcast::Broadcast<A>),
    /// Partition data to a set of `Channels` based on keyed hash
    Keyed(keyed::Keyed<A>),
    /// A strategy that prints to the console
    Console,
    /// A strategy that simply does nothing
    Mute,
}

impl<A> ChannelStrategy<A>
where
    A: ArconType,
{
    /// Add event to outgoing buffer
    ///
    /// The function returns a tuple of channel and msg if the buffers are full or a marker was inputted.
    #[inline]
    pub fn push(
        &mut self,
        event: ArconEvent<A>,
    ) -> impl IntoIterator<Item = (Arc<Channel<A>>, ArconMessage<A>)> {
        match self {
            ChannelStrategy::Forward(s) => s.add(event),
            ChannelStrategy::Keyed(s) => s.add(event),
            ChannelStrategy::Broadcast(s) => s.add(event),
            ChannelStrategy::Console => {
                println!("{:?}", event);
                Vec::new()
            }
            ChannelStrategy::Mute => Vec::new(),
        }
    }

    /// Returns number of outgoing channels
    #[inline]
    #[allow(dead_code)]
    pub fn num_channels(&self) -> usize {
        match self {
            ChannelStrategy::Forward(_) => 1,
            ChannelStrategy::Broadcast(s) => s.num_channels(),
            ChannelStrategy::Keyed(s) => s.num_channels(),
            ChannelStrategy::Console => 0,
            ChannelStrategy::Mute => 0,
        }
    }
}

/// `send` pushes an ArconMessage onto a Component queue
///
/// The message may be sent to a local or remote component
#[inline]
pub(crate) fn send<A: ArconType>(
    channel: &Channel<A>,
    message: ArconMessage<A>,
    source: &impl ComponentDefinition,
) -> Result<(), SerError> {
    match channel {
        Channel::Local(actor_ref) => {
            actor_ref.tell(message);
            Ok(())
        }
        Channel::Remote(actor_path, FlightSerde::Reliable) => {
            let reliable_msg = ReliableSerde(message.into());
            actor_path.tell_serialised(reliable_msg, source)
        }
    }
}

#[cfg(test)]
pub mod tests {
    #[derive(Arcon, prost::Message, Clone)]
    #[arcon(reliable_ser_id = 13, version = 1)]
    pub struct Input {
        #[prost(uint32, tag = "1")]
        pub id: u32,
    }
}