Method Exec

pocketflow.go:433–486

Exec iterates through the prepResult slice, calling ExecItemFunc for each item with retries.

(ctx *PfContext, prepResult any)

// Exec iterates through the prepResult slice, calling ExecItemFunc for each item with retries.
func (bn *BatchNode) Exec(ctx *PfContext, prepResult any) (any, error) {
	if prepResult == nil {
		return []any{}, nil // Return empty slice if prep was nil
	}

	// Type assertion to get the slice from Prep result
	items, ok := prepResult.([]any)
	if !ok {
		return nil, newPocketFlowError(fmt.Sprintf("Prep phase of BatchNode %T did not return []any, got %T", bn, prepResult), nil)
	}

	if len(items) == 0 {
		return []any{}, nil // Return empty slice for empty input
	}

	results := make([]any, len(items))
	var itemResult any
	var lastItemErr error
	currentRetry := 0

	for i, item := range items {
		lastItemErr = nil // Reset error for each item
		itemSuccess := false
		for currentRetry = 0; currentRetry < bn.MaxRetries; currentRetry++ {
			itemResult, lastItemErr = bn.ExecItemFunc(ctx, bn.params, item)
			if lastItemErr == nil {
				itemSuccess = true
				break // Success for this item
			}
			if currentRetry < bn.MaxRetries-1 && bn.WaitMilliseconds > 0 {
				time.Sleep(bn.WaitMilliseconds)
			}
		}

		// If all retries failed for this item
		if !itemSuccess {
			if bn.ExecItemFallbackFunc != nil {
				fallbackResult, fallbackErr := bn.ExecItemFallbackFunc(ctx, bn.params, item, lastItemErr)
				if fallbackErr != nil {
					// Fallback failed, return error for the whole batch
					return nil, newPocketFlowError(fmt.Sprintf("ExecItemFallback failed for item %d (%v) in %T after %d retries", i, item, bn, bn.MaxRetries), fallbackErr)
				}
				itemResult = fallbackResult // Use fallback result
				lastItemErr = nil           // Mark as success via fallback
			} else {
				// No fallback, fail the whole batch
				return nil, newPocketFlowError(fmt.Sprintf("ExecItem failed for item %d (%v) in %T after %d retries", i, item, bn, bn.MaxRetries), lastItemErr)
			}
		}
		results[i] = itemResult
	}

	return results, nil // Return the slice of results
}

// Post calls the user-defined PostFunc.
func (bn *BatchNode) Post(ctx *PfContext, prepResult any, execResult any) (string, error) {