pub(super) async fn add_admin_interfaces_via_handle(
        &mut self,
        configs: Vec<AdminInterfaceConfig>,
        handle: ConductorHandle,
    ) -> ConductorResult<()>
    where
        DS: SafStore + 'static,
    {
        let admin_api = RealAdminInterfaceApi::new(handle);
        let stop_tx = self
            .task_manager
            .as_ref()
            .expect("Task manager not started yet")
            .task_stop_broadcaster()
            .clone();

        // Closure to process each admin config item
        let spawn_from_config = |AdminInterfaceConfig { driver, .. }| {
            let admin_api = admin_api.clone();
            let stop_tx = stop_tx.clone();
            async move {
                match driver {
                    InterfaceDriver::Websocket { port } => {
                        let (listener_handle, listener) = spawn_websocket_listener(port).await?;
                        let port = listener_handle.local_addr().port().unwrap_or(port);
                        let handle: ManagedTaskHandle = spawn_admin_interface_task(
                            listener_handle,
                            listener,
                            admin_api.clone(),
                            stop_tx.subscribe(),
                        )?;
                        InterfaceResult::Ok((port, handle))
                    }
                }
            }
        };

        // spawn interface tasks, collect their JoinHandles,
        // panic on errors.
        let handles: Result<Vec<_>, _> =
            future::join_all(configs.into_iter().map(spawn_from_config))
                .await
                .into_iter()
                .collect();
        // Exit if the admin interfaces fail to be created
        let handles = handles.map_err(Box::new)?;

        {
            let mut ports = Vec::new();

            // First, register the keepalive task, to ensure the conductor doesn't shut down
            // in the absence of other "real" tasks
            self.manage_task(ManagedTaskAdd::ignore(
                tokio::spawn(keep_alive_task(stop_tx.subscribe())),
                "keepalive task",
            ))
            .await?;

            // Now that tasks are spawned, register them with the TaskManager
            for (port, handle) in handles {
                ports.push(port);
                self.manage_task(ManagedTaskAdd::ignore(
                    handle,
                    &format!("admin interface, port {}", port),
                ))
                .await?
            }
            for p in ports {
                self.add_admin_port(p);
            }
        }
        Ok(())
    }
/// Spawn all admin interface tasks, register them with the TaskManager,
/// and modify the conductor accordingly, based on the config passed in
Spawn all admin interface tasks, register them with the TaskManager,
and modify the conductor accordingly, based on the config passed in
Spawn all admin interface tasks register them with the TaskManager and modify the conductor accordingly based on the config passed in
pub(super) async fn add_admin_interfaces_via_handle(
        &mut self,
        configs: Vec<AdminInterfaceConfig>,
        handle: ConductorHandle,
    ) -> ConductorResult<()>
    where
        DS: SafStore + 'static,
    {
        let admin_api = RealAdminInterfaceApi::new(handle);
        let stop_tx = self
            .task_manager
            .as_ref()
            .expect("Task manager not started yet")
            .task_stop_broadcaster()
            .clone();
        // Closure to process each admin config item
        let spawn_from_config = |AdminInterfaceConfig { driver, .. }| {
            let admin_api = admin_api.clone();
            let stop_tx = stop_tx.clone();
            async move {
                match driver {
                    InterfaceDriver::Websocket { port } => {
                        let (listener_handle, listener) = spawn_websocket_listener(port).await?;
                        let port = listener_handle.local_addr().port().unwrap_or(port);
                        let handle: ManagedTaskHandle = spawn_admin_interface_task(
                            listener_handle,
                            listener,
                            admin_api.clone(),
                            stop_tx.subscribe(),
                        )?;
                        InterfaceResult::Ok((port, handle))
                    }
                }
            }
        };
        // spawn interface tasks, collect their JoinHandles,
        // panic on errors.
        let handles: Result<Vec<_>, _> =
            future::join_all(configs.into_iter().map(spawn_from_config))
                .await
                .into_iter()
                .collect();
        // Exit if the admin interfaces fail to be created
        let handles = handles.map_err(Box::new)?;
        {
            let mut ports = Vec::new();
            // First, register the keepalive task, to ensure the conductor doesn't shut down
            // in the absence of other "real" tasks
            self.manage_task(ManagedTaskAdd::ignore(
                tokio::spawn(keep_alive_task(stop_tx.subscribe())),
                "keepalive task",
            ))
            .await?;
            // Now that tasks are spawned, register them with the TaskManager
            for (port, handle) in handles {
                ports.push(port);
                self.manage_task(ManagedTaskAdd::ignore(
                    handle,
                    &format!("admin interface, port {}", port),
                ))
                .await?
            }
            for p in ports {
                self.add_admin_port(p);
            }
        }
        Ok(())
    }
pub ( super ) async fn add_admin_interfaces_via_handle ( & mut self , configs : Vec < AdminInterfaceConfig > , handle : ConductorHandle , ) -> ConductorResult < ( ) > where DS : SafStore + ' static , { let admin_api = RealAdminInterfaceApi :: new ( handle ) ; let stop_tx = self . task_manager . as_ref ( ) . expect ( " " ) . task_stop_broadcaster ( ) . clone ( ) ; // Closure to process each admin config item let spawn_from_config = | AdminInterfaceConfig { driver , .. } | { let admin_api = admin_api . clone ( ) ; let stop_tx = stop_tx . clone ( ) ; async move { match driver { InterfaceDriver :: Websocket { port } => { let ( listener_handle , listener ) = spawn_websocket_listener ( port ) . await ? ; let port = listener_handle . local_addr ( ) . port ( ) . unwrap_or ( port ) ; let handle : ManagedTaskHandle = spawn_admin_interface_task ( listener_handle , listener , admin_api . clone ( ) , stop_tx . subscribe ( ) , ) ? ; InterfaceResult :: Ok ( ( port , handle ) ) } } } } ; // spawn interface tasks, collect their JoinHandles, // panic on errors. let handles : Result < Vec < _ > , _ > = future :: join_all ( configs . into_iter ( ) . map ( spawn_from_config ) ) . await . into_iter ( ) . collect ( ) ; // Exit if the admin interfaces fail to be created let handles = handles . map_err ( Box :: new ) ? ; { let mut ports = Vec :: new ( ) ; // First, register the keepalive task, to ensure the conductor doesn't shut down // in the absence of other "real" tasks self . manage_task ( ManagedTaskAdd :: ignore ( tokio :: spawn ( keep_alive_task ( stop_tx . subscribe ( ) ) ) , " " , ) ) . await ? ; // Now that tasks are spawned, register them with the TaskManager for ( port , handle ) in handles { ports . push ( port ) ; self . manage_task ( ManagedTaskAdd :: ignore ( handle , & format ! ( " " port ) , ) ) . await ? } for p in ports { self . add_admin_port ( p ) ; } } Ok ( ( ) ) }
Spawn all admin interface tasks, register them with the TaskManager,
and modify the conductor accordingly, based on the config passed in
Spawn all admin interface tasks register them with the TaskManager and modify the conductor accordingly based on the config passed in
&mut self
configs
Vec<AdminInterfaceConfig>
handle
ConductorHandle
&mut self
configs
Vec<AdminInterfaceConfig>
handle
ConductorHandle