pub async fn spawn_queue_consumer_tasks(
    env: &EnvironmentWrite,
    cell_network: HolochainP2pCell,
    conductor_api: impl CellConductorApiT + 'static,
    mut task_sender: sync::mpsc::Sender<ManagedTaskAdd>,
    stop: sync::broadcast::Sender<()>,
) -> (QueueTriggers, InitialQueueTriggers) {
    // Publish
    let (tx_publish, handle) =
        spawn_publish_dht_ops_consumer(env.clone(), stop.subscribe(), cell_network.clone());
    task_sender
        .send(ManagedTaskAdd::dont_handle(handle))
        .await
        .expect("Failed to manage workflow handle");

    let (create_tx_sys, get_tx_sys) = tokio::sync::oneshot::channel();

    // Integration
    let (tx_integration, handle) =
        spawn_integrate_dht_ops_consumer(env.clone(), stop.subscribe(), get_tx_sys);
    task_sender
        .send(ManagedTaskAdd::dont_handle(handle))
        .await
        .expect("Failed to manage workflow handle");

    // App validation
    let (tx_app, handle) = spawn_app_validation_consumer(
        env.clone(),
        stop.subscribe(),
        tx_integration.clone(),
        conductor_api.clone(),
        cell_network.clone(),
    );
    task_sender
        .send(ManagedTaskAdd::dont_handle(handle))
        .await
        .expect("Failed to manage workflow handle");

    // Sys validation
    let (tx_sys, handle) = spawn_sys_validation_consumer(
        env.clone(),
        stop.subscribe(),
        tx_app.clone(),
        cell_network,
        conductor_api,
    );
    task_sender
        .send(ManagedTaskAdd::dont_handle(handle))
        .await
        .expect("Failed to manage workflow handle");
    if create_tx_sys.send(tx_sys.clone()).is_err() {
        panic!("Failed to send tx_sys");
    }

    // Produce
    let (tx_produce, handle) =
        spawn_produce_dht_ops_consumer(env.clone(), stop.subscribe(), tx_publish.clone());
    task_sender
        .send(ManagedTaskAdd::dont_handle(handle))
        .await
        .expect("Failed to manage workflow handle");

    (
        QueueTriggers::new(tx_sys.clone(), tx_produce.clone()),
        InitialQueueTriggers::new(tx_sys, tx_produce, tx_publish, tx_app, tx_integration),
    )
}
/// Spawns several long-running tasks which are responsible for processing work
/// which shows up on various databases.
///
/// Waits for the initial loop to complete before returning, to prevent causing
/// a race condition by trying to run a workflow too soon after cell creation.
Spawns several long-running tasks which are responsible for processing work
which shows up on various databases.

