arcon_codegen 0.1.4

Code generation for Arcon 
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

use crate::types::to_token_stream;
use proc_macro2::{Ident, Span, TokenStream};
use spec::{SinkKind, SocketKind, Type};
use crate::common::verify_and_start;

pub fn sink(name: &str, input_type: &Type, sink_type: &SinkKind, spec_id: &String) -> TokenStream {
    let sink_name = Ident::new(&name, Span::call_site());
    let input_type = to_token_stream(input_type, spec_id);

    let sink_stream = match sink_type {
        SinkKind::Debug => debug_sink(&sink_name, &input_type),
        SinkKind::LocalFile { path } => local_file_sink(&sink_name, &input_type, &path),
        SinkKind::Socket { addr, kind } => socket_sink(&sink_name, &input_type, addr, kind),
    };

    sink_stream
}

fn debug_sink(sink_name: &Ident, input_type: &TokenStream) -> TokenStream {
    let verify = verify_and_start(sink_name, "system");
    quote! {
        let (#sink_name, reg)= system.create_and_register(move || {
            let sink: DebugSink<#input_type> = DebugSink::new();
            sink
        });
        #verify
    }
}

fn local_file_sink(sink_name: &Ident, input_type: &TokenStream, file_path: &str) -> TokenStream {
    let verify = verify_and_start(sink_name, "system");
    quote! {
        let (#sink_name, reg) = system.create_and_register(move || {
            let sink: LocalFileSink<#input_type> = LocalFileSink::new(#file_path);
            sink
        });
        #verify
    }
}

fn socket_sink(
    sink_name: &Ident,
    input_type: &TokenStream,
    addr: &str,
    kind: &SocketKind,
) -> TokenStream {
    let verify = verify_and_start(sink_name, "system");
    let sock_sink = {
        match kind {
            SocketKind::Tcp => unimplemented!(),
            SocketKind::Udp => quote! {  SocketSink::udp(sock_addr); },
        }
    };

    quote! {
        let (#sink_name, reg) = system.create_and_register(move || {
            let sock_addr = #addr.parse().expect("Failed to parse SocketAddr");
            let sink: SocketSink<#input_type> = #sock_sink
            sink
        });
        #verify
    }
}