gantryclient 0.1.0

Gantry (waSCC actor registry) client
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
118
119
120
121
122

use gantry_protocol as protocol;
use nats::Connection;
use protocol::catalog::*;
use protocol::stream::*;
use protocol::PingResponse;
use protocol::{deserialize, serialize};
use std::time::Duration;

pub const CHUNK_SIZE: u64 = 256 * 1024; // 256KB
pub const DOWNLOAD_TIMEOUT_SECONDS: u64 = 3;
pub const MIN_WASM_SIZE: usize = 2 * 1024; // A wasm module < 2KB is either corrupt or non-existent

pub(crate) fn query(
    client: &Connection,
    query: &CatalogQuery,
    timeout: Duration,
) -> Result<CatalogQueryResults, Box<dyn ::std::error::Error>> {
    let buf = serialize(&query)?;
    let reply = client.request_timeout("gantry.catalog.tokens.query", &buf, timeout)?;

    Ok(deserialize::<CatalogQueryResults>(reply.data.as_ref())?)
}

pub(crate) fn get_detail(
    client: &Connection,
    key: &str,
    timeout: Duration,
) -> Result<TokenDetail, Box<dyn ::std::error::Error>> {
    let reply = client.request_timeout("gantry.catalog.tokens.get", key.as_bytes(), timeout)?;

    Ok(deserialize::<TokenDetail>(reply.data.as_ref())?)
}

pub(crate) fn put(
    client: &Connection,
    token: &Token,
    timeout: Duration,
) -> Result<PutTokenResponse, Box<dyn ::std::error::Error>> {
    let buf = serialize(token)?;
    let reply = client.request_timeout("gantry.catalog.tokens.put", &buf, timeout)?;

    let res = deserialize::<PutTokenResponse>(reply.data.as_ref())?;
    Ok(res)
}

pub(crate) fn ping(
    client: &Connection,
    timeout: Duration,
) -> Result<PingResponse, Box<dyn ::std::error::Error>> {
    let res = client.request_timeout("gantry.ping", &[], timeout)?;
    let pong = deserialize::<PingResponse>(res.data.as_ref())?;
    Ok(pong)
}

pub(crate) fn start_upload(
    client: &Connection,
    req: &UploadRequest,
    timeout: Duration,
) -> Result<TransferAck, Box<dyn ::std::error::Error>> {
    let buf = serialize(req)?;

    let res = client.request_timeout(protocol::stream::SUBJECT_STREAM_UPLOAD, &buf, timeout)?;
    let tack = deserialize::<TransferAck>(res.data.as_ref())?;
    Ok(tack)
}

pub(crate) fn request_download(
    client: &Connection,
    req: DownloadRequest,
) -> Result<Vec<u8>, Box<dyn ::std::error::Error>> {
    let buf = serialize(&req)?;

    let mut res = Vec::new();
    for (idx, msg) in client
        .request_multi(SUBJECT_STREAM_DOWNLOAD, &buf)?
        .timeout_iter(Duration::from_secs(DOWNLOAD_TIMEOUT_SECONDS))
        .enumerate()
    {
        if idx > 0 {
            // skip the ack packet
            let chunk: FileChunk = deserialize(&msg.data)?;
            res.extend_from_slice(&chunk.chunk_bytes);
            if res.len() == chunk.total_bytes as usize {
                break;
            }
        }
    }

    if res.len() > MIN_WASM_SIZE {
        Ok(res)
    } else {
        Err("Did not receive an actor from Gantry".into())
    }
}

pub(crate) fn upload_chunk(
    c: &Connection,
    sequence_no: u64,
    actor: &str,
    chunk_size: u64,
    total_bytes: u64,
    total_chunks: u64,
    bytes: Vec<u8>,
    timeout: Duration,
) -> Result<(), Box<dyn ::std::error::Error>> {
    let chunk = protocol::stream::FileChunk {
        actor: actor.to_string(),
        chunk_bytes: bytes,
        chunk_size,
        sequence_no,
        total_bytes,
        total_chunks,
    };
    let buf = serialize(&chunk)?;
    let subject = format!(
        "{}{}",
        protocol::stream::SUBJECT_STREAM_UPLOAD_PREFIX,
        actor
    );
    let _res = c.request_timeout(&subject, &buf, timeout)?;
    Ok(())
}