pravega-client 0.1.0

A Rust client for Pravega. (Pravega.io)
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

//
// Copyright (c) Dell Inc., or its subsidiaries. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//

#![allow(dead_code)]
#![deny(
    clippy::all,
    clippy::cargo,
    clippy::else_if_without_else,
    clippy::empty_line_after_outer_attr,
    clippy::multiple_inherent_impl,
    clippy::mut_mut,
    clippy::path_buf_push_overwrite
)]
#![warn(
    clippy::cargo_common_metadata,
    clippy::mutex_integer,
    clippy::needless_borrow,
    clippy::similar_names
)]
#![allow(clippy::multiple_crate_versions, clippy::needless_doctest_main)]

//! A Rust client for [Pravega].
//!
//! [Pravega] is an open source storage system implementing Streams as first-class
//! primitive for storing/serving continuous and unbounded data. It has a number of exciting features
//! including Exactly Once Semantics, Auto Scaling, Unlimited Retention and etc. More details
//! at the [website].
//!
//! Pravega client in Rust provides a few APIs at high level:
//! * [EventStreamWriter] and [EventStreamReader] provide a way to write and read discrete item.
//! * [ByteStream] API provides a way to write and read raw bytes.
//! * [Transaction] API provides a mechanism for writing many events atomically.
//!
//! [Pravega]: https://www.pravega.io/
//! [website]: http://pravega.io/docs/latest/key-features/#pravega-key-features
//! [EventStreamWriter]: crate::event_stream_writer
//! [EventStreamReader]: crate::event_reader
//! [ByteStream]: byte_stream
//! [Transaction]: transaction
//!
//!
use crate::client_factory::ClientFactory;
use pcg_rand::Pcg32;
use pravega_client_shared::{
    Retention, RetentionType, ScaleType, Scaling, Scope, ScopedStream, Stream, StreamConfiguration,
};
use rand::{Rng, SeedableRng};
use std::cell::RefCell;
use std::sync::atomic::{AtomicI64, Ordering};

pub mod byte_stream;
pub mod client_factory;
pub mod error;
pub mod event_reader;
pub mod event_stream_writer;
#[macro_use]
pub mod metric;
pub mod event_reader_group;
pub mod raw_client;
mod reactor;
pub mod reader_group;
pub mod reader_group_config;
pub mod segment_metadata;
pub mod segment_reader;
pub mod segment_slice;
mod stream;
pub mod table_synchronizer;
pub mod tablemap;
pub mod trace;
pub mod transaction;

thread_local! {
    pub(crate) static RNG: RefCell<Pcg32> = RefCell::new(Pcg32::from_entropy());
}

pub(crate) static REQUEST_ID_GENERATOR: AtomicI64 = AtomicI64::new(0);

/// Function used to generate request ids for all the modules.
pub(crate) fn get_request_id() -> i64 {
    REQUEST_ID_GENERATOR.fetch_add(1, Ordering::SeqCst) + 1
}

/// Function used to generate random u64.
pub(crate) fn get_random_u64() -> u64 {
    RNG.with(|rng| rng.borrow_mut().gen())
}

/// Function used to generate random u128.
pub(crate) fn get_random_u128() -> u128 {
    RNG.with(|rng| rng.borrow_mut().gen())
}

/// Function used to generate random i64.
pub(crate) fn get_random_f64() -> f64 {
    RNG.with(|rng| rng.borrow_mut().gen())
}

#[macro_use]
extern crate derive_new;

// helper method
async fn create_stream(factory: &ClientFactory, scope: &str, stream: &str) {
    factory
        .get_controller_client()
        .create_scope(&Scope {
            name: scope.to_string(),
        })
        .await
        .unwrap();
    factory
        .get_controller_client()
        .create_stream(&StreamConfiguration {
            scoped_stream: ScopedStream {
                scope: Scope {
                    name: scope.to_string(),
                },
                stream: Stream {
                    name: stream.to_string(),
                },
            },
            scaling: Scaling {
                scale_type: ScaleType::FixedNumSegments,
                target_rate: 0,
                scale_factor: 0,
                min_num_segments: 1,
            },
            retention: Retention {
                retention_type: RetentionType::None,
                retention_param: 0,
            },
        })
        .await
        .unwrap();
}