goiteratorhelper

package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/ngicks/go-iterator-helper/hiter"
	"github.com/ngicks/go-iterator-helper/hiter/async"
)

func main() {
	var (
		wg sync.WaitGroup
		in = make(chan int)
	)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	wg.Add(1)
	go func() {
		defer wg.Done()
		ticker := time.NewTicker(500 * time.Nanosecond)
		defer ticker.Stop()
		_, _ = hiter.ChanSend(ctx, in, hiter.Tap(func(int) { <-ticker.C }, hiter.Range(0, 20)))
		close(in)
	}()

	first := true
	var count int
	for c := range async.Chunk(time.Microsecond, 5, hiter.Chan(ctx, in)) {
		count++
		for _, i := range c {
			if !first {
				fmt.Print(", ")
			}
			first = false
			fmt.Printf("%d", i)
		}
	}
	fmt.Println()
	wg.Wait()
	fmt.Printf("count > 0 = %t\n", count > 0)
}

Output:

0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19
count > 0 = true

Example async worker channel demonstrates usage of [hiter.Chan], [hiter.ChanSend]. It sends values from seq to worker running on separates goroutines. Workers work on values and then send results back to the main goroutine.

package main

import (
	"context"
	"fmt"
	"maps"
	"slices"
	"sync"

	"github.com/ngicks/go-iterator-helper/hiter"
	"github.com/ngicks/go-iterator-helper/hiter/iterable"
	"github.com/ngicks/go-iterator-helper/x/exp/xiter"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	works := []string{"foo", "bar", "baz"}

	in := make(chan string, 5)
	out := make(chan hiter.KeyValue[string, error])

	var wg sync.WaitGroup
	wg.Add(3)
	for range 3 {
		go func() {
			defer wg.Done()
			_, _ = hiter.ChanSend(
				ctx,
				out,
				xiter.Map(
					func(s string) hiter.KeyValue[string, error] {
						return hiter.KeyValue[string, error]{
							K: "✨" + s + "✨" + s + "✨",
							V: nil,
						}
					},
					hiter.Chan(ctx, in),
				),
			)
		}()
	}

	var wg2 sync.WaitGroup
	wg2.Add(1)
	go func() {
		defer wg2.Done()
		wg.Wait()
		close(out)
	}()

	_, _ = hiter.ChanSend(ctx, in, slices.Values(works))
	close(in)

	results := maps.Collect(hiter.FromKeyValue(hiter.Chan(ctx, out)))

	for result, err := range iterable.MapSorted[string, error](results).Iter2() {
		fmt.Printf("result = %s, err = %v\n", result, err)
	}

	wg2.Wait()

}

Output:

result = ✨bar✨bar✨, err = <nil>
result = ✨baz✨baz✨, err = <nil>
result = ✨foo✨foo✨, err = <nil>

Example async worker map demonstrates usage of async.Map. At the surface it is similar to [xiter.Map2]. Actually it calls mapper in separate goroutine. If you don't care about order of element, just send values to workers through a channel and send back through another channel.

package main

import (
	"context"
	"fmt"
	"slices"

	"github.com/ngicks/go-iterator-helper/hiter/async"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	works := []string{"foo", "bar", "baz"}

	// The order is kept.
	for result, err := range async.Map(
		ctx,
		/*queueLimit*/ 10,
		/*workerLimit*/ 5,
		/*mapper*/ func(ctx context.Context, s string) (string, error) {
			return "✨" + s + "✨" + s + "✨", nil
		},
		slices.Values(works),
	) {
		fmt.Printf("result = %s, err = %v\n", result, err)
	}
}

Output:

result = ✨foo✨foo✨, err = <nil>
result = ✨bar✨bar✨, err = <nil>
result = ✨baz✨baz✨, err = <nil>

package main

