hakuban 0.7.2

use std::{sync::{atomic::Ordering, Arc, Mutex}, task::Waker};

use super::{core::ObjectCore, state_sink::{ObjectStateSink, ObjectStateSinkSharedMutable}};
use crate::{contract::Contract, exchange::core::ExchangeCore, expose_contract::ExposeContract, object::state_sink::ExposeContractMutable, LocalExchange, ObjectDescriptor};


/// Represents a wish, a __contract__ to expose an object
///
/// ObjectExposeContract is a [Stream](futures_util::stream::Stream), emitting [ObjectStateSink] objects.
///
/// ObjectStateSink will only get emitted if/when [LocalExchange] gets to know at least one observer ([ObjectObserveContract](crate::ObjectObserveContract) or [TagObserveContract](crate::TagObserveContract)) of the object the contract pertains to.
/// ObjectExposeContract will not emit more than one [ObjectStateSink] at a time. New [ObjectStateSink] will get emitted only when previous one gets dropped.
///
/// The [Stream](futures_util::stream::Stream) will end when contract gets [ObjectExposeContract::terminate]d.
///
/// # Cloning
/// ObjectExposeContract can be cloned, but cloning does not create a new contract. Think of it as creating o paper copy of an on-paper contract.
/// It's a new sheet of paper, but not a new deal/agreement. Clones are basically references to a logical agreement.
///
/// Accordingly, terminating any of the clones (or the original), renders all of them terminated.
#[derive(Clone)]
pub struct ObjectExposeContract {
	drop_guard: Arc<ObjectExposeContractDropGuard>,
}

struct ObjectExposeContractDropGuard {
	shared: Arc<ObjectExposeContractShared>,
}


pub(crate) struct ObjectExposeContractShared {
	local_exchange: LocalExchange,
	object_core: Arc<ObjectCore>,
	composition_cost: u32,
	mutable: Mutex<ObjectExposeContractSharedMutable>,
}

struct ObjectExposeContractSharedMutable {
	terminated: bool,
	object_state_sink_ready_for_check_out: bool,
	object_state_sink: Option<Arc<Mutex<ObjectStateSinkSharedMutable>>>,
	assigned: bool,
	waker: Vec<Waker>,
}


//LockCheck:  scope=ObjectExposeContract
impl ObjectExposeContract {
	//LockCheck:  call=ObjectCore::link_expose_contract
	pub(super) fn new(exchange: LocalExchange, object_core: Arc<ObjectCore>, composition_cost: u32) -> ObjectExposeContract {
		let shared = Arc::new(ObjectExposeContractShared {
			object_core,
			local_exchange: exchange,
			composition_cost,
			mutable: Mutex::new(ObjectExposeContractSharedMutable {
				waker: Vec::new(),
				object_state_sink: None,
				object_state_sink_ready_for_check_out: false,
				assigned: false,
				terminated: false,
			}),
		});
		shared.object_core.link_expose_contract(shared.clone());
		ObjectExposeContract { drop_guard: Arc::new(ObjectExposeContractDropGuard { shared }) }
	}

	pub fn descriptor(&self) -> &ObjectDescriptor {
		&self.drop_guard.shared.object_core.descriptor
	}

	//LockCheck:  lock=mutable  call=ObjectExposeContractMutable::check_out_object_state_sink   unlock=mutable
	pub fn ready(&mut self) -> Option<ObjectStateSink> {
		let mut shared_mutable = self.drop_guard.shared.mutable.lock().unwrap();
		shared_mutable.check_out_object_state_sink(&self.drop_guard.shared).flatten()
	}

	pub fn terminate(&mut self) {
		self.drop_guard.shared.terminate();
	}
}

impl Drop for ObjectExposeContractDropGuard {
	fn drop(&mut self) {
		self.shared.contract_dropped();
	}
}

//LockCheck:  scope=ObjectExposeContract
impl ObjectExposeContractShared {
	//LockCheck:  call=terminate
	fn contract_dropped(self: &Arc<Self>) {
		self.terminate();
		assert_eq!(Arc::strong_count(self), 1);
	}

	//LockCheck:  lock=mutable  unlock=mutable
	pub(crate) fn object_state_sink_dropped(self: &Arc<Self>) {
		let mut mutable = self.mutable.lock().unwrap();
		mutable.object_state_sink.take().unwrap();
		if mutable.assigned {
			mutable.object_state_sink_ready_for_check_out = true;
		};
		mutable.wake();
	}

