### Go Karta Pipeline Usage Example

Source: https://github.com/shengyanli1982/karta/blob/main/README.md

Demonstrates how to use the Karta Pipeline component in Go. It shows setting up configuration, creating a pipeline with a fake delaying queue, submitting tasks with default and custom handler functions, and stopping the pipeline.

```go
package main

import (
	"fmt"
	"time"

	k "github.com/shengyanli1982/karta"
	wkq "github.com/shengyanli1982/workqueue/v2"
)

// handleFunc 是一个处理函数，它接收一个任意类型的消息，打印该消息，然后返回该消息和nil错误。
// handleFunc is a handler function that takes a message of any type, prints the message, and then returns the message and a nil error.
func handleFunc(msg any) (any, error) {
	// 打印接收到的消息。
	// Print the received message.
	fmt.Println("default:", msg)

	// 返回接收到的消息和nil错误。
	// Return the received message and a nil error.
	return msg, nil
}

func main() {
	// 创建一个新的配置对象。
	// Create a new configuration object.
	c := k.NewConfig()

	// 设置处理函数和工作线程数量。
	// Set the handler function and the number of worker threads.
	c.WithHandleFunc(handleFunc).WithWorkerNumber(2)

	// 创建一个新的假延迟队列。
	// Create a new fake delaying queue.
	queue := k.NewFakeDelayingQueue(wkq.NewQueue(nil))

	// 使用队列和配置创建一个新的管道。
	// Create a new pipeline using the queue and configuration.
	pl := k.NewPipeline(queue, c)

	// 确保在main函数结束时停止管道。
	// Ensure the pipeline is stopped when the main function ends.
	defer pl.Stop()

	// 提交一个消息到管道。
	// Submit a message to the pipeline.
	_ = pl.Submit("foo")

	// 使用特定的处理函数提交一个消息到管道。
	// Submit a message to the pipeline using a specific handler function.
	_ = pl.SubmitWithFunc(func(msg any) (any, error) {
		// 打印接收到的消息。
		// Print the received message.
		fmt.Println("SpecFunc:", msg)

		// 返回接收到的消息和nil错误。
		// Return the received message and a nil error.
		return msg, nil
	}, "bar")

	// 暂停一秒钟。
	// Pause for one second.
	time.Sleep(time.Second)
}
```

--------------------------------

### Go - Karta Group Processing Example

Source: https://github.com/shengyanli1982/karta/blob/main/README.md

Demonstrates the use of the Karta Group component for batch processing tasks. It configures the number of workers, handle function, and result recording, then maps a function to a set of input values.

```Go
package main

import (
	"time"

	k "github.com/shengyanli1982/karta"
)

// handleFunc 是一个处理函数，它接收一个任意类型的消息，暂停一段时间（消息值的100毫秒），然后返回该消息和nil错误。
// handleFunc is a handler function that takes a message of any type, sleeps for a duration (100 milliseconds of the message value), and then returns the message and a nil error.
func handleFunc(msg any) (any, error) {
	// 将消息转换为整数，然后暂停该整数值的100毫秒。
	// Convert the message to an integer, then pause for 100 milliseconds of the integer value.
	time.Sleep(time.Duration(msg.(int)) * time.Millisecond * 100)

	// 返回接收到的消息和nil错误。
	// Return the received message and a nil error.
	return msg, nil
}

func main() {
	// 创建一个新的配置对象。
	// Create a new configuration object.
	c := k.NewConfig()
	// 设置处理函数，工作线程数量和结果处理。
	// Set the handler function, the number of worker threads, and result processing.
	c.WithHandleFunc(handleFunc).WithWorkerNumber(2).WithResult()

	// 使用配置创建一个新的工作组。
	// Create a new work group using the configuration.
	g := k.NewGroup(c)

	// 确保在main函数结束时停止工作组。
	// Ensure the work group is stopped when the main function ends.
	defer g.Stop()

	// 将处理函数映射到一组输入值。
	// Map the handler function to a set of input values.
	r0 := g.Map([]any{3, 5, 2})

	// 打印第一个结果的整数值。
	// Print the integer value of the first result.
	println(r0[0].(int))
}
```

--------------------------------

### Go: Pipeline with Real Delaying Queue for Scheduled Execution

Source: https://context7.com/shengyanli1982/karta/llms.txt

Illustrates the use of a real delaying queue in Go for scheduling tasks with specific execution delays. This example configures a pipeline with a timestamp handler and a limited number of workers, then submits immediate tasks alongside tasks scheduled to run after 1 and 2 seconds. It highlights the time-based execution order.

