Struct Process 

pub struct Process<'a, ProcessTag = ()> { /* private fields */ }

A single-node location in a distributed Hydro program.

Process represents exactly one machine (or OS process) and is one of the core location types that implements the Location trait. Live collections placed on a Process are materialized entirely on that single node.

The type parameter ProcessTag is a compile-time marker that differentiates distinct processes in the same dataflow graph (e.g. Process<'a, Leader> vs Process<'a, Follower>). It defaults to () when only one process is needed.

Creating a Process

let mut flow = FlowBuilder::new();
let node = flow.process::<MyTag>();

Implementations

impl<'a, P> Process<'a, P>

pub fn sim_atomic_input<T, O: Ordering, R: Retries>( &self, ) -> (SimAtomicSender<T, O, R>, Stream<T, Atomic<Self>, Unbounded, O, R>)

where T: Serialize + DeserializeOwned,

Available on crate feature sim only.

Sets up a simulated atomic input port on this process for testing.

The source is placed directly inside the atomic tick, so sent values are immediately available in the next atomic slice without requiring a separate tick to batch them in. send_atomic is synchronous.

Trait Implementations

impl<P> Clone for Process<'_, P>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<P> Debug for Process<'_, P>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<'a, P> Location<'a> for Process<'a, P>

type Root = Process<'a, P>

The root location type for this location.

fn root(&self) -> Self::Root

Returns the root location for this location.

fn try_tick(&self) -> Option<Tick<Self>>

Attempts to create a new Tick clock domain at this location.

fn id(&self) -> LocationId

Returns the unique identifier for this location.

fn tick(&self) -> Tick<Self>

where Self: NoTick,

Creates a new Tick clock domain at this location.

fn spin(&self) -> Stream<(), Self, Unbounded, TotalOrder, ExactlyOnce>

where Self: Sized + NoTick,

Creates an unbounded stream that continuously emits unit values ().

