disruptor 4.4.0

//! Module for structs for building a Single Producer Disruptor in a type safe way.
//!
//! To get started building a Single Producer Disruptor, invoke [super::build_single_producer].

use std::{marker::PhantomData, sync::Arc};

use crate::{Sequence, barrier::Barrier, builder::ProcessorSettings, consumer::{MultiConsumerBarrier, SingleConsumerBarrier, event_poller::{EventPoller, Branch}}, producer::single::{SingleProducer, SingleProducerBarrier}, wait_strategies::WaitStrategy};

use super::{Builder, Shared, MC, NC, SC};

/// First step in building a Disruptor with a [SingleProducer].
pub struct SPBuilder<State, E, W, B> {
	state:             PhantomData<State>,
	shared:            Shared<E, W>,
	producer_barrier:  Arc<SingleProducerBarrier>,
	dependent_barrier: Arc<B>,
}

impl<E, W, B, S> ProcessorSettings<E, W> for SPBuilder<S, E, W, B> {
	fn shared(&mut self) -> &mut Shared<E, W> {
		&mut self.shared
	}
}

impl<E, W, B, S> SPBuilder<S, E, W, B>
where
	E: 'static + Send + Sync,
	B: 'static + Barrier,
	W: 'static + WaitStrategy,
{
	/// Branch from the current flow. The [`EventPoller`] for the branch can only be optained when
	/// joining this branch back into the main flow by calling [`join`](Self::join) and passing
	/// the `JoinPromise` returned by this method.
	///
	/// This is intended for building DAG topologies where a branch runs in parallel with multiple
	/// stages, and is only joined back at a later point.
	#[must_use]
	pub fn new_branch(&mut self) -> Branch<E, B> {
		self.build_branch()
	}
}

impl<E, W, B, S> Builder<E, W, B> for SPBuilder<S, E, W, B>
where
	E: 'static + Send + Sync,
	W: 'static + WaitStrategy,
	B: 'static + Barrier,
{
	fn dependent_barrier(&self) -> Arc<B> {
		Arc::clone(&self.dependent_barrier)
	}
}

impl<E, W, B> SPBuilder<NC, E, W, B>
where
	E: 'static + Send + Sync,
	W: 'static + WaitStrategy,
	B: 'static + Barrier,
{
	pub(super) fn new<F>(size: usize, event_factory: F, wait_strategy: W, producer_barrier: Arc<SingleProducerBarrier>, dependent_barrier: Arc<B>) -> Self
	where
		F: FnMut() -> E
	{
		let shared = Shared::new(size, event_factory, wait_strategy);
		Self {
			state: PhantomData,
			shared,
			producer_barrier,
			dependent_barrier,
		}
	}

	/// Transition the builder directly to the multi consumer state (and skip the single consumer optimization).
	///
	/// This is useful if you want to add multiple consumers in e.g. a loop or you only know the number of
	/// consumers at runtime.
	///
	/// # Examples
	///
	/// ```
	///# #[cfg(miri)] fn main() {}
	///# #[cfg(not(miri))]
	///# fn main() {
	///# use disruptor::*;
	///#
	///# struct Event {
	///#     price: f64
	///# }
	///# let factory = || { Event { price: 0.0 }};
	/// let mut builder = build_single_producer(8, factory, BusySpin)
	///     .with_multi_consumer();
	///
	/// let consumer_count = 8; // Could be read from a config file.
	///
	/// // Create a dynamic number of `EventPoller`'s:
	/// let mut pollers = Vec::new();
	/// for _ in 0..consumer_count {
	///     let (poller, next_builder) = builder.new_event_poller();
	///     pollers.push(poller);
	///     builder = next_builder;
	/// }
	///# }
	/// ```
	pub fn with_multi_consumer(self) -> SPBuilder<MC, E, W, B> {
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		}
	}

	/// Get an EventPoller.
	pub fn new_event_poller(mut self) -> (EventPoller<E, B>, SPBuilder<SC, E, W, B>) {
		let event_poller = self.build_event_poller();

		(event_poller,
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		})
	}

	/// Join a branch back into the main flow and get the `EventPoller` for the branch.
	pub fn join<B2>(mut self, branch: Branch<E, B2>) -> (EventPoller<E, B2>, SPBuilder<SC, E, W, B>) {
		self.add_cursor_from_branch(&branch);

		(
			branch.into_poller(),
			SPBuilder {
				state:             PhantomData,
				shared:            self.shared,
				producer_barrier:  self.producer_barrier,
				dependent_barrier: self.dependent_barrier,
			}
		)
	}

	/// Add an event handler.
	pub fn handle_events_with<EH>(mut self, event_handler: EH) -> SPBuilder<SC, E, W, B>
	where
		EH: 'static + Send + FnMut(&E, Sequence, bool)
	{
		self.add_event_handler(event_handler);
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		}
	}

	/// Add an event handler with a processor-specific wait strategy.
	///
	/// Unlike [`handle_events_with`](Self::handle_events_with), this does not use the builder's default strategy.
	pub fn handle_events_with_wait_strategy<EH, PW>(mut self, event_handler: EH, wait_strategy: PW) -> SPBuilder<SC, E, W, B>
	where
		EH: 'static + Send + FnMut(&E, Sequence, bool),
		PW: 'static + WaitStrategy,
	{
		self.add_event_handler_with_wait_strategy(event_handler, wait_strategy);
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		}
	}

	/// Add an event handler with state.
	pub fn handle_events_and_state_with<EH, S, IS>(mut self, event_handler: EH, initialize_state: IS) -> SPBuilder<SC, E, W, B>
	where
		EH: 'static + Send + FnMut(&mut S, &E, Sequence, bool),
		IS: 'static + Send + FnOnce() -> S,
	{
		self.add_event_handler_with_state(event_handler, initialize_state);
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		}
	}

	/// Add an event handler with state and a processor-specific wait strategy.
	///
	/// Unlike [`handle_events_and_state_with`](Self::handle_events_and_state_with), this does not use the builder's default strategy.
	pub fn handle_events_and_state_with_wait_strategy<EH, S, IS, PW>(mut self, event_handler: EH, initialize_state: IS, wait_strategy: PW) -> SPBuilder<SC, E, W, B>
	where
		EH: 'static + Send + FnMut(&mut S, &E, Sequence, bool),
		IS: 'static + Send + FnOnce() -> S,
		PW: 'static + WaitStrategy,
	{
		self.add_event_handler_and_state_with_wait_strategy(event_handler, initialize_state, wait_strategy);
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		}
	}
}