import (
	"context"
	"fmt"
	"slices"

	"github.com/ngicks/go-iterator-helper/hiter/async"
	"github.com/ngicks/go-iterator-helper/hiter/mapper"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	works := []string{"foo", "bar", "baz"}

	workerCtx, cancelWorker := context.WithCancel(context.Background())
	defer cancelWorker()

	for result, err := range async.Map(
		ctx,
		/*queueLimit*/ 1,
		/*workerLimit*/ 1,
		/*mapper*/ func(ctx context.Context, s string) (string, error) {
			combined, cancel := context.WithCancel(ctx)
			defer cancel()
			go func() {
				select {
				case <-ctx.Done():
				case <-combined.Done():
				case <-workerCtx.Done():
				}
				cancel()
			}()
			if combined.Err() != nil {
				return "", combined.Err()
			}
			return "✨" + s + "✨" + s + "✨", nil
		},
		mapper.Cancellable(1, workerCtx, slices.Values(works)),
	) {
		fmt.Printf("result = %s, err = %v\n", result, err)
		cancelWorker()
	}
}

Output:

result = ✨foo✨foo✨, err = <nil>
result = ✨bar✨bar✨, err = <nil>

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"maps"
	"os"

	"github.com/ngicks/go-iterator-helper/hiter/encodingiter"
	"github.com/ngicks/go-iterator-helper/hiter/errbox"
	"github.com/ngicks/go-iterator-helper/x/exp/xiter"
)

func main() {
	src := []byte(`
	{"foo":"foo"}
	{"bar":"bar"}
	{"baz":"baz"}
	`)

	rawDec := json.NewDecoder(bytes.NewReader(src))
	dec := errbox.New(encodingiter.Decode[map[string]string](rawDec))

	enc := json.NewEncoder(os.Stdout)

	err := encodingiter.Encode(
		enc,
		xiter.Map(
			func(m map[string]string) map[string]string {
				return maps.Collect(
					xiter.Map2(
						func(k, v string) (string, string) { return k + k, v + v },
						maps.All(m),
					),
				)
			},
			dec.IntoIter(),
		),
	)

	fmt.Printf("dec error = %v\n", dec.Err())
	fmt.Printf("enc error = %v\n", err)
}

Output:

{"foofoo":"foofoo"}
{"barbar":"barbar"}
{"bazbaz":"bazbaz"}
dec error = <nil>
enc error = <nil>

Example error handle demonstrates various way to handle error.

package main

import (
	"errors"
	"fmt"
	"slices"

	"github.com/ngicks/go-iterator-helper/hiter"
	"github.com/ngicks/go-iterator-helper/hiter/errbox"
	"github.com/ngicks/go-iterator-helper/hiter/mapper"
	"github.com/ngicks/go-iterator-helper/x/exp/xiter"
)

func main() {
	var (
		errSample  = errors.New("sample")
		errSample2 = errors.New("sample2")
	)

	erroneous := hiter.Pairs(
		hiter.Range(0, 6),
		xiter.Concat(
			hiter.Repeat(error(nil), 2),
			hiter.Repeat(errSample2, 2),
			hiter.Once(errSample),
			hiter.Once(error(nil)),
		),
	)

	fmt.Println("TryFind:")
	v, idx, err := hiter.TryFind(func(i int) bool { return i > 0 }, erroneous)
	fmt.Printf("v = %d, idx = %d, err = %v\n", v, idx, err)
	v, idx, err = hiter.TryFind(func(i int) bool { return i > 5 }, erroneous)
	fmt.Printf("v = %d, idx = %d, err = %v\n", v, idx, err)
	fmt.Println()

	fmt.Println("TryForEach:")
	err = hiter.TryForEach(func(i int) { fmt.Printf("i = %d\n", i) }, erroneous)
	fmt.Printf("err = %v\n", err)
	fmt.Println()

	fmt.Println("TryReduce:")
	collected, err := hiter.TryReduce(func(c []int, i int) []int { return append(c, i) }, nil, erroneous)
	fmt.Printf("collected = %#v, err = %v\n", collected, err)
	fmt.Println()

	fmt.Println("HandleErr:")
	var handled error
	collected = slices.Collect(
		mapper.HandleErr(
			func(i int, err error) bool {
				handled = err
				return errors.Is(err, errSample2)
			},
			erroneous,
		),
	)
	fmt.Printf("collected = %#v, err = %v\n", collected, handled)
	fmt.Println()

	fmt.Println("*errbox.Box:")
	box := errbox.New(erroneous)
	collected = slices.Collect(box.IntoIter())
	fmt.Printf("collected = %#v, err = %v\n", collected, box.Err())
	fmt.Println()
}

