Struct ProducerCtx 

pub struct ProducerCtx<'a> { /* private fields */ }

Context handed to a producer’s build closure on activation.

Provides:

• Self::node_id / Self::core — identity + Core access for sink callbacks that re-enter Core.
• Self::subscribe_to — subscribe to an upstream Core node; the resulting Subscription is auto-tracked under node_id in the binding’s producer storage and dropped on producer deactivation.

Implementations

impl<'a> ProducerCtx<'a>

pub fn new( node_id: NodeId, core: &'a Core, storage: &'a ProducerStorage, ) -> Self

Construct a new context for the binding’s invoke_fn dispatch to call build closures. Internal — bindings call this; user code receives the constructed ctx via the build closure’s arg.

pub fn node_id(&self) -> NodeId

The producer node’s id.

pub fn core(&self) -> &Core

The Core dispatcher. Sink closures use this to re-enter Core — core.emit(self.node_id(), h) to emit a value, etc.

pub fn subscribe_to(&self, source: NodeId, sink: Sink)

Subscribe sink to upstream source. The Subscription is auto-tracked under the producer’s node_id; on producer deactivation, the binding drops the storage entry, which drops the Subscription, which unsubscribes the sink.

Phase H+ option (d) /qa N1(a) (2026-05-09): the sink is wrapped in ProducerSinkGuard so for the duration of every sink invocation, the per-thread IN_PRODUCER_BUILD refcount is bumped, suppressing the H+ ascending-order check. This preserves the existing producer-pattern operator architecture against the widened H+ gate. The producer-pattern operators (zip, concat, race, take_until, switch_map, exhaust_map, concat_map, merge_map) all do cross-partition Core::subscribe plus Core::emit from inside their inner-source sink callbacks; the wrapping carve-out lets those continue to work while non-producer sink-callback re-entry IS checked by the widened gate. Refactoring the operators to defer their sink-time inner subscribes and re-emits to wave-end is the broader Phase H+ STRICT variant scope (option b defer-to-post-flush, estimated 1500+ LOC) per docs/porting-deferred.md.