```go
package main

import (
    "fmt"
    "time"
    k "github.com/shengyanli1982/karta"
    wkq "github.com/shengyanli1982/workqueue/v2"
)

func timestampHandler(msg any) (any, error) {
    fmt.Printf("[%s] Processed: %v\n",
        time.Now().Format("15:04:05.000"), msg)
    return msg, nil
}

func main() {
    config := k.NewConfig()
    config.WithHandleFunc(timestampHandler).WithWorkerNumber(2)

    // Use real delaying queue for time-based scheduling
    queue := wkq.NewDelayingQueue(nil)
    pipeline := k.NewPipeline(queue, config)
    defer pipeline.Stop()

    fmt.Printf("Start time: %s\n", time.Now().Format("15:04:05.000"))

    // Submit immediate task
    pipeline.Submit("immediate")

    // Submit task with 1 second delay
    pipeline.SubmitAfter("delayed-1s", 1*time.Second)

    // Submit task with 2 second delay using custom handler
    pipeline.SubmitAfterWithFunc(func(msg any) (any, error) {
        fmt.Printf("[%s] Custom delayed: %v\n",
            time.Now().Format("15:04:05.000"), msg)
        return msg, nil
    }, "delayed-2s-custom", 2*time.Second)

    // Wait for all tasks
    time.Sleep(3 * time.Second)
}

// Output:
// Start time: 14:30:00.000
// [14:30:00.001] Processed: immediate
// [14:30:01.002] Processed: delayed-1s
// [14:30:02.003] Custom delayed: delayed-2s-custom

```

--------------------------------

### Go: Asynchronous Task Pipeline with Fake Delaying Queue

Source: https://context7.com/shengyanli1982/karta/llms.txt

Demonstrates setting up an asynchronous task processing pipeline in Go using a fake delaying queue. It shows how to configure the pipeline with a default handler, set the number of concurrent workers, submit tasks with default and custom handlers, and schedule delayed tasks. The example also includes checking the active worker count and ensuring all tasks are completed.

```go
package main

import (
    "fmt"
    "time"
    k "github.com/shengyanli1982/karta"
    wkq "github.com/shengyanli1982/workqueue/v2"
)

// Default handler for pipeline
func defaultHandler(msg any) (any, error) {
    fmt.Printf("[Default Handler] Processing: %v\n", msg)
    time.Sleep(100 * time.Millisecond)
    return msg, nil
}

func main() {
    // Configure pipeline
    config := k.NewConfig()
    config.WithHandleFunc(defaultHandler). 
           WithWorkerNumber(5)  // Maximum 5 concurrent workers

    // Create a delaying queue (required for Pipeline)
    queue := k.NewFakeDelayingQueue(wkq.NewQueue(nil))

    // Create pipeline instance
    pipeline := k.NewPipeline(queue, config)
    defer pipeline.Stop()

    // Submit task with default handler
    err := pipeline.Submit("task-1")
    if err != nil {
        fmt.Printf("Error submitting task: %v\n", err)
        return
    }

    // Submit task with custom handler function
    err = pipeline.SubmitWithFunc(func(msg any) (any, error) {
        fmt.Printf("[Custom Handler] Processing: %v\n", msg)
        return fmt.Sprintf("Custom result: %v", msg), nil
    }, "task-2")
    if err != nil {
        fmt.Printf("Error submitting task: %v\n", err)
        return
    }

    // Submit delayed task (executes after 500ms)
    err = pipeline.SubmitAfter("delayed-task", 500*time.Millisecond)
    if err != nil {
        fmt.Printf("Error submitting delayed task: %v\n", err)
        return
    }

    // Check current worker count
    workerCount := pipeline.GetWorkerNumber()
    fmt.Printf("Active workers: %d\n", workerCount)

    // Wait for all tasks to complete
    time.Sleep(2 * time.Second)
}

// Output:
// [Default Handler] Processing: task-1
// [Custom Handler] Processing: task-2
// Active workers: 1
// [Default Handler] Processing: delayed-task

```

--------------------------------

### Go Queue and DelayingQueue Interface Definitions

Source: https://github.com/shengyanli1982/karta/blob/main/README.md

Defines the Queue and DelayingQueue interfaces in Go. The Queue interface includes methods for putting, getting, marking as done, shutting down, and checking the status of a queue. The DelayingQueue interface extends Queue by adding a PutWithDelay method.

