pub(crate) async fn next_pending(&self) -> Option<DispatchOrTimeoutLA> {
        let new_or_retry = match self.rcvs.lock().await.next(&self.semaphore).await? {
            NewOrCancel::Cancel(c) => {
                return match c {
                    CancelOrTimeout::Cancel(c) => Some(DispatchOrTimeoutLA::Dispatch(c)),
                    CancelOrTimeout::Timeout {
                        run_id,
                        resolution,
                        dispatch_cancel,
                    } => {
                        let task = if dispatch_cancel {
                            let tt = self
                                .dat
                                .lock()
                                .id_to_tt
                                .get(&ExecutingLAId {
                                    run_id: run_id.clone(),
                                    seq_num: resolution.seq,
                                })
                                .map(Clone::clone);
                            if let Some(task_token) = tt {
                                self.complete(&task_token, &resolution.result);
                                Some(ActivityTask {
                                    task_token: task_token.0,
                                    variant: Some(activity_task::Variant::Cancel(Cancel {
                                        reason: ActivityCancelReason::TimedOut as i32,
                                    })),
                                })
                            } else {
                                None
                            }
                        } else {
                            None
                        };
                        Some(DispatchOrTimeoutLA::Timeout {
                            run_id,
                            resolution,
                            task,
                        })
                    }
                };
            }
            NewOrCancel::New(n) => n,
        };

        // It is important that there are no await points after receiving from the channel, as
        // it would mean dropping this future would cause us to drop the activity request.
        let (new_la, attempt) = match new_or_retry {
            NewOrRetry::New(n) => {
                let explicit_attempt_num_or_1 = n.schedule_cmd.attempt.max(1);
                (n, explicit_attempt_num_or_1)
            }
            NewOrRetry::Retry { in_flight, attempt } => (in_flight, attempt),
        };
        let orig = new_la.clone();
        let id = ExecutingLAId {
            run_id: new_la.workflow_exec_info.run_id.clone(),
            seq_num: new_la.schedule_cmd.seq,
        };
        let sa = new_la.schedule_cmd;

        let mut dat = self.dat.lock();
        // If this request originated from a local backoff task, clear the entry for it. We
        // don't await the handle because we know it must already be done, and there's no
        // meaningful value.
        dat.backing_off_tasks.remove(&id);

        // If this task sat in the queue for too long, return a timeout for it instead
        if let Some(s2s) = sa.schedule_to_start_timeout.as_ref() {
            let sat_for = new_la.schedule_time.elapsed().unwrap_or_default();
            if sat_for > *s2s {
                return Some(DispatchOrTimeoutLA::Timeout {
                    run_id: new_la.workflow_exec_info.run_id,
                    resolution: LocalActivityResolution {
                        seq: sa.seq,
                        result: LocalActivityExecutionResult::timeout(TimeoutType::ScheduleToStart),
                        runtime: sat_for,
                        attempt,
                        backoff: None,
                        original_schedule_time: Some(new_la.schedule_time),
                    },
                    task: None,
                });
            }
        }

        let tt = dat
            .id_to_tt
            .get(&id)
            .expect("Task token must exist")
            .clone();
        dat.outstanding_activity_tasks.insert(
            tt.clone(),
            LocalInFlightActInfo {
                la_info: orig,
                dispatch_time: Instant::now(),
                attempt,
            },
        );
        if let Some(to) = dat.timeout_tasks.get_mut(&id) {
            to.mark_started();
        }

        let (schedule_to_close, start_to_close) = sa.close_timeouts.into_sched_and_start();
        Some(DispatchOrTimeoutLA::Dispatch(ActivityTask {
            task_token: tt.0,
            variant: Some(activity_task::Variant::Start(Start {
                workflow_namespace: self.namespace.clone(),
                workflow_type: new_la.workflow_type,
                workflow_execution: Some(new_la.workflow_exec_info),
                activity_id: sa.activity_id,
                activity_type: sa.activity_type,
                header_fields: sa.headers,
                input: sa.arguments,
                heartbeat_details: vec![],
                scheduled_time: Some(new_la.schedule_time.into()),
                current_attempt_scheduled_time: Some(new_la.schedule_time.into()),
                started_time: Some(SystemTime::now().into()),
                attempt,
                schedule_to_close_timeout: schedule_to_close.map(Into::into),
                start_to_close_timeout: start_to_close.map(Into::into),
                heartbeat_timeout: None,
                retry_policy: Some(sa.retry_policy),
                is_local: true,
            })),
        }))
    }
