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

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

use super::core::Tag;
use crate::{
	object::ObjectStateStream,
	observe_contract::{ObserveContractInlet, ObserveContractShared},
	TagDescriptor,
};

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

/// Represents a wish, a __contract__ to observe all objects with specific tag
///
/// TagObserveContract is a [futures::stream::Stream], emitting [ObjectStateStream] objects.
///
/// A new ObjectStateStream will get emitted if/when [Exchange](crate::Exchange) gets to know at least one [ObjectState](crate::ObjectState) of an object tagged with the tag the contract pertains to.
/// TagObserveContract will not emit more than one [ObjectStateStream] per object at a time. It will emit multiple ObjectStateStreams carrying ObjectStates of different objects.
/// New [ObjectStateStream] for a specific object will only get emitted when previous ObjectStateStream of that object gets dropped.
///
/// All emitted [ObjectStateStream]s will end when contract gets dropped.
pub struct TagObserveContract {
	id: TagObserveContractId,
	tag_core: Arc<Tag>,
	mutable: Arc<Mutex<ObserveContractShared>>,
}

impl TagObserveContract {
	pub(super) fn new(tag_core: Arc<Tag>, id: TagObserveContractId) -> TagObserveContract {
		let shared = Arc::new(Mutex::new(ObserveContractShared::new()));
		tag_core.link_local_tag_observe_contract(id, ObserveContractInlet::new(shared.clone()));
		TagObserveContract { id, tag_core, mutable: shared }
	}

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

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

impl futures::Stream for TagObserveContract {
	type Item = ObjectStateStream;

	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 TagObserveContract {
	fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		formatter.write_str(&format!("TagObserveContract:{}", self.tag_core.descriptor))
	}
}