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

use std::sync::{Arc, Mutex};

use serde::Serialize;

use super::core::Tag;
use crate::{
	expose_contract::{ExposeContractInlet, ExposeContractShared},
	object::state_sink::ObjectStateSink,
	TagDescriptor,
};

#[derive(PartialEq, Eq, Hash, Copy, Clone, Debug, Serialize)]
pub(crate) struct TagExposeContractId(pub u64);

/// Represents a wish, a __contract__ to expose any object with specific tag
///
/// TagExposeContract is a [futures::stream::Stream], emitting [ObjectStateSink] objects.
///
/// A new ObjectStateSink will get emitted if/when [Exchange](crate::Exchange) gets to know at least one observer ([ObjectObserveContract](crate::ObjectObserveContract)) of an object tagged with the tag the contract pertains to.
/// TagExposeContract will not emit more than one [ObjectStateSink] per object at a time. It will emit multiple ObjectStateSinks accepting ObjectStates of different objects.
/// New [ObjectStateSink] for a specific object will only get emitted when previous ObjectStateSink of that object gets dropped.
///
/// The stream part of all emitted [ObjectStateSink]s will end when contract gets dropped.
pub struct TagExposeContract {
	id: TagExposeContractId,
	tag_core: Arc<Tag>,
	mutable: Arc<Mutex<ExposeContractShared>>,
}

impl TagExposeContract {
	pub(super) fn new(tag_core: Arc<Tag>, id: TagExposeContractId, capacity: u32) -> TagExposeContract {
		let mutable = Arc::new(Mutex::new(ExposeContractShared::new()));
		tag_core.link_local_tag_expose_contract(id, ExposeContractInlet::new(mutable.clone()), capacity);
		TagExposeContract { id, tag_core, mutable }
	}

	pub fn descriptor(&self) -> &TagDescriptor {
		&self.tag_core.descriptor
	}
}

impl Drop for TagExposeContract {
	fn drop(&mut self) {
		self.tag_core.unlink_local_tag_expose_contract(self.id);
	}
}

impl futures::Stream for TagExposeContract {
	type Item = ObjectStateSink;

	fn poll_next(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Option<Self::Item>> {
		self.mutable.lock().unwrap().poll_next(cx)
	}
}

impl std::fmt::Debug for TagExposeContract {
	fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		formatter.write_str(&format!("TagExposeContract:{}", self.tag_core.descriptor))
	}
}