hakuban 0.8.5

Data-object sharing library
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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203

#![allow(clippy::tabs_in_doc_comments)] //because "cargo fmt" keeps replacing comment spaces with tabs, despite "format_code_in_doc_comments=false"

/*!
Hakuban is a data-object sharing library.
Check out the [README](https://gitlab.com/yunta/hakuban/-/blob/main/README.md) file if you haven't been there yet.

Hakuban is a simple mechanism which allows exposing data-objects in some processes, and observing them in other processes.
Hakuban takes care of all the boring details of data transmission, synchronization, object lifetime management, load balancing etc.

At a high level the API can be divided into 6 parts:
* Network management: [Exchange], [tokio_runtime::WebsocketListener], [tokio_runtime::WebsocketConnector]
* Descriptors ("rich labels" for objects and tags): [ObjectDescriptor], [TagDescriptor]
* Object and tag contracts: [ObjectObserveContract], [ObjectExposeContract], [TagObserveContract], [TagExposeContract]
* Object state access: [ObjectState], [ObjectStateStream], [ObjectStateSink]
* [FFI](ffi), WASM

# Concepts

A hakuban __object__ represents a shared variable.
Atomically writable through a selected ExposeContract. And atomically readable through ObserveContracts.
Among existing ExposeContracts, the hakuban network elects a single[^1] one, which becomes responsible for producing new states of the object.
ObserveContracts can only access the most recent object state (that they've received so far).

Every __object__ has a unique identifier - an [ObjectDescriptor].
Every __object__ belongs to some set of __tags__, which are used for wildcard observing/exposing[^2].
Every __tag__ has a unique identifier - a [TagDescriptor].

ExposeContracts and ObserveContracts can be obtained from an [Exchange].
The Exchange controlls lifetime of objects, selects a single exposer for every object needing to be exposed, handles load-balancing, and facilitates object state propagation.
The exchange can be asked to emit:
* [ObjectObserveContract] - declaration of interest in observing values of a single object
* [ObjectExposeContract] - declaration of ability to expose (update, produce) new values of a single object
* [TagObserveContract] - declaration of interest in observing values of all[^3] objects tagged with a specific __tag__
* [TagExposeContract] - declaration of ability to expose (update, produce) new values of all[^3] objects tagged with specific __tag__

Observe contracts are streams emitting streams of [ObjectState] structs. Expose contracts are streams emiting sinks of [ObjectState] structs.

Exchanges in separate processes can be connected to each other in a tree topology.
Processes (exchanges) at the leaf-end of the tree will only receive data relevant to them.

[^1]: It's **usually** a single ExposeContract, but the number it's not guaranteed. More than one ExposeContract may be elected at a time, typically during load-balancing hand-over.
[^2]: Tags are not meant to represent data model level collections. They are merely wildcards for declaring observe/expose contracts for objects of yet-unknown [ObjectDescriptor]s. `TODO: explain with an example`
[^3]: All "live" objects. Those of which exists at least one ObjectObserveContract, or, both: ObjectExposeContract and a TagObserveContract.

# Example

```rust
# use hakuban::{Exchange, ObjectState, JsonSerializeState, JsonDeserializeState};
# use futures::{stream::StreamExt, sink::SinkExt};
# tokio_test::block_on(async {
let exchange = Exchange::new();

let mut observe_contract = exchange.object_observe_contract((["some-tag"],"xxx")).build();
let mut expose_contract = exchange.object_expose_contract((["some-tag"],"xxx")).build();

tokio::spawn(async move {
	//ExposeContract emits ObjectStateSink when first observer appears
	let mut object_state_sink = expose_contract.next().await.unwrap();
	let object_state = ObjectState::new("my data").json_serialize();
	object_state_sink.send(object_state).await;
});

tokio::spawn(async move {
	//ObserveContract emits ObjectStateStream
	let mut object_state_stream = observe_contract.next().await.unwrap();
	let object_state = object_state_stream.next().await.unwrap().json_deserialize::<String>();
	println!("{:?}", object_state.data);
});
# });
```

# Connecting somewhere

Start the `hakuban-router` (binary built by this lib), and following 2 processes. Order doesn't matter, they will happily wait for each other ❤️.
The A and B processes will communicate through the `hakuban-router` instance.

You can also make them communicate directly, by making one of them create a [tokio_runtime::WebsocketListener] instead of the [tokio_runtime::WebsocketConnector].

You can not have more than one WebsocketConnector attached to a single Exchange, to keep the network tree-shaped.

Process A:
```rust
# use hakuban::{Exchange, tokio_runtime::WebsocketConnector};
# use futures::{stream::StreamExt, sink::SinkExt};
# tokio_test::block_on(async {
let exchange = Exchange::new();
let _upstream = WebsocketConnector::new(exchange.clone(), "ws://127.0.0.1:3001")?;
let mut observe_contract = exchange.object_observe_contract((["some-tag"],"xxx")).build();

let object_state = observe_contract.next().await.unwrap().next().await.unwrap();
println!("{:?}", object_state.data);
# Ok::<(),Box<dyn std::error::Error>>(())
# });
```

Process B:
```rust
# use hakuban::{Exchange, tokio_runtime::WebsocketConnector, ObjectState, JsonSerializeState};
# use futures::{stream::StreamExt, sink::SinkExt};
# tokio_test::block_on(async {
let exchange = Exchange::new();
let _upstream = WebsocketConnector::new(exchange.clone(), "ws://127.0.0.1:3001")?;
let mut expose_contract = exchange.object_expose_contract((["some-tag"],"xxx")).build();

let object_state = ObjectState::new("🦀").json_serialize();
let mut object_state_sink = expose_contract.next().await.unwrap();
object_state_sink.send(object_state).await;

//waiting for Process A to receive the goods and exit
while object_state_sink.next().await.is_some() {};
# Ok::<(),Box<dyn std::error::Error>>(())
# });
```


# Cargo feature flags
* `default`: `["tokio-runtime", "ffi", "downstream"]`
* `tokio-runtime`: enables compilation of tokio_runtime.rs, the only runtime&network transport implemented so far. You'll need it if you want to communicate with other processes.
* `musl`: allows building statically-linked binaries:
	```cargo build --target x86_64-unknown-linux-musl --release --features musl```
* `downstream`: enables processing of incoming connections. there is no corresponding "upstream" feature, outgoing connections are always enabled.
* `wasm`: used for building the browser-wasm

# FFI / .so

Check out documentation of the [ffi] module.

# Included binaries

There is one binary built by this crate - the `hakuban-router`.
It's basically a WebsocketListener surrounded by "main".
It accepts connections and will naturally route hakuban communication, like every other hakuban process with WebsocketListener would.

*/