/// Returns the next pending local-activity related action, or None if shutdown has initiated
/// and there are no more remaining actions to take.
Returns the next pending local-activity related action, or None if shutdown has initiated
and there are no more remaining actions to take.
Returns the next pending local - activity related action or None if shutdown has initiated and there are no more remaining actions to take .
pub(crate) async fn next_pending(&self) -> Option<DispatchOrTimeoutLA> {
        let new_or_retry = match self.rcvs.lock().await.next(&self.semaphore).await? {
            NewOrCancel::Cancel(c) => {
                return match c {
                    CancelOrTimeout::Cancel(c) => Some(DispatchOrTimeoutLA::Dispatch(c)),
                    CancelOrTimeout::Timeout {
                        run_id,
                        resolution,
                        dispatch_cancel,
                    } => {
                        let task = if dispatch_cancel {
                            let tt = self
                                .dat
                                .lock()
                                .id_to_tt
                                .get(&ExecutingLAId {
                                    run_id: run_id.clone(),
                                    seq_num: resolution.seq,
                                })
                                .map(Clone::clone);
                            if let Some(task_token) = tt {
                                self.complete(&task_token, &resolution.result);
                                Some(ActivityTask {
                                    task_token: task_token.0,
                                    variant: Some(activity_task::Variant::Cancel(Cancel {
                                        reason: ActivityCancelReason::TimedOut as i32,
                                    })),
                                })
                            } else {
                                None
                            }
                        } else {
                            None
                        };
                        Some(DispatchOrTimeoutLA::Timeout {
                            run_id,
                            resolution,
                            task,
                        })
                    }
                };
            }
            NewOrCancel::New(n) => n,
        };
        // It is important that there are no await points after receiving from the channel, as
        // it would mean dropping this future would cause us to drop the activity request.
        let (new_la, attempt) = match new_or_retry {
            NewOrRetry::New(n) => {
                let explicit_attempt_num_or_1 = n.schedule_cmd.attempt.max(1);
                (n, explicit_attempt_num_or_1)
            }
            NewOrRetry::Retry { in_flight, attempt } => (in_flight, attempt),
        };
        let orig = new_la.clone();
        let id = ExecutingLAId {
            run_id: new_la.workflow_exec_info.run_id.clone(),
            seq_num: new_la.schedule_cmd.seq,
        };
        let sa = new_la.schedule_cmd;
        let mut dat = self.dat.lock();
        // If this request originated from a local backoff task, clear the entry for it. We
        // don't await the handle because we know it must already be done, and there's no
        // meaningful value.
        dat.backing_off_tasks.remove(&id);
        // If this task sat in the queue for too long, return a timeout for it instead
        if let Some(s2s) = sa.schedule_to_start_timeout.as_ref() {
            let sat_for = new_la.schedule_time.elapsed().unwrap_or_default();
            if sat_for > *s2s {
                return Some(DispatchOrTimeoutLA::Timeout {
                    run_id: new_la.workflow_exec_info.run_id,
                    resolution: LocalActivityResolution {
                        seq: sa.seq,
                        result: LocalActivityExecutionResult::timeout(TimeoutType::ScheduleToStart),
                        runtime: sat_for,
                        attempt,
                        backoff: None,
                        original_schedule_time: Some(new_la.schedule_time),
                    },
                    task: None,
                });
            }
        }
        let tt = dat
            .id_to_tt
            .get(&id)
            .expect("Task token must exist")
            .clone();
        dat.outstanding_activity_tasks.insert(
            tt.clone(),
            LocalInFlightActInfo {
                la_info: orig,
                dispatch_time: Instant::now(),
                attempt,
            },
        );
        if let Some(to) = dat.timeout_tasks.get_mut(&id) {
            to.mark_started();
        }
        let (schedule_to_close, start_to_close) = sa.close_timeouts.into_sched_and_start();
        Some(DispatchOrTimeoutLA::Dispatch(ActivityTask {
            task_token: tt.0,
            variant: Some(activity_task::Variant::Start(Start {
                workflow_namespace: self.namespace.clone(),
                workflow_type: new_la.workflow_type,
                workflow_execution: Some(new_la.workflow_exec_info),
                activity_id: sa.activity_id,
                activity_type: sa.activity_type,
                header_fields: sa.headers,
                input: sa.arguments,
                heartbeat_details: vec![],
                scheduled_time: Some(new_la.schedule_time.into()),
                current_attempt_scheduled_time: Some(new_la.schedule_time.into()),
                started_time: Some(SystemTime::now().into()),
                attempt,
                schedule_to_close_timeout: schedule_to_close.map(Into::into),
                start_to_close_timeout: start_to_close.map(Into::into),
                heartbeat_timeout: None,
                retry_policy: Some(sa.retry_policy),
                is_local: true,
            })),
        }))
    }
