asyncigo

Basic usage of UntilErr.

package main

import (
	"errors"
	"fmt"

	"github.com/arvidfm/asyncigo"
)

func main() {
	it := asyncigo.AsyncIter(func(yield func(int) error) error {
		for i := range 5 {
			if err := yield(i); err != nil {
				return err
			}
		}

		return errors.New("oops")
	})

	var err error
	for i := range it.UntilErr(&err) {
		fmt.Println(i)
	}
	fmt.Println(err)
}

Output:

0
1
2
3
4
oops

UntilErr is particularly useful for ranging over network streams.

package main

import (
	"context"
	"errors"
	"fmt"
	"net"
	"os"
	"path/filepath"
	"sync"
	"syscall"
	"time"

	"github.com/arvidfm/asyncigo"
)

func main() {
	defer serveFile("stream_example1.txt", "6172")()

	if err := asyncigo.NewEventLoop().Run(context.Background(), func(ctx context.Context) error {
		stream, _ := asyncigo.RunningLoop(ctx).Dial(ctx, "tcp", "localhost:6172")

		var err error
		for line := range stream.Lines(ctx).UntilErr(&err) {
			fmt.Printf("got line: %s", line)
		}

		return err
	}); err != nil {
		panic(err)
	}
}

func serveFile(path, port string) (close func()) {
	data, err := os.ReadFile(filepath.Join("tests", path))
	if err != nil {
		panic(err)
	}

	address := net.JoinHostPort("localhost", port)
	var waiter sync.WaitGroup
	var l net.Listener
	waiter.Add(1)
	go func() {
		defer waiter.Done()

		var err error
		l, err = net.Listen("tcp", address)
		if err != nil {
			panic(err)
		}

		for {
			conn, err := l.Accept()
			if err != nil {
				return
			}

			if _, err := conn.Write(data); err != nil {
				panic(err)
			}
			_ = conn.Close()
		}
	}()

	for {
		conn, err := net.Dial("tcp", address)
		if errors.Is(err, syscall.ECONNREFUSED) {
			time.Sleep(time.Millisecond * 10)
		} else if err != nil {
			panic(err)
		} else {
			_ = conn.Close()
			break
		}
	}

	return func() {
		_ = l.Close()
		waiter.Wait()
	}
}

Output:

got line: Lorem ipsum dolor sit amet.
got line: Donec non velit consequat.
got line: Donec interdum in nulla ac scelerisque.
got line: Duis commodo, neque ac luctus eleifend.
got line: Fusce lacinia id quam ac porttitor.

Since UntilErr returns a standard single-valued iterator, you can easily manipulate the iterator using functions like asyncigo.Map and asyncigo.Enumerate. This example shows how you could combine multiple asynchronous iterators using asyncigo.Chain.

package main

import (
	"context"
	"errors"
	"fmt"
	"net"
	"os"
	"path/filepath"
	"sync"
	"syscall"
	"time"

	"github.com/arvidfm/asyncigo"
)

func main() {
	defer serveFile("stream_example1.txt", "6172")()
	defer serveFile("stream_example2.txt", "6173")()

	if err := asyncigo.NewEventLoop().Run(context.Background(), func(ctx context.Context) error {
		loop := asyncigo.RunningLoop(ctx)

		var err error
		for line := range asyncigo.Chain(
			loop.DialLines(ctx, "tcp", "localhost:6172").UntilErr(&err),
			loop.DialLines(ctx, "tcp", "localhost:6173").UntilErr(&err),
		) {
			fmt.Printf("got line: %s", line)
		}

		return err
	}); err != nil {
		panic(err)
	}
}

func serveFile(path, port string) (close func()) {
	data, err := os.ReadFile(filepath.Join("tests", path))
	if err != nil {
		panic(err)
	}

	address := net.JoinHostPort("localhost", port)
	var waiter sync.WaitGroup
	var l net.Listener
	waiter.Add(1)
	go func() {
		defer waiter.Done()

		var err error
		l, err = net.Listen("tcp", address)
		if err != nil {
			panic(err)
		}

		for {
			conn, err := l.Accept()
			if err != nil {
				return
			}

			if _, err := conn.Write(data); err != nil {
				panic(err)
			}
			_ = conn.Close()
		}
	}()

	for {
		conn, err := net.Dial("tcp", address)
		if errors.Is(err, syscall.ECONNREFUSED) {
			time.Sleep(time.Millisecond * 10)
		} else if err != nil {
			panic(err)
		} else {
			_ = conn.Close()
			break
		}
	}

	return func() {
		_ = l.Close()
		waiter.Wait()
	}
}

