kratos-ego

Type-safe batch task processing for Kratos with *errkratos.Erk error handling.

Built on egobatch generic foundation.

---

Features

🎯 Kratos Integration: Specialized with *errkratos.Erk error type ⚡ Batch Processing: Concurrent task execution with type-safe errors 🔄 Flexible Modes: Glide mode and fast-exit mode 🌍 Context Support: Complete context propagation and timeout handling 📋 Result Filtering: OkTasks/WaTasks methods in result aggregation

Installation

go get github.com/yylego/kratos-ego/egokratos

Quick Start

Basic errgroup with Kratos Errors

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/yylego/kratos-ego/erkgroup"
	"github.com/yylego/kratos-errors"
)

func main() {
	ctx := context.Background()
	ego := erkgroup.NewGroup(ctx)

	// Add task 1: takes 100ms to finish
	ego.Go(func(ctx context.Context) *errkratos.Erk {
		time.Sleep(100 * time.Millisecond)
		fmt.Println("Task 1 finished OK")
		return nil
	})

	// Add task 2: takes 50ms to finish
	ego.Go(func(ctx context.Context) *errkratos.Erk {
		time.Sleep(50 * time.Millisecond)
		fmt.Println("Task 2 finished OK")
		return nil
	})

	// Add task 3: takes 80ms to finish
	ego.Go(func(ctx context.Context) *errkratos.Erk {
		time.Sleep(80 * time.Millisecond)
		fmt.Println("Task 3 finished OK")
		return nil
	})

	// Wait until tasks finish and get the first error
	if erk := ego.Wait(); erk != nil {
		fmt.Printf("Got error: %s\n", erk.Error())
	} else {
		fmt.Println("Tasks finished OK")
	}
}

⬆️ Source: Source

Batch Task Processing

package main

import (
	"context"
	"fmt"

	"github.com/go-kratos/kratos/v2/errors"
	"github.com/yylego/kratos-ego"
	"github.com/yylego/kratos-ego/erkgroup"
	"github.com/yylego/kratos-errors/must/erkmust"
)

func main() {
	// Create batch with arguments
	args := []int{1, 2, 3, 4, 5}
	batch := egokratos.NewTaskBatch[int, string](args)

	// Configure glide mode - keep going even when errors happen
	batch.SetGlide(true)

	// Execute batch tasks
	ctx := context.Background()
	ego := erkgroup.NewGroup(ctx)

	batch.EgoRun(ego, func(ctx context.Context, num int) (string, *errors.Error) {
		if num%2 == 0 {
			// Even numbers finish OK
			return fmt.Sprintf("even-%d", num), nil
		}
		// Odd numbers have errors
		return "", errors.BadRequest("ODD_NUMBER", "odd number")
	})

	// In glide mode, ego.Wait() returns nil because errors are captured in tasks
	erkmust.Done(ego.Wait())

	// Get and handle task results
	okTasks := batch.Tasks.OkTasks()
	waTasks := batch.Tasks.WaTasks()

	fmt.Printf("Success: %d, Failed: %d\n", len(okTasks), len(waTasks))

	// Show OK results
	for _, task := range okTasks {
		fmt.Printf("Arg: %d -> Result: %s\n", task.Arg, task.Res)
	}

	// Show failed results
	for _, task := range waTasks {
		fmt.Printf("Arg: %d -> Error: %s\n", task.Arg, task.Erx.Error())
	}
}

⬆️ Source: Source

Core Components

erkgroup.Group

Type-safe errgroup for Kratos:

type Group = erxgroup.Group[*errkratos.Erk]

func NewGroup(ctx context.Context) *Group

TaskBatch[A, R]

Batch task execution:

type TaskBatch[A, R] = egobatch.TaskBatch[A, R, *errkratos.Erk]

func NewTaskBatch[A, R](args []A) *TaskBatch[A, R]

Methods:

• SetGlide(bool) - Configure execution mode
• SetWaCtx(func(error) *errkratos.Erk) - Handle context errors
• EgoRun(ego, func) - Run batch with errgroup

Tasks[A, R]

Task collection with filtering:

type Tasks[A, R] = egobatch.Tasks[A, R, *errkratos.Erk]

Methods:

• OkTasks() - Get success tasks
• WaTasks() - Get failed tasks
• Flatten(func) - Transform results

Examples

See examples for complete demos:

• example1 - Guest order processing
• example2 - Student score processing
• example3 - Multi-step pipeline

Relationship with egobatch

egokratos is built on top of egobatch using type aliases:

// egokratos provides Kratos-specific types
type Task[A, R] = egobatch.Task[A, R, *errkratos.Erk]
type Tasks[A, R] = egobatch.Tasks[A, R, *errkratos.Erk]
type TaskBatch[A, R] = egobatch.TaskBatch[A, R, *errkratos.Erk]

This approach:

• ✅ Reduces code duplication
• ✅ Maintains type-safe operations
• ✅ Provides Kratos-optimized API
• ✅ Benefits from egobatch improvements

License

MIT License. See LICENSE.

Contributing

Issues and pull requests welcome!