Output:

TryFind:
v = 1, idx = 1, err = <nil>
v = 0, idx = -1, err = sample2

TryForEach:
i = 0
i = 1
err = sample2

TryReduce:
collected = []int{0, 1}, err = sample2

HandleErr:
collected = []int{0, 1}, err = sample

*errbox.Box:
collected = []int{0, 1}, err = sample2

package main

import (
	"fmt"

	"github.com/ngicks/go-iterator-helper/hiter"
	"github.com/ngicks/go-iterator-helper/hiter/iterable"
	"github.com/ngicks/go-iterator-helper/x/exp/xiter"
)

func main() {
	// iterator that yields 0 to 9 sequentially.
	src := hiter.Range(0, 10)

	fmt.Println("It replays data if break-ed and resumed.")

	count := 3
	first := true
	for v := range src {
		count--
		if count < 0 {
			break
		}
		if !first {
			fmt.Print(", ")
		}
		first = false
		fmt.Printf("%d", v)
	}
	fmt.Println()
	fmt.Println("break and resume")
	first = true
	for v := range xiter.Limit(src, 3) {
		if !first {
			fmt.Print(", ")
		}
		first = false
		fmt.Printf("%d", v)
	}
	fmt.Print("\n\n")

	fmt.Println("converting it to be resumable.")
	resumable := iterable.NewResumable(src)

	v0, _ := hiter.First(resumable.IntoIter())
	fmt.Printf("first:  %d\n", v0)
	v1, _ := hiter.First(resumable.IntoIter())
	fmt.Printf("second: %d\n", v1)

	fmt.Println()
	fmt.Println("reconnect them to whole iterator.")
	first = true
	for v := range xiter.Concat(hiter.Once(v0), hiter.Once(v1), resumable.IntoIter()) {
		if !first {
			fmt.Print(", ")
		}
		first = false
		fmt.Printf("%d", v)
	}
	fmt.Println()

	fmt.Println("\nYou can achieve above also with iterable.Peekable")
	peekable := iterable.NewPeekable(src)
	fmt.Printf("%#v\n", peekable.Peek(5))
	first = true
	for v := range peekable.IntoIter() {
		if !first {
			fmt.Print(", ")
		}
		first = false
		fmt.Printf("%d", v)
	}
	fmt.Println()

}

Output:

It replays data if break-ed and resumed.
0, 1, 2
break and resume
0, 1, 2

converting it to be resumable.
first:  0
second: 1

reconnect them to whole iterator.
0, 1, 2, 3, 4, 5, 6, 7, 8, 9

You can achieve above also with iterable.Peekable
[]int{0, 1, 2, 3, 4}
0, 1, 2, 3, 4, 5, 6, 7, 8, 9

package main

import (
	"fmt"
	"sync"

	"github.com/ngicks/go-iterator-helper/hiter"
	"github.com/ngicks/go-iterator-helper/hiter/tee"
	"github.com/ngicks/go-iterator-helper/x/exp/xiter"
)

func main() {
	src := hiter.Range(0, 5)
	seqPiped, seq := tee.TeeSeqPipe(0, src)

	var found bool

	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		found = hiter.Contains(3, seqPiped.IntoIter())
		// Don't forget to discard all elements from seq!
		// Without this, tee could not proceed.
		hiter.Discard(seqPiped.IntoIter())
	}()

	for i := range xiter.Map(func(i int) int { return i * i }, seq.IntoIter()) {
		fmt.Printf("i = %02d\n", i)
	}
	wg.Wait()
	fmt.Printf("\nfound=%t\n", found)
}

Output:

i = 00
i = 01
i = 04
i = 09
i = 16

found=true