Waits for the initial loop to complete before returning, to prevent causing
a race condition by trying to run a workflow too soon after cell creation.
Spawns several long - running tasks which are responsible for processing work which shows up on various databases . Waits for the initial loop to complete before returning to prevent causing a race condition by trying to run a workflow too soon after cell creation .
pub async fn spawn_queue_consumer_tasks(
    env: &EnvironmentWrite,
    cell_network: HolochainP2pCell,
    conductor_api: impl CellConductorApiT + 'static,
    mut task_sender: sync::mpsc::Sender<ManagedTaskAdd>,
    stop: sync::broadcast::Sender<()>,
) -> (QueueTriggers, InitialQueueTriggers) {
    // Publish
    let (tx_publish, handle) =
        spawn_publish_dht_ops_consumer(env.clone(), stop.subscribe(), cell_network.clone());
    task_sender
        .send(ManagedTaskAdd::dont_handle(handle))
        .await
        .expect("Failed to manage workflow handle");
    let (create_tx_sys, get_tx_sys) = tokio::sync::oneshot::channel();
    // Integration
    let (tx_integration, handle) =
        spawn_integrate_dht_ops_consumer(env.clone(), stop.subscribe(), get_tx_sys);
    task_sender
        .send(ManagedTaskAdd::dont_handle(handle))
        .await
        .expect("Failed to manage workflow handle");
    // App validation
    let (tx_app, handle) = spawn_app_validation_consumer(
        env.clone(),
        stop.subscribe(),
        tx_integration.clone(),
        conductor_api.clone(),
        cell_network.clone(),
    );
    task_sender
        .send(ManagedTaskAdd::dont_handle(handle))
        .await
        .expect("Failed to manage workflow handle");
    // Sys validation
    let (tx_sys, handle) = spawn_sys_validation_consumer(
        env.clone(),
        stop.subscribe(),
        tx_app.clone(),
        cell_network,
        conductor_api,
    );
    task_sender
        .send(ManagedTaskAdd::dont_handle(handle))
        .await
        .expect("Failed to manage workflow handle");
    if create_tx_sys.send(tx_sys.clone()).is_err() {
        panic!("Failed to send tx_sys");
    }
    // Produce
    let (tx_produce, handle) =
        spawn_produce_dht_ops_consumer(env.clone(), stop.subscribe(), tx_publish.clone());
    task_sender
        .send(ManagedTaskAdd::dont_handle(handle))
        .await
        .expect("Failed to manage workflow handle");
    (
        QueueTriggers::new(tx_sys.clone(), tx_produce.clone()),
        InitialQueueTriggers::new(tx_sys, tx_produce, tx_publish, tx_app, tx_integration),
    )
}
pub async fn spawn_queue_consumer_tasks ( env : & EnvironmentWrite , cell_network : HolochainP2pCell , conductor_api : impl CellConductorApiT + ' static , mut task_sender : sync :: mpsc :: Sender < ManagedTaskAdd > , stop : sync :: broadcast :: Sender < ( ) > , ) -> ( QueueTriggers , InitialQueueTriggers ) { // Publish let ( tx_publish , handle ) = spawn_publish_dht_ops_consumer ( env . clone ( ) , stop . subscribe ( ) , cell_network . clone ( ) ) ; task_sender . send ( ManagedTaskAdd :: dont_handle ( handle ) ) . await . expect ( " " ) ; let ( create_tx_sys , get_tx_sys ) = tokio :: sync :: oneshot :: channel ( ) ; // Integration let ( tx_integration , handle ) = spawn_integrate_dht_ops_consumer ( env . clone ( ) , stop . subscribe ( ) , get_tx_sys ) ; task_sender . send ( ManagedTaskAdd :: dont_handle ( handle ) ) . await . expect ( " " ) ; // App validation let ( tx_app , handle ) = spawn_app_validation_consumer ( env . clone ( ) , stop . subscribe ( ) , tx_integration . clone ( ) , conductor_api . clone ( ) , cell_network . clone ( ) , ) ; task_sender . send ( ManagedTaskAdd :: dont_handle ( handle ) ) . await . expect ( " " ) ; // Sys validation let ( tx_sys , handle ) = spawn_sys_validation_consumer ( env . clone ( ) , stop . subscribe ( ) , tx_app . clone ( ) , cell_network , conductor_api , ) ; task_sender . send ( ManagedTaskAdd :: dont_handle ( handle ) ) . await . expect ( " " ) ; if create_tx_sys . send ( tx_sys . clone ( ) ) . is_err ( ) { panic ! ( " " ) ; } // Produce let ( tx_produce , handle ) = spawn_produce_dht_ops_consumer ( env . clone ( ) , stop . subscribe ( ) , tx_publish . clone ( ) ) ; task_sender . send ( ManagedTaskAdd :: dont_handle ( handle ) ) . await . expect ( " " ) ; ( QueueTriggers :: new ( tx_sys . clone ( ) , tx_produce . clone ( ) ) , InitialQueueTriggers :: new ( tx_sys , tx_produce , tx_publish , tx_app , tx_integration ) , ) }
Spawns several long-running tasks which are responsible for processing work
which shows up on various databases.
Spawns several long - running tasks which are responsible for processing work which shows up on various databases .
env
&EnvironmentWrite
cell_network
HolochainP2pCell
conductor_api
impl CellConductorApiT + 'static
task_sender
sync::mpsc::Sender<ManagedTaskAdd>
stop
sync::broadcast::Sender<()>
env
&EnvironmentWrite
cell_network
HolochainP2pCell
conductor_api
impl CellConductorApiT + 'static
task_sender
sync::mpsc::Sender<ManagedTaskAdd>
stop
sync::broadcast::Sender<()>