	//LockCheck:  lock=mutable  call=ObjectExposeContractMutable::terminate  unlock=mutable  call=ObjectCore::unlink_expose_contract  call=LocalExchange::object_changed
	fn terminate(self: &Arc<Self>) {
		let mut mutable = self.mutable.lock().unwrap();
		if !mutable.terminated {
			mutable.terminate();
			drop(mutable);
			//TODO: is this retarded?
			let dyn_self: Arc<dyn ExposeContract> = self.clone();
			self.object_core.unlink_expose_contract(&dyn_self);
			self.local_exchange.shared.object_changed(self.object_core.descriptor.clone());
			drop(dyn_self);
		}
	}
}

impl Contract for ObjectExposeContractShared {
	fn id(&self) -> usize {
		(self as *const Self) as usize
	}

	fn exchange_core(&self) -> &Arc<ExchangeCore> {
		self.local_exchange.shared.exchange_core.as_ref().unwrap()
	}
}

//LockCheck:  scope=ObjectExposeContractMutable
impl ObjectExposeContractSharedMutable {
	fn wake(&mut self) {
		for waker in self.waker.drain(..) {
			waker.wake();
		}
	}

	//LockCheck:  call=ObjectStateSink::new
	fn check_out_object_state_sink(&mut self, shared: &Arc<ObjectExposeContractShared>) -> Option<Option<ObjectStateSink>> {
		if self.terminated {
			Some(None)
		} else if self.object_state_sink_ready_for_check_out {
			if self.object_state_sink.is_some() {
				panic!("Double accept")
			};
			self.object_state_sink_ready_for_check_out = false;
			let object_state_sink = ObjectStateSink::new(shared.object_core.clone(), ExposeContractMutable::Object(shared.clone()));
			self.object_state_sink = Some(object_state_sink.drop_guard.shared.clone());
			Some(Some(object_state_sink))
		} else {
			None
		}
	}

	//LockCheck:  lock=ObjectStateSink::mutable  call=ObjectStateSink::contract_terminated  unlock=ObjectStateSink::mutable
	fn terminate(&mut self) {
		if !self.terminated {
			self.terminated = true;
			self.wake();
			if let Some(object_state_sink) = self.object_state_sink.take() {
				object_state_sink.lock().unwrap().contract_terminated();
			};
		}
	}
}

//LockCheck:  scope=ObjectExposeContract
impl ExposeContract for ObjectExposeContractShared {
	fn composition_cost(&self) -> u32 {
		self.composition_cost
	}

	//LockCheck:  lock=mutable  lock=ObjectStateSink::mutable  call=ObjectStateSink::assign  unlock=ObjectStateSink::mutable  unlock=mutable
	fn assign(&self, _object_core: &Arc<ObjectCore>) -> bool {
		let mut mutable = self.mutable.lock().unwrap();

		if mutable.assigned {
			panic!("Double assign");
		};
		mutable.assigned = true;

		if let Some(object_state_sink) = &mutable.object_state_sink {
			object_state_sink.lock().unwrap().assign();
		} else {
			mutable.object_state_sink_ready_for_check_out = true;
		};

		self.exchange_core().load.fetch_add(self.composition_cost(), Ordering::Relaxed);

		mutable.wake();
		true
	}

	//LockCheck:  lock=mutable  lock=ObjectStateSink::mutable  call=ObjectStateSink::unassign  unlock=ObjectStateSink::mutable  unlock=mutable
	fn unassign(&self, _object_core: &Arc<ObjectCore>) -> bool {
		//TODO: panic if object doesn't match?
		let mut mutable = self.mutable.lock().unwrap();

		if !mutable.assigned {
			panic!("Double unassign (#1)");
		};
		mutable.assigned = false;

		if let Some(object_state_sink) = &mutable.object_state_sink {
			object_state_sink.lock().unwrap().unassign();
		} else {
			mutable.object_state_sink_ready_for_check_out = false;
		}

		self.exchange_core().load.fetch_sub(self.composition_cost(), Ordering::Relaxed);

		mutable.wake();
		true
	}

	fn object(&self) -> bool {
		true
	}
}



//LockCheck:  scope=ObjectExposeContract
impl futures::Stream for ObjectExposeContract {
	type Item = ObjectStateSink;

	//LockCheck:  lock=mutable  call=ObjectExposeContractMutable::check_out_object_state_sink  unlock=mutable
	fn poll_next(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Option<Self::Item>> {
		let mut shared_mutable = self.drop_guard.shared.mutable.lock().unwrap();
		match shared_mutable.check_out_object_state_sink(&self.drop_guard.shared) {
			Some(object_state_sink) => std::task::Poll::Ready(object_state_sink),
			None => {
				shared_mutable.waker.push(cx.waker().clone());
				std::task::Poll::Pending
			}
		}
	}
}


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