/*

* bugs to fix
	* statically detect deadlocks and/or prove there aren't any

* features
	* everything listed in the README/roadmap
	* gap-less expose handover
	* apply diff/patch to descriptors
	* error!() in ffi functions, before better error handling gets implemented

* clean-up
	* check if error returns in ffi don't cut through allocations
	* de arc-ify selfs where possible
	* merge *contract.rs into a single file
	* make diff caching smarter
	* rewrite all inline type constraints to "where"

* consider
	* can expose contract and observe contract be structs instead of traits? enums instead of dyns?
	* maybe get rid of million hashbrown monomorphizations
	* batch-lock ???
	* split "links" into multiple locks
	* drop multiple-tags-per-object support in favour of exactly-one-tag-per-object (groups, possibly tree-structured)
	* json_(de)serialize on stream
	* checksum data
	* https://github.com/rust-lang/api-guidelines/blob/master/src/SUMMARY.md
	  * C-DEBUG: implement debug on all public types
	  * C-NEWTYPE: use struct DataVersion(Vec<i64>);  instead of type =, etc.
	* limit assignments by derivative instead of linear slope

* bindings

*/

mod connection;
mod descriptor;
mod exchange;
mod expose_contract;
#[cfg(feature = "ffi")]
pub mod ffi;
mod format;
#[cfg(not(target_family = "wasm"))]
pub mod monitor;
mod object;
mod observe_contract;
mod tag;
#[cfg(feature = "tokio-runtime")]
pub mod tokio_runtime;
mod utils;
#[cfg(target_family = "wasm")]
pub mod wasm;

pub use descriptor::*;
pub use exchange::Exchange;
pub use format::{JsonDeserializeError, JsonDeserializeState, JsonSerializeState};
pub use object::{
	DataBytes, DataFormat, DataSynchronized, DataVersion, ObjectExposeContract, ObjectExposeContractBuilder, ObjectObserveContract,
	ObjectObserveContractBuilder, ObjectState, ObjectStateSink, ObjectStateSinkParams, ObjectStateStream,
};
pub use tag::{TagExposeContract, TagExposeContractBuilder, TagObserveContract, TagObserveContractBuilder};
#[cfg(feature = "jemalloc")]
use tikv_jemallocator::Jemalloc;
pub use utils::HakubanStreamExt;

#[cfg(feature = "jemalloc")]
#[global_allocator]
static GLOBAL: Jemalloc = Jemalloc;