fn source_stream<T, E>( &self, e: impl QuotedWithContext<'a, E, Self>, ) -> Stream<T, Self, Unbounded, TotalOrder, ExactlyOnce>

where E: FuturesStream<Item = T> + Unpin, Self: Sized + NoTick,

Creates a stream from an async [FuturesStream].

fn source_iter<T, E>( &self, e: impl QuotedWithContext<'a, E, Self>, ) -> Stream<T, Self, Bounded, TotalOrder, ExactlyOnce>

where E: IntoIterator<Item = T>, Self: Sized,

Creates a bounded stream from an iterator.

fn source_cluster_members<C: 'a>( &self, cluster: &Cluster<'a, C>, ) -> KeyedStream<MemberId<C>, MembershipEvent, Self, Unbounded>

where Self: Sized + NoTick,

Creates a stream of membership events for a cluster.

fn source_external_bytes<L>( &self, from: &External<'_, L>, ) -> (ExternalBytesPort, Stream<BytesMut, Self, Unbounded, TotalOrder, ExactlyOnce>)

where Self: Sized + NoTick,

Creates a one-way connection from an external process to receive raw bytes.

fn source_external_bincode<L, T, O: Ordering, R: Retries>( &self, from: &External<'_, L>, ) -> (ExternalBincodeSink<T, NotMany, O, R>, Stream<T, Self, Unbounded, O, R>)

where Self: Sized + NoTick, T: Serialize + DeserializeOwned,

Creates a one-way connection from an external process to receive bincode-serialized data.

fn sim_input<T, O: Ordering, R: Retries>( &self, ) -> (SimSender<T, O, R>, Stream<T, Self, Unbounded, O, R>)

where Self: Sized + NoTick, T: Serialize + DeserializeOwned,

Available on crate feature sim only.

Sets up a simulated input port on this location for testing.

fn embedded_input<T>( &self, name: impl Into<String>, ) -> Stream<T, Self, Unbounded, TotalOrder, ExactlyOnce>

where Self: Sized + NoTick,

Creates an external input stream for embedded deployment mode.

fn embedded_singleton_input<T>( &self, name: impl Into<String>, ) -> Singleton<T, Self, Bounded>

where Self: Sized + NoTick,

Creates an embedded singleton input for embedded deployment mode.

fn bind_single_client<L, T, Codec: Encoder<T> + Decoder>( &self, from: &External<'_, L>, port_hint: NetworkHint, ) -> (ExternalBytesPort<NotMany>, Stream<<Codec as Decoder>::Item, Self, Unbounded, TotalOrder, ExactlyOnce>, ForwardHandle<'a, Stream<T, Self, Unbounded, TotalOrder, ExactlyOnce>>)

where Self: Sized + NoTick,

Establishes a server on this location to receive a bidirectional connection from a single client, identified by the given External handle. Returns a port handle for the external process to connect to, a stream of incoming messages, and a handle to send outgoing messages.

fn bind_single_client_bincode<L, InT: DeserializeOwned, OutT: Serialize>( &self, from: &External<'_, L>, ) -> (ExternalBincodeBidi<InT, OutT, NotMany>, Stream<InT, Self, Unbounded, TotalOrder, ExactlyOnce>, ForwardHandle<'a, Stream<OutT, Self, Unbounded, TotalOrder, ExactlyOnce>>)

where Self: Sized + NoTick,

Establishes a bidirectional connection from a single external client using bincode serialization.

fn bidi_external_many_bytes<L, T, Codec: Encoder<T> + Decoder>( &self, from: &External<'_, L>, port_hint: NetworkHint, ) -> (ExternalBytesPort<Many>, KeyedStream<u64, <Codec as Decoder>::Item, Self, Unbounded, TotalOrder, ExactlyOnce>, KeyedStream<u64, MembershipEvent, Self, Unbounded, TotalOrder, ExactlyOnce>, ForwardHandle<'a, KeyedStream<u64, T, Self, Unbounded, NoOrder, ExactlyOnce>>)

where Self: Sized + NoTick,

Establishes a server on this location to receive bidirectional connections from multiple external clients using raw bytes.

fn bidi_external_many_bincode<L, InT: DeserializeOwned, OutT: Serialize>( &self, from: &External<'_, L>, ) -> (ExternalBincodeBidi<InT, OutT, Many>, KeyedStream<u64, InT, Self, Unbounded, TotalOrder, ExactlyOnce>, KeyedStream<u64, MembershipEvent, Self, Unbounded, TotalOrder, ExactlyOnce>, ForwardHandle<'a, KeyedStream<u64, OutT, Self, Unbounded, NoOrder, ExactlyOnce>>)

where Self: Sized + NoTick,

Establishes a server on this location to receive bidirectional connections from multiple external clients using bincode serialization.

fn singleton<T>( &self, e: impl QuotedWithContext<'a, T, Self>, ) -> Singleton<T, Self, Bounded>

where Self: Sized + NoTick,

Constructs a Singleton materialized at this location with the given static value.

fn singleton_future<F>( &self, e: impl QuotedWithContext<'a, F, Self>, ) -> Singleton<F::Output, Self, Bounded>

where F: Future, Self: Sized + NoTick,

Constructs a Singleton by resolving an async Future to completion.

fn source_interval( &self, interval: impl QuotedWithContext<'a, Duration, Self> + Copy + 'a, _nondet: NonDet, ) -> Stream<Instant, Self, Unbounded, TotalOrder, ExactlyOnce>

where Self: Sized + NoTick,

Generates a stream with values emitted at a fixed interval, with each value being the current time (as an [tokio::time::Instant]).

fn source_interval_delayed( &self, delay: impl QuotedWithContext<'a, Duration, Self> + Copy + 'a, interval: impl QuotedWithContext<'a, Duration, Self> + Copy + 'a, _nondet: NonDet, ) -> Stream<Instant, Self, Unbounded, TotalOrder, ExactlyOnce>

where Self: Sized + NoTick,

Generates a stream with values emitted at a fixed interval (with an initial delay), with each value being the current time (as an [tokio::time::Instant]).

fn forward_ref<S>(&self) -> (ForwardHandle<'a, S>, S)

where S: CycleCollection<'a, ForwardRef, Location = Self>,

Creates a forward reference, allowing a stream to be used before its source is defined.

impl<P> PartialEq for Process<'_, P>

fn eq(&self, other: &Self) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl<P> Eq for Process<'_, P>

impl<T> NoAtomic for Process<'_, T>

impl<T> NoTick for Process<'_, T>