```go
package main

import (
	"fmt"
	"time"

	k "github.com/shengyanli1982/karta"
	wkq "github.com/shengyanli1982/workqueue/v2"
)

// Queue 接口定义了一个队列应该具备的基本操作。
// The Queue interface defines the basic operations that a queue should have.
type Queue = interface {
	// Put 方法用于将元素放入队列。
	// The Put method is used to put an element into the queue.
	Put(value interface{}) error

	// Get 方法用于从队列中获取元素。
	// The Get method is used to get an element from the queue.
	Get() (value interface{}, err error)

	// Done 方法用于标记元素处理完成。
	// The Done method is used to mark the element as done.
	Done(value interface{})

	// Shutdown 方法用于关闭队列。
	// The Shutdown method is used to shut down the queue.
	Shutdown()

	// IsClosed 方法用于检查队列是否已关闭。
	// The IsClosed method is used to check if the queue is closed.
	IsClosed() bool
}

// DelayingQueue 接口继承了 Queue 接口，并添加了一个 PutWithDelay 方法，用于将元素延迟放入队列。
// The DelayingQueue interface inherits from the Queue interface and adds a PutWithDelay method to put an element into the queue with delay.
type DelayingQueue = interface {
	Queue

	// PutWithDelay 方法用于将元素延迟放入队列。
	// The PutWithDelay method is used to put an element into the queue with delay.
	PutWithDelay(value interface{}, delay int64) error
}
```

--------------------------------

### Track Task Errors via Callbacks in Karta (Go)

Source: https://context7.com/shengyanli1982/karta/llms.txt

Demonstrates registering a callback to observe lifecycle events and count successes/failures while processing a list of tasks concurrently. Uses Group.Map to submit tasks and collects ordered results. The example fails on even numbers to illustrate error handling.

```go
package main

import (
    "errors"
    "fmt"
    k "github.com/shengyanli1982/karta"
)

// Callback that logs task lifecycle events
type ErrorTrackingCallback struct {
    successCount int
    errorCount   int
}

func (cb *ErrorTrackingCallback) OnBefore(msg any) {
    fmt.Printf("=> Starting: %v\n", msg)
}

func (cb *ErrorTrackingCallback) OnAfter(msg, result any, err error) {
    if err != nil {
        cb.errorCount++
        fmt.Printf("=> Failed: %v (error: %v)\n", msg, err)
    } else {
        cb.successCount++
        fmt.Printf("=> Success: %v (result: %v)\n", msg, result)
    }
}

// Handler that may fail
func riskyHandler(msg any) (any, error) {
    val, ok := msg.(int)
    if !ok {
        return nil, errors.New("invalid type")
    }

    // Fail on even numbers
    if val%2 == 0 {
        return nil, fmt.Errorf("rejected even number: %d", val)
    }

    return val * 10, nil
}

func main() {
    callback := &ErrorTrackingCallback{}

    config := k.NewConfig()
    config.WithHandleFunc(riskyHandler).
           WithCallback(callback).
           WithWorkerNumber(2).
           WithResult()

    group := k.NewGroup(config)
    defer group.Stop()

    // Process mixed input
    tasks := []any{1, 2, 3, 4, 5}
    results := group.Map(tasks)

    fmt.Printf("\nFinal Statistics:\n")
    fmt.Printf("Success: %d, Errors: %d\n",
        callback.successCount, callback.errorCount)
    fmt.Printf("Results: %v\n", results)
}

// Output:
// => Starting: 1
// => Success: 1 (result: 10)
// => Starting: 2
// => Failed: 2 (error: rejected even number: 2)
// => Starting: 3
// => Success: 3 (result: 30)
// => Starting: 4
// => Failed: 4 (error: rejected even number: 4)
// => Starting: 5
// => Success: 5 (result: 50)
//
// Final Statistics:
// Success: 3, Errors: 2
// Results: [10 <nil> 30 <nil> 50]
```

--------------------------------

### Batch process tasks using Group with fixed worker pool

Source: https://context7.com/shengyanli1982/karta/llms.txt

Shows how to use Karta's Group model for concurrent batch task processing with a fixed worker pool. Tasks are submitted as a slice and processed concurrently while maintaining result order. The example includes result collection and demonstrates processing multiple batches sequentially. Requires a configured handler function and result collection enabled.