Output:

got line: Lorem ipsum dolor sit amet.
got line: Donec non velit consequat.
got line: Donec interdum in nulla ac scelerisque.
got line: Duis commodo, neque ac luctus eleifend.
got line: Fusce lacinia id quam ac porttitor.
got line: 生麦生米生卵
got line: すもももももももものうち
got line: 東京特許許可局長今日急遽休暇許可却下
got line: 斜め77度の並びで泣く泣く嘶くナナハン7台難なく並べて長眺め

package main

import (
	"context"
	"fmt"

	"github.com/arvidfm/asyncigo"
)

func main() {
	_ = asyncigo.NewEventLoop().Run(context.Background(), func(ctx context.Context) error {
		shielded := asyncigo.NewFuture[any]()
		task1 := asyncigo.SpawnTask(ctx, func(ctx context.Context) (any, error) {
			fmt.Println("waiting for shielded...")
			return shielded.Shield().Await(ctx)
		})

		unshielded := asyncigo.NewFuture[any]()
		task2 := asyncigo.SpawnTask(ctx, func(ctx context.Context) (any, error) {
			fmt.Println("waiting for unshielded...")
			return unshielded.Await(ctx)
		})

		// yield to the event loop for one tick to initialise the tasks
		_ = asyncigo.RunningLoop(ctx).Yield(ctx, nil)

		task1.Cancel(nil)
		task2.Cancel(nil)

		fmt.Println("task1:", task1.Err())
		fmt.Println("task2:", task2.Err())
		fmt.Println("shielded:", shielded.Err())
		fmt.Println("unshielded:", unshielded.Err())
		return nil
	})
}

Output:

waiting for shielded...
waiting for unshielded...
task1: context canceled
task2: context canceled
shielded: <nil>
unshielded: context canceled

package main

import (
	"fmt"

	"github.com/arvidfm/asyncigo"
)

func main() {
	it := asyncigo.Iter(func(yield func(int) bool) {
		a, b := 0, 1
		for a < 20 {
			if !yield(a) {
				return
			}
			a, b = a+b, a
		}
	})

	fmt.Println(it.Collect())
}

Output:

[0 1 1 2 3 5 8 13]

package main

import (
	"fmt"

	"github.com/arvidfm/asyncigo"
)

func main() {
	it := asyncigo.MapIter(func(yield func(int, string) bool) {
		_ = yield(5, "a") &&
			yield(10, "b") &&
			yield(15, "c")
	})

	m := it.Collect()
	for k := range m {
		fmt.Printf("%d: %s\n", k, m[k])
	}
}

Output:

5: a
10: b
15: c

When a task has been cancelled, it will continue running, but any calls to [Awaitable.Await] will immediately return context.Canceled. It's the responsibility of the task to stop early when cancelled.

package main

import (
	"context"
	"fmt"

	"github.com/arvidfm/asyncigo"
)

func main() {
	_ = asyncigo.NewEventLoop().Run(context.Background(), func(ctx context.Context) error {
		futs := make([]asyncigo.Future[int], 10)
		task := asyncigo.SpawnTask(ctx, func(ctx context.Context) (int, error) {
			for i := range futs {
				result, err := futs[i].Await(ctx)
				fmt.Printf("%d: (%v, %v)\n", i, result, err)
			}
			return 0, nil
		})

		loop := asyncigo.RunningLoop(ctx)
		for i := range futs {
			if i == 5 {
				task.Cancel(nil)
			}

			_ = loop.Yield(ctx, nil)
			futs[i].SetResult(i, nil)
		}

		result, err := task.Await(ctx)
		fmt.Printf("task result: (%v, %v)", result, err)
		return nil
	})
}

Output:

0: (0, <nil>)
1: (1, <nil>)
2: (2, <nil>)
3: (3, <nil>)
4: (4, <nil>)
5: (0, context canceled)
6: (0, context canceled)
7: (0, context canceled)
8: (0, context canceled)
9: (0, context canceled)
task result: (0, context canceled)