impl <E, W, B> SPBuilder<SC, E, W, B>
where
	E: 'static + Send + Sync,
	W: 'static + WaitStrategy,
	B: 'static + Barrier,
{
	/// Finish the build and get a [`SingleProducer`].
	pub fn build(mut self) -> SingleProducer<E, SingleConsumerBarrier> {
		let mut consumer_cursors = self.shared().current_consumer_cursors.take().unwrap();
		// Guaranteed to be present by construction.
		let consumer_barrier     = SingleConsumerBarrier::new(consumer_cursors.remove(0));
		SingleProducer::new(
			self.shared.shutdown_at_sequence,
			self.shared.ring_buffer,
			self.producer_barrier,
			self.shared.consumers,
			consumer_barrier)
	}

	/// Get an EventPoller.
	pub fn new_event_poller(mut self) -> (EventPoller<E, B>, SPBuilder<MC, E, W, B>) {
		let event_poller = self.build_event_poller();

		(event_poller,
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		})
	}

	/// Join a branch back into the main flow and get the `EventPoller` for the branch.
	pub fn join<B2>(mut self, branch: Branch<E, B2>) -> (EventPoller<E, B2>, SPBuilder<MC, E, W, B>) {
		self.add_cursor_from_branch(&branch);
		
		(
			branch.into_poller(),
			SPBuilder {
				state:             PhantomData,
				shared:            self.shared,
				producer_barrier:  self.producer_barrier,
				dependent_barrier: self.dependent_barrier,
			}
		)
	}

	/// Complete the (concurrent) consumption of events so far and let new consumers process
	/// events after all previous consumers have read them.
	pub fn and_then(mut self) -> SPBuilder<NC, E, W, SingleConsumerBarrier> {
		// Guaranteed to be present by construction.
		let consumer_cursors  = self.shared().current_consumer_cursors.as_mut().unwrap();
		let dependent_barrier = Arc::new(SingleConsumerBarrier::new(consumer_cursors.remove(0)));

		SPBuilder {
			state:            PhantomData,
			shared:           self.shared,
			producer_barrier: self.producer_barrier,
			dependent_barrier,
		}
	}

	/// Add an event handler.
	pub fn handle_events_with<EH>(mut self, event_handler: EH) -> SPBuilder<MC, E, W, B>
	where
		EH: 'static + Send + FnMut(&E, Sequence, bool)
	{
		self.add_event_handler(event_handler);
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		}
	}

	/// Add an event handler with a processor-specific wait strategy.
	///
	/// Unlike [`handle_events_with`](Self::handle_events_with), this does not use the builder's default strategy.
	pub fn handle_events_with_wait_strategy<EH, PW>(mut self, event_handler: EH, wait_strategy: PW) -> SPBuilder<MC, E, W, B>
	where
		EH: 'static + Send + FnMut(&E, Sequence, bool),
		PW: 'static + WaitStrategy,
	{
		self.add_event_handler_with_wait_strategy(event_handler, wait_strategy);
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		}
	}

	/// Add an event handler with state.
	pub fn handle_events_and_state_with<EH, S, IS>(mut self, event_handler: EH, initalize_state: IS) -> SPBuilder<MC, E, W, B>
	where
		EH: 'static + Send + FnMut(&mut S, &E, Sequence, bool),
		IS: 'static + Send + FnOnce() -> S,
	{
		self.add_event_handler_with_state(event_handler, initalize_state);
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		}
	}

	/// Add an event handler with state and a processor-specific wait strategy.
	///
	/// Unlike [`handle_events_and_state_with`](Self::handle_events_and_state_with), this does not use the builder's default strategy.
	pub fn handle_events_and_state_with_wait_strategy<EH, S, IS, PW>(mut self, event_handler: EH, initialize_state: IS, wait_strategy: PW) -> SPBuilder<MC, E, W, B>
	where
		EH: 'static + Send + FnMut(&mut S, &E, Sequence, bool),
		IS: 'static + Send + FnOnce() -> S,
		PW: 'static + WaitStrategy,
	{
		self.add_event_handler_and_state_with_wait_strategy(event_handler, initialize_state, wait_strategy);
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		}
	}
}