```go
package main

import (
    "fmt"
    "time"
    k "github.com/shengyanli1982/karta"
)

// Handler function that simulates work
func processItem(msg any) (any, error) {
    // Simulate processing time
    time.Sleep(time.Duration(msg.(int)) * 100 * time.Millisecond)

    // Return processed result
    return fmt.Sprintf("Processed: %d", msg.(int)), nil
}

func main() {
    // Configure the group
    config := k.NewConfig()
    config.WithHandleFunc(processItem).
           WithWorkerNumber(3).
           WithResult()  // Important: enable result collection

    // Create group instance
    group := k.NewGroup(config)
    defer group.Stop()

    // Submit batch of tasks
    tasks := []any{1, 2, 3, 4, 5, 6}
    results := group.Map(tasks)

    // Process results in original order
    for i, result := range results {
        fmt.Printf("Task %d result: %v\n", i, result)
    }

    // Submit another batch
    moreTasks := []any{7, 8, 9}
    moreResults := group.Map(moreTasks)

    fmt.Printf("Second batch completed: %d results\n", len(moreResults))
}

// Output (order preserved):
// Task 0 result: Processed: 1
// Task 1 result: Processed: 2
// Task 2 result: Processed: 3
// Task 3 result: Processed: 4
// Task 4 result: Processed: 5
// Task 5 result: Processed: 6
// Second batch completed: 3 results
```

--------------------------------

### Create Karta configuration with custom handler and callback

Source: https://context7.com/shengyanli1982/karta/llms.txt

Demonstrates how to create and customize a Karta configuration object with worker count, custom handler function, lifecycle callbacks, and result collection. This configuration can be used with both Group and Pipeline processing models. Requires importing the karta package and implementing callback interface methods.

```go
package main

import (
    "fmt"
    k "github.com/shengyanli1982/karta"
)

// Custom callback to track task lifecycle
type MyCallback struct{}

func (cb *MyCallback) OnBefore(msg any) {
    fmt.Printf("Starting to process: %v\n", msg)
}

func (cb *MyCallback) OnAfter(msg, result any, err error) {
    if err != nil {
        fmt.Printf("Failed to process %v: %v\n", msg, err)
    } else {
        fmt.Printf("Completed processing %v with result: %v\n", msg, result)
    }
}

// Custom handler function
func customHandler(msg any) (any, error) {
    // Transform the message (e.g., multiply by 2)
    if val, ok := msg.(int); ok {
        return val * 2, nil
    }
    return msg, nil
}

func main() {
    // Create configuration with all options
    config := k.NewConfig()
    config.WithWorkerNumber(4).              // Set 4 worker goroutines
           WithHandleFunc(customHandler).     // Set custom handler function
           WithCallback(&MyCallback{}).       // Set lifecycle callbacks
           WithResult()                       // Enable result collection

    // Config is now ready to use with Group or Pipeline
    fmt.Println("Configuration created successfully")
}

// Output:
// Configuration created successfully
```

--------------------------------

### Integrate Custom Priority Queue into Karta Pipeline (Go)

Source: https://context7.com/shengyanli1982/karta/llms.txt

Shows how to implement a minimal queue satisfying DelayingQueue and plug it into a Pipeline. The custom queue provides thread-safe Put/Get with basic close semantics; Pipeline uses it for asynchronous task processing with configured workers.

```go
package main

import (
    "fmt"
    "sync"
    k "github.com/shengyanli1982/karta"
)

// Simple priority queue implementation
type PriorityQueue struct {
    items  []any
    mu     sync.Mutex
    closed bool
}

func (pq *PriorityQueue) Put(value interface{}) error {
    pq.mu.Lock()
    defer pq.mu.Unlock()
    if pq.closed {
        return fmt.Errorf("queue is closed")
    }
    pq.items = append(pq.items, value)
    return nil
}

func (pq *PriorityQueue) Get() (interface{}, error) {
    pq.mu.Lock()
    defer pq.mu.Unlock()
    if len(pq.items) == 0 {
        return nil, fmt.Errorf("queue is empty")
    }
    item := pq.items[0]
    pq.items = pq.items[1:]
    return item, nil
}

func (pq *PriorityQueue) Done(value interface{}) {}

func (pq *PriorityQueue) Shutdown() {
    pq.mu.Lock()
    defer pq.mu.Unlock()
    pq.closed = true
}

func (pq *PriorityQueue) IsClosed() bool {
    pq.mu.Lock()
    defer pq.mu.Unlock()
    return pq.closed
}

func (pq *PriorityQueue) PutWithDelay(value interface{}, delay int64) error {
    return pq.Put(value)  // Simple implementation
}

func main() {
    // Create custom queue
    customQueue := &PriorityQueue{items: make([]any, 0)}

    config := k.NewConfig()
    config.WithHandleFunc(func(msg any) (any, error) {
        fmt.Printf("Processing: %v\n", msg)
        return msg, nil
    }).WithWorkerNumber(2)

    pipeline := k.NewPipeline(customQueue, config)
    defer pipeline.Stop()

    // Use pipeline with custom queue
    pipeline.Submit("task-A")
    pipeline.Submit("task-B")

    fmt.Println("Tasks submitted to custom queue")
}

// Output:
// Tasks submitted to custom queue
// Processing: task-A
// Processing: task-B
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.