worker := func(ctx context.Context, input interface{}) (interface{}, error) {
    num := input.(int)
    return num+1, nil
}

inputs := []interface{}{1,2,3,4,5}
outputs, err := channelx.RunInParallel(context.Background(), inputs, worker, 4)

logger := channelx.NewConsoleLogger()
eventBus := channelx.NewEventBus(logger,  4,4,2, time.Second, 5 * time.Second)
eventBus.Start()

handler := NewExampleHandler(logger)
eventBus.Subscribe(ExampleEventID, handler)
eventBus.Publish(NewExampleEvent())

eventBus.Stop()

promise := NewPromise(func() (res interface{}, err error) {
    // do work asynchronously here
    reuturn
}).Then(func(input interface{}) (interface{}, error) {
    // here is the succss handler, which aslo runs asynchronously 
}, func(err error) interface{} {
    // here is the error handler, which aslo runs asynchronously 
})

// await: wait until it completes.
res, _ := promise.Done()

actor := NewActor(SetActorBuffer(0))
defer actor.Close()

// do some work asynchroniously.
call := actor.Do(func() (interface{}, error) {
    time.Sleep(0 * time.Second)
    return 0, nil
})

// can to some other synchroniouse work here
// ......

// wait for the call completes.
res, err := call.Done()

var multipleChan = make(chan int, 4)
var minusChan = make(chan int, 4)
var harvestChan = make(chan int, 4)

defer close(multipleChan)
defer close(minusChan)
defer close(harvestChan)

go func() {
    for i:=1;i<=100;i++{
        multipleChan <- i
    }
}()

for i:=0; i<4; i++{
    go func() {
        for data := range multipleChan {
            minusChan <- data * 2
            time.Sleep(10* time.Millisecond)
        }
    }()

    go func() {
        for data := range minusChan {
            harvestChan <- data - 1
            time.Sleep(10* time.Millisecond)
        }
    }()
}

var sum = 0
var index = 0
for data := range harvestChan{
    sum += data
    index++
    if index == 100{
        break
    }
}

fmt.Println(sum)

var sum = 0

NewChannelStream(func(seedChan chan<- Item, quitChannel chan struct{}) {
    for i:=1; i<=100;i++{
        seedChan <- Item{Data:i}
    }
    close(seedChan) //don't forget to close it
}).Pipe(func(Item Item) Item {
    return Item{Data: Item.Data.(int) * 2}
}).Pipe(func(Item Item) Item {
    return Item{Data: Item.Data.(int) - 1}
}).Harvest(func(Item Item) {
    sum += Item.Data.(int)
})

fmt.Println(sum)

// YourKnownType, YourBatchHandler, yourRequest are faked type or object

batchProcess := func(items []interface{}) error {
    var arr YourKnownType 
    for _, item := range items{
        ykt := item.(YourKnownType)
        arr = append(arr, ykt)
    }
    
    YourBatchHandler(arr)
}

aggregator := NewAggregator(batchProcess)

aggregator.Start()

aggregator.Enqueue(yourRequest)

aggregator.Stop()