pub ( crate ) async fn next_pending ( & self ) -> Option < DispatchOrTimeoutLA > { let new_or_retry = match self . rcvs . lock ( ) . await . next ( & self . semaphore ) . await ? { NewOrCancel :: Cancel ( c ) => { return match c { CancelOrTimeout :: Cancel ( c ) => Some ( DispatchOrTimeoutLA :: Dispatch ( c ) ) , CancelOrTimeout :: Timeout { run_id , resolution , dispatch_cancel , } => { let task = if dispatch_cancel { let tt = self . dat . lock ( ) . id_to_tt . get ( & ExecutingLAId { run_id : run_id . clone ( ) , seq_num : resolution . seq , } ) . map ( Clone :: clone ) ; if let Some ( task_token ) = tt { self . complete ( & task_token , & resolution . result ) ; Some ( ActivityTask { task_token : task_token . 0 , variant : Some ( activity_task :: Variant :: Cancel ( Cancel { reason : ActivityCancelReason :: TimedOut as i32 , } ) ) , } ) } else { None } } else { None } ; Some ( DispatchOrTimeoutLA :: Timeout { run_id , resolution , task , } ) } } ; } NewOrCancel :: New ( n ) => n , } ; // It is important that there are no await points after receiving from the channel, as // it would mean dropping this future would cause us to drop the activity request. let ( new_la , attempt ) = match new_or_retry { NewOrRetry :: New ( n ) => { let explicit_attempt_num_or_1 = n . schedule_cmd . attempt . max ( 1 ) ; ( n , explicit_attempt_num_or_1 ) } NewOrRetry :: Retry { in_flight , attempt } => ( in_flight , attempt ) , } ; let orig = new_la . clone ( ) ; let id = ExecutingLAId { run_id : new_la . workflow_exec_info . run_id . clone ( ) , seq_num : new_la . schedule_cmd . seq , } ; let sa = new_la . schedule_cmd ; let mut dat = self . dat . lock ( ) ; // If this request originated from a local backoff task, clear the entry for it. We // don't await the handle because we know it must already be done, and there's no // meaningful value. dat . backing_off_tasks . remove ( & id ) ; // If this task sat in the queue for too long, return a timeout for it instead if let Some ( s2s ) = sa . schedule_to_start_timeout . as_ref ( ) { let sat_for = new_la . schedule_time . elapsed ( ) . unwrap_or_default ( ) ; if sat_for > * s2s { return Some ( DispatchOrTimeoutLA :: Timeout { run_id : new_la . workflow_exec_info . run_id , resolution : LocalActivityResolution { seq : sa . seq , result : LocalActivityExecutionResult :: timeout ( TimeoutType :: ScheduleToStart ) , runtime : sat_for , attempt , backoff : None , original_schedule_time : Some ( new_la . schedule_time ) , } , task : None , } ) ; } } let tt = dat . id_to_tt . get ( & id ) . expect ( " " ) . clone ( ) ; dat . outstanding_activity_tasks . insert ( tt . clone ( ) , LocalInFlightActInfo { la_info : orig , dispatch_time : Instant :: now ( ) , attempt , } , ) ; if let Some ( to ) = dat . timeout_tasks . get_mut ( & id ) { to . mark_started ( ) ; } let ( schedule_to_close , start_to_close ) = sa . close_timeouts . into_sched_and_start ( ) ; Some ( DispatchOrTimeoutLA :: Dispatch ( ActivityTask { task_token : tt . 0 , variant : Some ( activity_task :: Variant :: Start ( Start { workflow_namespace : self . namespace . clone ( ) , workflow_type : new_la . workflow_type , workflow_execution : Some ( new_la . workflow_exec_info ) , activity_id : sa . activity_id , activity_type : sa . activity_type , header_fields : sa . headers , input : sa . arguments , heartbeat_details : vec ! [ ] , scheduled_time : Some ( new_la . schedule_time . into ( ) ) , current_attempt_scheduled_time : Some ( new_la . schedule_time . into ( ) ) , started_time : Some ( SystemTime :: now ( ) . into ( ) ) , attempt , schedule_to_close_timeout : schedule_to_close . map ( Into :: into ) , start_to_close_timeout : start_to_close . map ( Into :: into ) , heartbeat_timeout : None , retry_policy : Some ( sa . retry_policy ) , is_local : true , } ) ) , } ) ) }
Returns the next pending local-activity related action, or None if shutdown has initiated
and there are no more remaining actions to take.
Returns the next pending local - activity related action or None if shutdown has initiated and there are no more remaining actions to take .
&self
&self