impl <E, W, B> SPBuilder<MC, E, W, B>
where
	E: 'static + Send + Sync,
	W: 'static + WaitStrategy,
	B: 'static + Barrier,
{
	/// Get an EventPoller.
	pub fn new_event_poller(mut self) -> (EventPoller<E, B>, SPBuilder<MC, E, W, B>) {
		let event_poller = self.build_event_poller();

		(event_poller,
		SPBuilder {
			state:             PhantomData,
			shared:            self.shared,
			producer_barrier:  self.producer_barrier,
			dependent_barrier: self.dependent_barrier,
		})
	}

	/// Join a branch back into the main flow and get the `EventPoller` for the branch.
	pub fn join<B2>(mut self, branch: Branch<E, B2>) -> (EventPoller<E, B2>, SPBuilder<MC, E, W, B>) {
		self.add_cursor_from_branch(&branch);

		(
			branch.into_poller(),
			SPBuilder {
				state:             PhantomData,
				shared:            self.shared,
				producer_barrier:  self.producer_barrier,
				dependent_barrier: self.dependent_barrier,
			}
		)
	}

	/// Add an event handler.
	pub fn handle_events_with<EH>(mut self, event_handler: EH) -> SPBuilder<MC, E, W, B>
	where
		EH: 'static + Send + FnMut(&E, Sequence, bool)
	{
		self.add_event_handler(event_handler);
		self
	}

	/// Add an event handler with a processor-specific wait strategy.
	///
	/// Unlike [`handle_events_with`](Self::handle_events_with), this does not use the builder's default strategy.
	pub fn handle_events_with_wait_strategy<EH, PW>(mut self, event_handler: EH, wait_strategy: PW) -> SPBuilder<MC, E, W, B>
	where
		EH: 'static + Send + FnMut(&E, Sequence, bool),
		PW: 'static + WaitStrategy,
	{
		self.add_event_handler_with_wait_strategy(event_handler, wait_strategy);
		self
	}

	/// Add an event handler with state.
	pub fn handle_events_and_state_with<EH, S, IS>(mut self, event_handler: EH, initialize_state: IS) -> SPBuilder<MC, E, W, B>
	where
		EH: 'static + Send + FnMut(&mut S, &E, Sequence, bool),
		IS: 'static + Send + FnOnce() -> S,
	{
		self.add_event_handler_with_state(event_handler, initialize_state);
		self
	}

	/// Add an event handler with state and a processor-specific wait strategy.
	///
	/// Unlike [`handle_events_and_state_with`](Self::handle_events_and_state_with), this does not use the builder's default strategy.
	pub fn handle_events_and_state_with_wait_strategy<EH, S, IS, PW>(mut self, event_handler: EH, initialize_state: IS, wait_strategy: PW) -> SPBuilder<MC, E, W, B>
	where
		EH: 'static + Send + FnMut(&mut S, &E, Sequence, bool),
		IS: 'static + Send + FnOnce() -> S,
		PW: 'static + WaitStrategy,
	{
		self.add_event_handler_and_state_with_wait_strategy(event_handler, initialize_state, wait_strategy);
		self
	}

	/// Complete the (concurrent) consumption of events so far and let new consumers process
	/// events after all previous consumers have read them.
	pub fn and_then(mut self) -> SPBuilder<NC, E, W, MultiConsumerBarrier> {
		let consumer_cursors  = self.shared().current_consumer_cursors.replace(vec![]).unwrap();
		let dependent_barrier = Arc::new(MultiConsumerBarrier::new(consumer_cursors));

		SPBuilder {
			dependent_barrier,
			state:            PhantomData,
			shared:           self.shared,
			producer_barrier: self.producer_barrier,
		}
	}

	/// Finish the build and get a [`SingleProducer`].
	pub fn build(mut self) -> SingleProducer<E, MultiConsumerBarrier> {
		let consumer_cursors = self.shared().current_consumer_cursors.take().unwrap();
		let consumer_barrier = MultiConsumerBarrier::new(consumer_cursors);
		SingleProducer::new(
			self.shared.shutdown_at_sequence,
			self.shared.ring_buffer,
			self.producer_barrier,
			self.shared.consumers,
			consumer_barrier)
	}
}