rill

package module

v0.3.0 Latest Latest Go to latest Published: Jul 21, 2024 License: MIT Imports: 3 Imported by: 16

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/destel/rill

Links

Open Source Insights

README ¶

Rill

Rill (noun: a small stream) is a Go toolkit that offers a collection of easy-to-use functions for concurrency, streaming, batching and pipeline construction. It abstracts away the complexities of concurrency, removes boilerplate, provides a structured way to handle errors and allows developers to focus on core logic. Whether you need to perform a basic concurrent ForEach or construct a complex multi-stage processing pipeline, Rill has got you covered.

Key Features

Easy to Use: the complexity of managing goroutines, channels, wait groups, and atomics is abstracted away
Easy to Integrate: seamlessly integrates into existing projects without any setup or configuration
Concurrent: provides control over the level of concurrency for all operations
Error Handling: provides a structured way to handle errors in concurrent applications
Streaming: handles real-time data streams or large datasets with a minimal memory footprint
Modular: allows composing functions to create custom pipelines and higher-order operations
Batching: simplifies organizing and processing data in batches
Order Preservation: provides functions that maintain the original order of data during concurrent processing
Efficient Resource Use: ensures goroutine pool sizes and memory allocations are independent of input size
Generic: all operations are type-safe and can be used with any data type

Motivation

Rill is not an iterator library or a functional programming library, though it may look like one. It's a concurrency library built specifically for Go channels.

There is a consensus in the Go community that functional programming style operations like Map, Filter, ForEach and others are not idiomatic in Go, and that basic for-loops are better, faster, and more concise. This is true for slices, but for channels, the complexity can quickly escalate beyond a basic for-loop as requirements are added:

A basic for-range loop is sufficient to iterate over a channel
Adding concurrency requires goroutines and a WaitGroup
Adding error handling means replacing WaitGroup with ErrGroup
Controlling concurrency level is often done with a semaphore
If the above approach becomes a bottleneck, worker pools must be introduced and managed manually
For a multi-stage pipeline, everything must be manually managed at each stage, causing complexity to grow non-linearly
Features like batching or ordered fan-in require even more complex orchestration and synchronization

The list can be continued. And while tools like channels, ErrGroups or semaphores are powerful on their own, combining them into a more complex logic, can lead to code with lots of boilerplate that's difficult to write, read, and maintain.

Rill was born out of the desire to remove code duplication and to encapsulate all this complexity in a library with a simple, composable, and expressive API. The introduction of generics in Go 1.18 opened the door to creating functional-style operations on channels, providing a natural way to achieve this goal.

Example Usage

Consider an example application that loads users from an API concurrently, updates their status to active and saves them back, while controlling the level of concurrency for each operation.

Full runnable example

func main() {
	// In case of early exit this will cancel the user fetching,
	// which in turn will terminate the entire pipeline.	
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Start with a stream of user ids
	ids := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Read users from the API.
	// Concurrency = 3
	users := rill.Map(ids, 3, func(id int) (*User, error) {
		return getUser(ctx, id)
	})

	// Activate users.
	// Concurrency = 2
	err := rill.ForEach(users, 2, func(u *User) error {
		if u.IsActive {
			fmt.Printf("User %d is already active\n", u.ID)
			return nil
		}

		u.IsActive = true
		return saveUser(ctx, u)
	})

	fmt.Println("Error:", err)
}

Installation

go get -u github.com/destel/rill

Testing Strategy

Rill has a test coverage of over 95%, with testing focused on:

Correctness: ensuring that functions produce accurate results at different levels of concurrency
Concurrency: confirming that correct number of goroutines is spawned and utilized
Ordering: ensuring that ordered versions of functions preserve the order, while basic versions do not

Design Philosophy

At the heart of rill lies a simple yet powerful concept: operating on channels of wrapped values, encapsulated by the Try container. This allows to propagate both values and errors through the pipeline, ensuring that errors are handled correctly at each stage. Such wrapped channels can be created manually or through utilities like FromSlice or FromChan, and then transformed via non-blocking functions like Map or Filter. Finally, the transformed stream can be consumed by a blocking function such as ForEach, Reduce or MapReduce

One of the key features of Rill is the ability to control the level of concurrency for almost all operations through the n parameter. This is possible due to the channel and goroutine orchestration that library does under the hood. Rill's built-in functions manage worker pools internally, making the number of goroutines and allocations independent of the input size.

Finally, rill is designed to be modular and extensible. Most functions take streams as input and return transformed streams as output. It's easy to create custom reusable higher-order operations and pipelines by combining existing ones.

Batching

Batching is a common pattern in concurrent processing, especially when dealing with external services or databases. Rill provides a Batch function that transforms a stream of items into a stream of batches of a specified size. It's also possible to specify a timeout, after which the batch is emitted even if it's not full. This is useful for keeping an application reactive when input stream is slow or sparse.

Consider a modification of the previous example, where list of ids is streamed from a remote file, and users are fetched from the API in batches.

Full runnable example

func main() {
	// In case of early exit this will cancel the file streaming,
	// which in turn will terminate the entire pipeline.
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Stream a file with user ids as an io.Reader
	reader, err := downloadFile(ctx, "http://example.com/user_ids1.txt")
	if err != nil {
		fmt.Println("Error:", err)
		return
	}

	// Transform the reader into a stream of lines
	lines := streamLines(reader)

	// Parse lines as integers
	// Concurrency = 3
	ids := rill.Map(lines, 3, func(line string) (int, error) {
		return strconv.Atoi(line)
	})

	// Group IDs into batches of 5 for bulk processing
	idBatches := rill.Batch(ids, 5, 1*time.Second)

	// Fetch users for each batch of IDs
	// Concurrency = 3
	userBatches := rill.Map(idBatches, 3, func(ids []int) ([]*User, error) {
		return getUsers(ctx, ids...)
	})

	// Transform batches back into a stream of users
	users := rill.Unbatch(userBatches)

	// Activate users.
	// Concurrency = 2
	err = rill.ForEach(users, 2, func(u *User) error {
		if u.IsActive {
			fmt.Printf("User %d is already active\n", u.ID)
			return nil
		}

		u.IsActive = true
		return saveUser(ctx, u)
	})

	fmt.Println("Error:", err)	
}

Fan-in and Fan-out

Go channels support both Fan-in and Fan-out patterns, meaning that multiple goroutines can write to a single channel (fan-in) or read from a single channel (fan-out). On top of that Rill adds a Merge function that can be used to combine multiple streams into a single one.

Consider a basic example application that concurrently sends messages through multiple servers, then collects the results into a single stream and handles errors.

Full runnable example

func main() {
	messages := rill.FromSlice([]string{
		"message1", "message2", "message3", "message4", "message5",
		"message6", "message7", "message8", "message9", "message10",
	}, nil)

	// Fan-out the messages to three servers
	results1 := rill.Map(messages, 2, func(message string) (string, error) {
		return message, sendMessage(message, "server1")
	})

	results2 := rill.Map(messages, 2, func(message string) (string, error) {
		return message, sendMessage(message, "server2")
	})

	results3 := rill.Map(messages, 2, func(message string) (string, error) {
		return message, sendMessage(message, "server3")
	})

	// Fan-in the results from all servers into a single stream
	results := rill.Merge(results1, results2, results3)

	// Handle errors
	err := rill.Err(results)
	fmt.Println("Error:", err)
}

Errors, Termination and Contexts

Error handling can be non-trivial in concurrent applications. Rill simplifies this by providing a structured error handling approach. Usually rill pipelines consist of zero or more non-blocking stages that transform the input stream, and one blocking stage that returns the results. General rule is: any error happening anywhere in the pipeline is propagated down to the final stage, where it's caught by some blocking function and returned to the caller.

Rill provides several blocking functions out of the box:

ForEach: Concurrently applies a user function to each item in the stream. Example
ToSlice: Collects all stream items into a slice. Example
Reduce: Concurrently reduces the stream to a single value, using a user provided reducer function. Example
MapReduce: Performs a concurrent MapReduce operation one the stream, reducing it to Go map, using user provided mapper and reducer functions. Example
All: Concurrently checks if all items in the stream satisfy a user provided condition. Example
Any: Concurrently checks if at least one item in the stream satisfies a user provided condition. Example
First: Returns the first item or error encountered in the stream. Example
Err: Checks if there's an error somewhere in the stream an returns it. Example

All blocking functions share a common behavior. In case of an early termination (before reaching the end of the input stream), such functions initiate background draining of the remaining items. This is done to prevent goroutine leaks by ensuring that all goroutines feeding the stream are allowed to complete. The input stream should not be used anymore after calling a blocking function.

It also possible to use a for-range loop instead of a blocking function to consume the stream. In this case, the caller would be responsible for draining the stream in case of an early termination. See more details in the package documentation.

Rill is context-agnostic, meaning that it does not enforce any specific context usage. However, it's recommended to make user-defined pipeline stages context-aware. This is especially important for the initial stage, as it allows to finish background draining process, described above, faster.

In the example below the printOddSquares function initiates a pipeline that depends on a context. When an error occurs in one of the pipeline stages, it propagates down the pipeline, causing an early return, context cancellation (via defer) and resource cleanup.

Full runnable example

func main() {
	ctx := context.Background()

	err := printOddSquares(ctx)
	fmt.Println("Error:", err)
}

func printOddSquares(ctx context.Context) error {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	numbers := infiniteNumberStream(ctx)

	odds := rill.Filter(numbers, 3, func(x int) (bool, error) {
		if x == 20 {
			return false, fmt.Errorf("early exit")
		}
		return x%2 == 1, nil
	})

	return rill.ForEach(odds, 3, func(x int) error {
		fmt.Println(x * x)
		return nil
	})
}

Order Preservation

In concurrent applications, maintaining the original sequence of processed items is challenging due to the nature of parallel execution. When values are read from an input stream, concurrently processed through a function f, and written to an output stream, their order might not match the order of the input. To address this, rill provides ordered versions of its core functions, such as OrderedMap or OrderedFilter. These ensure that if value x precedes value y in the input channel, then f(x) will precede f(y) in the output, preserving the original order. It's important to note that these ordered functions have a small overhead compared to their unordered counterparts, due to more advanced orchestration and synchronization happening under the hood.

Order preservation is vital in scenarios where the sequence of data impacts the outcome, such as time-series data processing. Take, for instance, an application that retrieves daily temperature measurements over a specific period and calculates the change in temperature from one day to the next. Such application can benefit from concurrent data fetching, but need fetched data to be processed in the correct order.

Full runnable example

type Measurement struct {
    Date time.Time
    Temp float64
}

func main() {
	city := "New York"
	endDate := time.Now()
	startDate := endDate.AddDate(0, 0, -30)

	// Create a stream of all days between startDate and endDate
	days := make(chan rill.Try[time.Time])
	go func() {
		defer close(days)
		for date := startDate; date.Before(endDate); date = date.AddDate(0, 0, 1) {
			days <- rill.Wrap(date, nil)
		}
	}()

	// Fetch the temperature for each day from the API
	// Concurrency = 10; Ordered
	measurements := rill.OrderedMap(days, 10, func(date time.Time) (Measurement, error) {
		temp, err := getTemperature(city, date)
		return Measurement{Date: date, Temp: temp}, err
	})

	// Iterate over the measurements, calculate and print changes.
	// Concurrency = 1; Ordered
	prev := Measurement{Temp: math.NaN()}
	err := rill.ForEach(measurements, 1, func(m Measurement) error {
		change := m.Temp - prev.Temp
		prev = m

		fmt.Printf("%s: %.1f°C (change %+.1f°C)\n", m.Date.Format("2006-01-02"), m.Temp, change)
		return nil
	})

	fmt.Println("Error:", err)
}

Performance

Rill's performance is primarily bounded by the performance of channel operations. As a result, applications that already use channels and channel-based concurrency patterns can expect minimal overhead when adopting Rill. Additionally, Rill does fewer allocations compared to some traditional concurrency patterns that spawn a goroutine for each channel item and use a semaphore to control the level of concurrency. For more details, refer to the benchmarks.

This makes Rill well-suited for a wide range of tasks, especially I/O-bound workloads where the overhead of channel operations is typically negligible.

Documentation ¶

Overview ¶

Package rill is a collection of easy-to-use functions for concurrency, streaming, batching and pipeline construction. It abstracts away the complexities of concurrency, removes boilerplate, and provides a structured way to handle errors. Rill is modular and can be easily integrated into existing projects: it requires no setup and allows using only the necessary functions. At the same time, rill's functions can be composed into complex, concurrent, and reusable pipelines when needed.

Streams and Try Containers ¶

In this package, a stream refers to a channel of Try containers. A Try container is a simple struct that holds a value and an error. When an "empty stream" is referred to, it means a channel of Try containers that has been closed and was never written to.

Most functions in this package are concurrent, and the level of concurrency can be controlled by the argument n. Some functions share common behaviors and characteristics, which are described below.

Non-blocking functions ¶

Functions such as Map, Filter, and Batch take a stream as an input and return a new stream as an output. They do not block and return the output stream immediately. All the processing is done in the background by the goroutine pools they spawn. These functions forward all errors from the input stream to the output stream. Any errors returned by the user-provided functions are also sent to the output stream. When such function reaches the end of the input stream, it closes the output stream, stops processing and cleans up resources.

Such functions are designed to be composed together to build complex processing pipelines:

stage2 := rill.Map(input, ...)
stage3 := rill.Batch(stage2, ...)
stage4 := rill.Map(stage3, ...)
results := rill.Unbatch(stage4, ...)
// consume the results and handle errors with some blocking function

Blocking functions ¶

Functions such as ForEach, Reduce and MapReduce are used at the last stage of the pipeline to consume the stream and return the final result or error.

Usually, these functions block until one of the following conditions is met:

The end of the stream is reached. In this case, the function returns the final result.
An error is encountered either in the input stream or in some user-provided function. In this case, the function returns the error.

In case of an early termination (before reaching the end of the input stream), such functions initiate background draining of the remaining items. This is done to prevent goroutine leaks by ensuring that all goroutines feeding the stream are allowed to complete. The input stream should not be used anymore after calling such functions.

It's also possible to consume the pipeline results manually, for example using a for-range loop. In this case, add a deferred call to DrainNB before the loop to ensure that goroutines are not leaked.

 defer rill.DrainNB(results)

 for res := range results {
		if res.Error != nil {
			return res.Error
		}
		// process res.Value
 }

Unordered functions ¶

Functions such as Map, Filter and FlatMap write items to the output stream as soon as they become available. Due to the concurrent nature of these functions, the order of items in the output stream may not match the order of items in the input stream. These functions prioritize performance and concurrency over maintaining the original order.

Ordered functions ¶

Functions such as OrderedMap or OrderedFilter preserve the order of items from the input stream. These functions are still concurrent, but use special synchronization techniques to ensure that items are written to the output stream in the same order as they were read from the input stream. This additional synchronization has some overhead, but it is negligible for i/o bound workloads.

Some other functions, such as ToSlice, Batch or First are not concurrent and are ordered by nature.

Error handling ¶

Error handling can be non-trivial in concurrent applications. Rill simplifies this by providing a structured error handling approach. As described above, all errors are automatically propagated down the pipeline to the final stage, where they can be caught. This allows the pipeline to terminate after the first error is encountered and return it to the caller.

In cases where more complex error handling logic is required, the Catch function can be used. It allows to catch and handle errors at any point in the pipeline, providing the flexibility to handle not only the first error, but any of them.

Example ¶

This example demonstrates a Rill pipeline that fetches users from an API, and updates their status to active and saves them back. Both operations are done concurrently.

package main

import (
	"context"
	"errors"
	"fmt"
	"hash/fnv"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

type User struct {
	ID       int
	Username string
	IsActive bool
}

func main() {
	// In case of early exit this will cancel the user fetching,
	// which in turn will terminate the entire pipeline.
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Start with a stream of user ids
	ids := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Read users from the API.
	// Concurrency = 3
	users := rill.Map(ids, 3, func(id int) (*User, error) {
		return getUser(ctx, id)
	})

	// Activate users.
	// Concurrency = 2
	err := rill.ForEach(users, 2, func(u *User) error {
		if u.IsActive {
			fmt.Printf("User %d is already active\n", u.ID)
			return nil
		}

		u.IsActive = true
		return saveUser(ctx, u)
	})

	fmt.Println("Error:", err)
}

var adjs = []string{"big", "small", "fast", "slow", "smart", "happy", "sad", "funny", "serious", "angry"}
var nouns = []string{"dog", "cat", "bird", "fish", "mouse", "elephant", "lion", "tiger", "bear", "wolf"}

// getUsers simulates fetching multiple users from an API.
// User fields are pseudo-random, but deterministic based on the user ID.
func getUsers(ctx context.Context, ids ...int) ([]*User, error) {
	randomSleep(1000 * time.Millisecond)

	users := make([]*User, 0, len(ids))
	for _, id := range ids {
		if err := ctx.Err(); err != nil {
			return nil, err
		}

		user := User{
			ID:       id,
			Username: adjs[hash(id, "adj")%len(adjs)] + "_" + nouns[hash(id, "noun")%len(nouns)],
			IsActive: hash(id, "active")%100 < 60,
		}

		users = append(users, &user)
	}
	return users, nil
}

var ErrUserNotFound = errors.New("user not found")

// getUser simulates fetching a user from an API.
func getUser(ctx context.Context, id int) (*User, error) {
	users, err := getUsers(ctx, id)
	if err != nil {
		return nil, err
	}

	if len(users) == 0 {
		return nil, ErrUserNotFound
	}

	return users[0], nil
}

// saveUser simulates saving a user through an API.
func saveUser(ctx context.Context, user *User) error {
	randomSleep(1000 * time.Millisecond)

	if err := ctx.Err(); err != nil {
		return err
	}

	if user.Username == "" {
		return fmt.Errorf("empty username")
	}

	fmt.Printf("User saved: %+v\n", user)
	return nil
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

// hash is a simple hash function that returns an integer hash for a given input.
func hash(input ...any) int {
	hasher := fnv.New32()
	fmt.Fprintln(hasher, input...)
	return int(hasher.Sum32())
}

Output:

Example (Batching) ¶

This example showcases the use of Rill for building a multi-stage data processing pipeline, with a focus on batch processing. It streams user ids from a remote file, then fetches users from an API in batches, updates their status to active, and saves them back. All operations are done concurrently.

package main

import (
	"bufio"
	"context"
	"errors"
	"fmt"
	"hash/fnv"
	"io"
	"math/rand"
	"strconv"
	"strings"
	"time"

	"github.com/destel/rill"
)

type User struct {
	ID       int
	Username string
	IsActive bool
}

func main() {
	// In case of early exit this will cancel the file streaming,
	// which in turn will terminate the entire pipeline.
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Stream a file with user ids as an io.Reader
	reader, err := downloadFile(ctx, "http://example.com/user_ids1.txt")
	if err != nil {
		fmt.Println("Error:", err)
		return
	}

	// Transform the reader into a stream of words
	lines := streamLines(reader)

	// Parse lines as integers
	// Concurrency = 3
	ids := rill.Map(lines, 3, func(line string) (int, error) {
		return strconv.Atoi(line)
	})

	// Group IDs into batches of 5 for bulk processing
	idBatches := rill.Batch(ids, 5, 1*time.Second)

	// Fetch users for each batch of IDs
	// Concurrency = 3
	userBatches := rill.Map(idBatches, 3, func(ids []int) ([]*User, error) {
		return getUsers(ctx, ids...)
	})

	// Transform batches back into a stream of users
	users := rill.Unbatch(userBatches)

	// Activate users.
	// Concurrency = 2
	err = rill.ForEach(users, 2, func(u *User) error {
		if u.IsActive {
			fmt.Printf("User %d is already active\n", u.ID)
			return nil
		}

		u.IsActive = true
		return saveUser(ctx, u)
	})

	fmt.Println("Error:", err)
}

// streamLines converts an io.Reader into a stream of lines
func streamLines(r io.ReadCloser) <-chan rill.Try[string] {
	out := make(chan rill.Try[string])
	go func() {
		defer r.Close()
		defer close(out)

		scanner := bufio.NewScanner(r)
		for scanner.Scan() {
			out <- rill.Wrap(scanner.Text(), nil)
		}
		if err := scanner.Err(); err != nil {
			out <- rill.Wrap("", err)
		}
	}()
	return out
}

var ErrFileNotFound = errors.New("file not found")

var files = map[string]string{
	"http://example.com/user_ids1.txt": strings.ReplaceAll("1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20", " ", "\n"),
	"http://example.com/user_ids2.txt": strings.ReplaceAll("21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40", " ", "\n"),
	"http://example.com/user_ids3.txt": strings.ReplaceAll("41 42 43 44 45", " ", "\n"),
	"http://example.com/text1.txt":     "Early morning brings early birds to the early market. Birds sing, the market buzzes, and the morning shines.",
	"http://example.com/text2.txt":     "The birds often sing at the market",
	"http://example.com/text3.txt":     "The market closes, the birds rest, and the night brings peace to the town.",
}

// downloadFile simulates downloading a file from a URL.
// Returns a reader for the file content.
func downloadFile(ctx context.Context, url string) (io.ReadCloser, error) {
	content, ok := files[url]
	if !ok {
		return nil, ErrFileNotFound
	}

	return io.NopCloser(strings.NewReader(content)), nil
}

var adjs = []string{"big", "small", "fast", "slow", "smart", "happy", "sad", "funny", "serious", "angry"}
var nouns = []string{"dog", "cat", "bird", "fish", "mouse", "elephant", "lion", "tiger", "bear", "wolf"}

// getUsers simulates fetching multiple users from an API.
// User fields are pseudo-random, but deterministic based on the user ID.
func getUsers(ctx context.Context, ids ...int) ([]*User, error) {
	randomSleep(1000 * time.Millisecond)

	users := make([]*User, 0, len(ids))
	for _, id := range ids {
		if err := ctx.Err(); err != nil {
			return nil, err
		}

		user := User{
			ID:       id,
			Username: adjs[hash(id, "adj")%len(adjs)] + "_" + nouns[hash(id, "noun")%len(nouns)],
			IsActive: hash(id, "active")%100 < 60,
		}

		users = append(users, &user)
	}
	return users, nil
}

// saveUser simulates saving a user through an API.
func saveUser(ctx context.Context, user *User) error {
	randomSleep(1000 * time.Millisecond)

	if err := ctx.Err(); err != nil {
		return err
	}

	if user.Username == "" {
		return fmt.Errorf("empty username")
	}

	fmt.Printf("User saved: %+v\n", user)
	return nil
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

// hash is a simple hash function that returns an integer hash for a given input.
func hash(input ...any) int {
	hasher := fnv.New32()
	fmt.Fprintln(hasher, input...)
	return int(hasher.Sum32())
}

Output:

Example (Context) ¶

This example demonstrates how to use context cancellation to terminate a Rill pipeline in case of an early exit. The printOddSquares function initiates a pipeline that prints squares of odd numbers. The infiniteNumberStream function is the initial stage of the pipeline. It generates numbers indefinitely until the context is canceled. When an error occurs in one of the pipeline stages:

The error is propagated down the pipeline and reaches the ForEach stage.
The ForEach function returns the error.
The printOddSquares function returns, and the context is canceled using defer.
The infiniteNumberStream function terminates due to context cancellation.
The entire pipeline is cleaned up gracefully.

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/destel/rill"
)

func main() {
	ctx := context.Background()

	err := printOddSquares(ctx)
	fmt.Println("Error:", err)

	// Wait one more second to see "infiniteNumberStream terminated" printed
	time.Sleep(1 * time.Second)
}

func printOddSquares(ctx context.Context) error {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	numbers := infiniteNumberStream(ctx)

	odds := rill.Filter(numbers, 3, func(x int) (bool, error) {
		if x == 20 {
			return false, fmt.Errorf("early exit")
		}
		return x%2 == 1, nil
	})

	return rill.ForEach(odds, 3, func(x int) error {
		fmt.Println(x * x)
		return nil
	})
}

func infiniteNumberStream(ctx context.Context) <-chan rill.Try[int] {
	out := make(chan rill.Try[int])
	go func() {
		defer fmt.Println("infiniteNumberStream terminated")
		defer close(out)
		for i := 1; ; i++ {
			if err := ctx.Err(); err != nil {
				return
			}
			out <- rill.Wrap(i, nil)
			time.Sleep(100 * time.Millisecond)
		}
	}()
	return out
}

Output:

Example (FanIn_FanOut) ¶

This example demonstrates how to use the Fan-in and Fan-out patterns to send messages through multiple servers concurrently.

package main

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

func main() {
	messages := rill.FromSlice([]string{
		"message1", "message2", "message3", "message4", "message5",
		"message6", "message7", "message8", "message9", "message10",
	}, nil)

	// Fan-out the messages to three servers
	results1 := rill.Map(messages, 2, func(message string) (string, error) {
		return message, sendMessage(message, "server1")
	})

	results2 := rill.Map(messages, 2, func(message string) (string, error) {
		return message, sendMessage(message, "server2")
	})

	results3 := rill.Map(messages, 2, func(message string) (string, error) {
		return message, sendMessage(message, "server3")
	})

	// Fan-in the results from all servers into a single stream
	results := rill.Merge(results1, results2, results3)

	// Handle errors
	err := rill.Err(results)
	fmt.Println("Error:", err)
}

// Helper function that simulates sending a message through a server
func sendMessage(message string, server string) error {
	randomSleep(1000 * time.Millisecond)
	fmt.Printf("Sent through %s: %s\n", server, message)
	return nil
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

Example (MapReduce) ¶

This example demonstrates a concurrent MapReduce performed on a set of remote files. It downloads them and calculates how many times each word appears in all the files.

package main

import (
	"bufio"
	"context"
	"errors"
	"fmt"
	"io"
	"strings"

	"github.com/destel/rill"
)

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

	defer cancel()

	// Start with a stream of file URLs
	urls := rill.FromSlice([]string{
		"http://example.com/text1.txt",
		"http://example.com/text2.txt",
		"http://example.com/text3.txt",
	}, nil)

	// Download files concurrently, and get a stream of all words from all files
	// Concurrency = 2
	words := rill.FlatMap(urls, 2, func(url string) <-chan rill.Try[string] {
		reader, err := downloadFile(ctx, url)
		if err != nil {
			return rill.FromSlice[string](nil, err) // Wrap the error in a stream
		}

		return streamWords(reader)
	})

	// Count the number of occurrences of each word
	counts, err := rill.MapReduce(words,
		// Map phase: Use the word as key and "1" as value
		// Concurrency = 3
		3, func(word string) (string, int, error) {
			return strings.ToLower(word), 1, nil
		},
		// Reduce phase: Sum all "1" values for the same key
		// Concurrency = 2
		2, func(x, y int) (int, error) {
			return x + y, nil
		},
	)

	fmt.Println("Result:", counts)
	fmt.Println("Error:", err)
}

// streamWords is helper function that converts an io.Reader into a stream of words.
func streamWords(r io.ReadCloser) <-chan rill.Try[string] {
	words := make(chan rill.Try[string], 1)

	go func() {
		defer r.Close()
		defer close(words)

		scanner := bufio.NewScanner(r)
		scanner.Split(bufio.ScanWords)

		for scanner.Scan() {
			word := scanner.Text()
			word = strings.Trim(word, ".,;:!?&()")
			if len(word) > 0 {
				words <- rill.Wrap(word, nil)
			}
		}
		if err := scanner.Err(); err != nil {
			words <- rill.Wrap("", err)
		}
	}()

	return words
}

var ErrFileNotFound = errors.New("file not found")

var files = map[string]string{
	"http://example.com/user_ids1.txt": strings.ReplaceAll("1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20", " ", "\n"),
	"http://example.com/user_ids2.txt": strings.ReplaceAll("21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40", " ", "\n"),
	"http://example.com/user_ids3.txt": strings.ReplaceAll("41 42 43 44 45", " ", "\n"),
	"http://example.com/text1.txt":     "Early morning brings early birds to the early market. Birds sing, the market buzzes, and the morning shines.",
	"http://example.com/text2.txt":     "The birds often sing at the market",
	"http://example.com/text3.txt":     "The market closes, the birds rest, and the night brings peace to the town.",
}

// downloadFile simulates downloading a file from a URL.
// Returns a reader for the file content.
func downloadFile(ctx context.Context, url string) (io.ReadCloser, error) {
	content, ok := files[url]
	if !ok {
		return nil, ErrFileNotFound
	}

	return io.NopCloser(strings.NewReader(content)), nil
}

Output:

Example (Ordering) ¶

This example demonstrates how OrderedMap can be used to enforce ordering of processing results. Pipeline below fetches temperature measurements for a city and calculates daily temperature changes. Measurements are fetched concurrently, but ordered processing is used to calculate the changes.

package main

import (
	"fmt"
	"hash/fnv"
	"math"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

type Measurement struct {
	Date time.Time
	Temp float64
}

func main() {
	city := "New York"
	endDate := time.Now()
	startDate := endDate.AddDate(0, 0, -30)

	// Create a stream of all days between startDate and endDate
	days := make(chan rill.Try[time.Time])
	go func() {
		defer close(days)
		for date := startDate; date.Before(endDate); date = date.AddDate(0, 0, 1) {
			days <- rill.Wrap(date, nil)
		}
	}()

	// Fetch the temperature for each day from the API
	// Concurrency = 10; Ordered
	measurements := rill.OrderedMap(days, 10, func(date time.Time) (Measurement, error) {
		temp, err := getTemperature(city, date)
		return Measurement{Date: date, Temp: temp}, err
	})

	// Iterate over the measurements, calculate and print changes.
	// Concurrency = 1; Ordered
	prev := Measurement{Temp: math.NaN()}
	err := rill.ForEach(measurements, 1, func(m Measurement) error {
		change := m.Temp - prev.Temp
		prev = m

		fmt.Printf("%s: %.1f°C (change %+.1f°C)\n", m.Date.Format("2006-01-02"), m.Temp, change)
		return nil
	})

	fmt.Println("Error:", err)
}

// getTemperature simulates fetching a temperature reading for a city and date,
func getTemperature(city string, date time.Time) (float64, error) {
	randomSleep(1000 * time.Millisecond)

	cityHash := float64(hash(city))

	temp := 15 - 10*math.Sin(cityHash+float64(date.Unix()))

	return temp, nil
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

// hash is a simple hash function that returns an integer hash for a given input.
func hash(input ...any) int {
	hasher := fnv.New32()
	fmt.Fprintln(hasher, input...)
	return int(hasher.Sum32())
}

Output:

Index ¶

func All[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error)
func Any[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error)
func Batch[A any](in <-chan Try[A], n int, timeout time.Duration) <-chan Try[[]A]
func Buffer[A any](in <-chan A, n int) <-chan A
func Catch[A any](in <-chan Try[A], n int, f func(error) error) <-chan Try[A]
func Drain[A any](in <-chan A)
func DrainNB[A any](in <-chan A)
func Err[A any](in <-chan Try[A]) error
func Filter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[A]
func FilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, error)) <-chan Try[B]
func First[A any](in <-chan Try[A]) (value A, found bool, err error)
func FlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan Try[B]
func ForEach[A any](in <-chan Try[A], n int, f func(A) error) error
func FromChan[A any](values <-chan A, err error) <-chan Try[A]
func FromChans[A any](values <-chan A, errs <-chan error) <-chan Try[A]
func FromSlice[A any](slice []A, err error) <-chan Try[A]
func Map[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B]
func MapReduce[A any, K comparable, V any](in <-chan Try[A], nm int, mapper func(A) (K, V, error), nr int, ...) (map[K]V, error)
func Merge[A any](ins ...<-chan A) <-chan A
func OrderedCatch[A any](in <-chan Try[A], n int, f func(error) error) <-chan Try[A]
func OrderedFilter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[A]
func OrderedFilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, error)) <-chan Try[B]
func OrderedFlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan Try[B]
func OrderedMap[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B]
func OrderedSplit2[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (outTrue <-chan Try[A], outFalse <-chan Try[A])
func Reduce[A any](in <-chan Try[A], n int, f func(A, A) (A, error)) (result A, hasResult bool, err error)
func Split2[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (outTrue <-chan Try[A], outFalse <-chan Try[A])
func ToChans[A any](in <-chan Try[A]) (<-chan A, <-chan error)
func ToSlice[A any](in <-chan Try[A]) ([]A, error)
func Unbatch[A any](in <-chan Try[[]A]) <-chan Try[A]
type Try
- func Wrap[A any](value A, err error) Try[A]

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func All ¶ added in v0.2.0

func All[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error)

All checks if all items in the input stream satisfy the condition f. This function returns false as soon as it finds an item that does not satisfy the condition. Otherwise, it returns true, including the case when the stream was empty.

This is a blocking unordered function that processes items concurrently using n goroutines. When n = 1, processing becomes sequential, making the function ordered.

See the package documentation for more information on blocking unordered functions and error handling.

Example ¶

package main

import (
	"fmt"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Are all numbers even?
	// Concurrency = 3
	ok, err := rill.All(numbers, 3, func(x int) (bool, error) {
		return x%2 == 0, nil
	})

	fmt.Println("Result:", ok)
	fmt.Println("Error:", err)
}

Output:

func Any ¶ added in v0.2.0

func Any[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (bool, error)

Any checks if there is an item in the input stream that satisfies the condition f. This function returns true as soon as it finds such an item. Otherwise, it returns false.

Any is a blocking unordered function that processes items concurrently using n goroutines. When n = 1, processing becomes sequential, making the function ordered.

See the package documentation for more information on blocking unordered functions and error handling.

Example ¶

package main

import (
	"fmt"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Is there at least one even number?
	// Concurrency = 3
	ok, err := rill.Any(numbers, 3, func(x int) (bool, error) {
		return x%2 == 0, nil
	})

	fmt.Println("Result: ", ok)
	fmt.Println("Error: ", err)
}

Output:

func Batch ¶

func Batch[A any](in <-chan Try[A], n int, timeout time.Duration) <-chan Try[[]A]

Batch take a stream of items and returns a stream of batches based on a maximum size and a timeout.

A batch is emitted when one of the following conditions is met:

The batch reaches the size of n items
The time since the first item was added to the batch exceeds the timeout
The input stream is closed

This function never emits empty batches. To disable the timeout and emit batches only based on the size, set the timeout to -1. Setting the timeout to zero is not supported and will result in a panic

This is a non-blocking ordered function that processes items sequentially.

See the package documentation for more information on non-blocking ordered functions and error handling.

Example ¶

Also check out the package level examples to see Batch in action

package main

import (
	"fmt"
	"time"

	"github.com/destel/rill"
)

func main() {
	// New number is emitted every 50ms
	numbers := make(chan rill.Try[int])
	go func() {
		defer close(numbers)
		for i := 0; i < 50; i++ {
			numbers <- rill.Wrap(i, nil)
			time.Sleep(50 * time.Millisecond)
		}
	}()

	// Group numbers into batches of up to 5
	batches := rill.Batch(numbers, 5, 1*time.Second)

	printStream(batches)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

Output:

func Buffer ¶

func Buffer[A any](in <-chan A, n int) <-chan A

Buffer takes a channel of items and returns a buffered channel of exact same items in the same order. This can be useful for preventing write operations on the input channel from blocking, especially if subsequent stages in the processing pipeline are slow. Buffering allows up to n items to be held in memory before back pressure is applied to the upstream producer.

Typical usage of Buffer might look like this:

users := getUsers(ctx, companyID)
users = rill.Buffer(users, 100)
// Now work with the users channel as usual.
// Up to 100 users can be buffered if subsequent stages of the pipeline are slow.

func Catch ¶

func Catch[A any](in <-chan Try[A], n int, f func(error) error) <-chan Try[A]

Catch allows handling errors in the middle of a stream processing pipeline. Every error encountered in the input stream is passed to the function f for handling.

The outcome depends on the return value of f:

If f returns nil, the error is considered handled and filtered out from the output stream.
If f returns a non-nil error, the original error is replaced with the result of f.

This is a non-blocking unordered function that handles errors concurrently using n goroutines. An ordered version of this function, OrderedCatch, is also available.

See the package documentation for more information on non-blocking unordered functions and error handling.

Example ¶

package main

import (
	"errors"
	"fmt"
	"math/rand"
	"strconv"
	"time"

	"github.com/destel/rill"
)

func main() {
	strs := rill.FromSlice([]string{"1", "2", "3", "4", "5", "not a number 6", "7", "8", "9", "10"}, nil)

	// Convert strings to ints
	// Concurrency = 3; Unordered
	ids := rill.Map(strs, 3, func(s string) (int, error) {
		randomSleep(1000 * time.Millisecond) // simulate some additional work
		return strconv.Atoi(s)
	})

	// Catch and ignore number parsing errors
	// Concurrency = 2; Unordered
	ids = rill.Catch(ids, 2, func(err error) error {
		if errors.Is(err, strconv.ErrSyntax) {
			return nil // Ignore this error
		}
		return err
	})

	// No error will be printed
	printStream(ids)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func Drain ¶

func Drain[A any](in <-chan A)

Drain consumes and discards all items from an input channel, blocking until the channel is closed.

func DrainNB ¶

func DrainNB[A any](in <-chan A)

DrainNB is a non-blocking version of Drain. Is does draining in a separate goroutine.

func Err ¶ added in v0.2.0

func Err[A any](in <-chan Try[A]) error

Err returns the first error encountered in the input stream or nil if there were no errors.

This is a blocking ordered function that processes items sequentially. See the package documentation for more information on blocking ordered functions and error handling.

Example ¶

package main

import (
	"context"
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

type User struct {
	ID       int
	Username string
	IsActive bool
}

func main() {
	ctx := context.Background()

	users := rill.FromSlice([]*User{
		{ID: 1, Username: "foo"},
		{ID: 2, Username: "bar"},
		{ID: 3},
		{ID: 4, Username: "baz"},
		{ID: 5, Username: "qux"},
		{ID: 6, Username: "quux"},
	}, nil)

	// Save users. Use struct{} as a result type
	// Concurrency = 2; Unordered
	results := rill.Map(users, 2, func(user *User) (struct{}, error) {
		return struct{}{}, saveUser(ctx, user)
	})

	// We're interested only in side effects and errors from
	// the pipeline above
	err := rill.Err(results)
	fmt.Println("Error:", err)
}

// saveUser simulates saving a user through an API.
func saveUser(ctx context.Context, user *User) error {
	randomSleep(1000 * time.Millisecond)

	if err := ctx.Err(); err != nil {
		return err
	}

	if user.Username == "" {
		return fmt.Errorf("empty username")
	}

	fmt.Printf("User saved: %+v\n", user)
	return nil
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func Filter ¶

func Filter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[A]

Filter takes a stream of items of type A and filters them using a predicate function f. Returns a new stream of items that passed the filter.

This is a non-blocking unordered function that processes items concurrently using n goroutines. An ordered version of this function, OrderedFilter, is also available.

See the package documentation for more information on non-blocking unordered functions and error handling.

Example ¶

package main

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Keep only even numbers
	// Concurrency = 3; Unordered
	evens := rill.Filter(numbers, 3, func(x int) (bool, error) {
		randomSleep(1000 * time.Millisecond) // simulate some additional work
		return x%2 == 0, nil
	})

	printStream(evens)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func FilterMap ¶ added in v0.3.0

func FilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, error)) <-chan Try[B]

FilterMap takes a stream of items of type A, applies a function f that can filter and transform them into items of type B. Returns a new stream of transformed items that passed the filter. This operation is equivalent to a Filter followed by a Map.

This is a non-blocking unordered function that processes items concurrently using n goroutines. An ordered version of this function, OrderedFilterMap, is also available.

See the package documentation for more information on non-blocking unordered functions and error handling.

Example ¶

package main

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Keep only odd numbers and square them
	// Concurrency = 3; Unordered
	squares := rill.FilterMap(numbers, 3, func(x int) (int, bool, error) {
		if x%2 == 0 {
			return 0, false, nil
		}

		randomSleep(1000 * time.Millisecond) // simulate some additional work
		return x * x, true, nil
	})

	printStream(squares)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func First ¶ added in v0.2.0

func First[A any](in <-chan Try[A]) (value A, found bool, err error)

First returns the first item or error encountered in the input stream, whichever comes first. The found return flag is set to false if the stream was empty, otherwise it is set to true.

This is a blocking ordered function that processes items sequentially. See the package documentation for more information on blocking ordered functions and error handling.

Example ¶

package main

import (
	"fmt"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Keep only the numbers divisible by 4
	// Concurrency = 3; Ordered
	dvisibleBy4 := rill.OrderedFilter(numbers, 3, func(x int) (bool, error) {
		return x%4 == 0, nil
	})

	// Get the first number divisible by 4
	first, ok, err := rill.First(dvisibleBy4)

	fmt.Println("Result:", first, ok)
	fmt.Println("Error:", err)
}

Output:

func FlatMap ¶

func FlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan Try[B]

FlatMap takes a stream of items of type A and transforms each item into a new sub-stream of items of type B using a function f. Those sub-streams are then flattened into a single output stream, which is returned.

This is a non-blocking unordered function that processes items concurrently using n goroutines. An ordered version of this function, OrderedFlatMap, is also available.

See the package documentation for more information on non-blocking unordered functions and error handling.

Example ¶

package main

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5}, nil)

	// Replace each number with three strings
	// Concurrency = 3; Unordered
	result := rill.FlatMap(numbers, 3, func(x int) <-chan rill.Try[string] {
		randomSleep(1000 * time.Millisecond) // simulate some additional work

		return rill.FromSlice([]string{
			fmt.Sprintf("foo%d", x),
			fmt.Sprintf("bar%d", x),
			fmt.Sprintf("baz%d", x),
		}, nil)
	})

	printStream(result)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func ForEach ¶

func ForEach[A any](in <-chan Try[A], n int, f func(A) error) error

ForEach applies a function f to each item in an input stream.

This is a blocking unordered function that processes items concurrently using n goroutines. When n = 1, processing becomes sequential, making the function ordered and similar to a regular for-range loop.

See the package documentation for more information on blocking unordered functions and error handling.

Example ¶

package main

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Square and print each number
	// Concurrency = 3; Unordered
	err := rill.ForEach(numbers, 3, func(x int) error {
		randomSleep(1000 * time.Millisecond) // simulate some additional work

		y := x * x
		fmt.Println(y)
		return nil
	})

	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

Example (Ordered) ¶

There is no ordered version of the ForEach function. To achieve ordered processing, use concurrency set to 1. If you need a concurrent and ordered ForEach, then do all processing with the OrderedMap, and then use ForEach with concurrency set to 1 at the final stage.

package main

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Square each number.
	// Concurrency = 3; Ordered
	squares := rill.OrderedMap(numbers, 3, func(x int) (int, error) {
		randomSleep(1000 * time.Millisecond) // simulate some additional work
		return x * x, nil
	})

	// Print results.
	// Concurrency = 1; Ordered
	err := rill.ForEach(squares, 1, func(y int) error {
		fmt.Println(y)
		return nil
	})
	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func FromChan ¶

func FromChan[A any](values <-chan A, err error) <-chan Try[A]

FromChan converts a regular channel into a stream. Additionally, this function can take an error, that will be added to the output stream alongside the values. Either argument can be nil, in which case it is ignored. If both arguments are nil, the function returns nil.

Such function signature allows concise wrapping of functions that return a channel and an error:

stream := rill.FromChan(someFunc())

func FromChans ¶

func FromChans[A any](values <-chan A, errs <-chan error) <-chan Try[A]

FromChans converts a regular channel into a stream. Additionally, this function can take a channel of errors, which will be added to the output stream alongside the values. Either argument can be nil, in which case it is ignored. If both arguments are nil, the function returns nil.

Such function signature allows concise wrapping of functions that return two channels:

stream := rill.FromChans(someFunc())

func FromSlice ¶

func FromSlice[A any](slice []A, err error) <-chan Try[A]

FromSlice converts a slice into a stream. If err is not nil function returns a stream with a single error.

Such function signature allows concise wrapping of functions that return a slice and an error:

stream := rill.FromSlice(someFunc())

func Map ¶

func Map[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B]

Map takes a stream of items of type A and transforms them into items of type B using a function f. Returns a new stream of transformed items.

This is a non-blocking unordered function that processes items concurrently using n goroutines. An ordered version of this function, OrderedMap, is also available.

See the package documentation for more information on non-blocking unordered functions and error handling.

Example ¶

package main

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Square each number.
	// Concurrency = 3; Unordered
	squares := rill.Map(numbers, 3, func(x int) (int, error) {
		randomSleep(1000 * time.Millisecond) // simulate some additional work
		return x * x, nil
	})

	printStream(squares)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func MapReduce ¶ added in v0.2.0

func MapReduce[A any, K comparable, V any](in <-chan Try[A], nm int, mapper func(A) (K, V, error), nr int, reducer func(V, V) (V, error)) (map[K]V, error)

MapReduce transforms the input stream into a Go map using a mapper and a reducer functions. The transformation is performed in two concurrent phases.

The mapper function transforms each input item into a key-value pair.
The reducer function reduces values for the same key into a single value. This phase has the same semantics as the Reduce function, in particular the reducer function must be commutative and associative.

MapReduce is a blocking unordered function that processes items concurrently using nm and nr goroutines for the mapper and reducer functions respectively. Setting nr = 1 will make the reduce phase sequential and ordered, see Reduce for more information.

See the package documentation for more information on blocking unordered functions and error handling.

Example ¶

package main

import (
	"fmt"
	"regexp"
	"strings"

	"github.com/destel/rill"
)

func main() {
	var re = regexp.MustCompile(`\w+`)
	text := "Early morning brings early birds to the early market. Birds sing, the market buzzes, and the morning shines."

	// Start with a stream of words
	words := rill.FromSlice(re.FindAllString(text, -1), nil)

	// Count the number of occurrences of each word
	mr, err := rill.MapReduce(words,
		// Map phase: Use the word as key and "1" as value
		// Concurrency = 3
		3, func(word string) (string, int, error) {
			return strings.ToLower(word), 1, nil
		},
		// Reduce phase: Sum all "1" values for the same key
		// Concurrency = 2
		2, func(x, y int) (int, error) {
			return x + y, nil
		},
	)

	fmt.Println("Result:", mr)
	fmt.Println("Error:", err)
}

Output:

func Merge ¶

func Merge[A any](ins ...<-chan A) <-chan A

Merge performs a fan-in operation on the list of input channels, returning a single output channel. The resulting channel will contain all items from all inputs, and will be closed when all inputs are fully consumed.

This is a non-blocking function that processes items from each input sequentially.

See the package documentation for more information on non-blocking functions and error handling.

Example ¶

package main

import (
	"fmt"

	"github.com/destel/rill"
)

func main() {
	numbers1 := rill.FromSlice([]int{1, 2, 3, 4, 5}, nil)
	numbers2 := rill.FromSlice([]int{6, 7, 8, 9, 10}, nil)
	numbers3 := rill.FromSlice([]int{11, 12}, nil)

	numbers := rill.Merge(numbers1, numbers2, numbers3)

	printStream(numbers)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

Output:

func OrderedCatch ¶

func OrderedCatch[A any](in <-chan Try[A], n int, f func(error) error) <-chan Try[A]

OrderedCatch is the ordered version of Catch.

Example ¶

The same example as for the Catch, but using ordered versions of functions.

package main

import (
	"errors"
	"fmt"
	"math/rand"
	"strconv"
	"time"

	"github.com/destel/rill"
)

func main() {
	strs := rill.FromSlice([]string{"1", "2", "3", "4", "5", "not a number 6", "7", "8", "9", "10"}, nil)

	// Convert strings to ints
	// Concurrency = 3; Unordered
	ids := rill.OrderedMap(strs, 3, func(s string) (int, error) {
		randomSleep(1000 * time.Millisecond) // simulate some additional work
		return strconv.Atoi(s)
	})

	// Catch and ignore number parsing errors
	// Concurrency = 2; Unordered
	ids = rill.OrderedCatch(ids, 2, func(err error) error {
		if errors.Is(err, strconv.ErrSyntax) {
			return nil // Ignore this error
		}
		return err
	})

	// No error will be printed
	printStream(ids)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func OrderedFilter ¶

func OrderedFilter[A any](in <-chan Try[A], n int, f func(A) (bool, error)) <-chan Try[A]

OrderedFilter is the ordered version of Filter.

Example ¶

The same example as for the Filter, but using ordered versions of functions.

package main

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Keep only even numbers
	// Concurrency = 3; Ordered
	evens := rill.OrderedFilter(numbers, 3, func(x int) (bool, error) {
		randomSleep(1000 * time.Millisecond) // simulate some additional work
		return x%2 == 0, nil
	})

	printStream(evens)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func OrderedFilterMap ¶ added in v0.3.0

func OrderedFilterMap[A, B any](in <-chan Try[A], n int, f func(A) (B, bool, error)) <-chan Try[B]

OrderedFilterMap is the ordered version of FilterMap.

Example ¶

The same example as for the FilterMap, but using ordered versions of functions.

package main

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Keep only odd numbers and square them
	// Concurrency = 3; Ordered
	squares := rill.OrderedFilterMap(numbers, 3, func(x int) (int, bool, error) {
		if x%2 == 0 {
			return 0, false, nil
		}

		randomSleep(1000 * time.Millisecond) // simulate some additional work
		return x * x, true, nil
	})

	printStream(squares)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func OrderedFlatMap ¶

func OrderedFlatMap[A, B any](in <-chan Try[A], n int, f func(A) <-chan Try[B]) <-chan Try[B]

OrderedFlatMap is the ordered version of FlatMap.

Example ¶

The same example as for the FlatMap, but using ordered versions of functions.

package main

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5}, nil)

	// Replace each number with three strings
	// Concurrency = 3; Ordered
	result := rill.OrderedFlatMap(numbers, 3, func(x int) <-chan rill.Try[string] {
		randomSleep(1000 * time.Millisecond) // simulate some additional work

		return rill.FromSlice([]string{
			fmt.Sprintf("foo%d", x),
			fmt.Sprintf("bar%d", x),
			fmt.Sprintf("baz%d", x),
		}, nil)
	})

	printStream(result)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func OrderedMap ¶

func OrderedMap[A, B any](in <-chan Try[A], n int, f func(A) (B, error)) <-chan Try[B]

OrderedMap is the ordered version of Map.

Example ¶

The same example as for the Map, but using ordered versions of functions.

package main

import (
	"fmt"
	"math/rand"
	"time"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Square each number.
	// Concurrency = 3; Ordered
	squares := rill.OrderedMap(numbers, 3, func(x int) (int, error) {
		randomSleep(1000 * time.Millisecond) // simulate some additional work
		return x * x, nil
	})

	printStream(squares)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

func randomSleep(max time.Duration) {
	time.Sleep(time.Duration(rand.Intn(int(max))))
}

Output:

func OrderedSplit2 ¶

func OrderedSplit2[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (outTrue <-chan Try[A], outFalse <-chan Try[A])

OrderedSplit2 is the ordered version of Split2.

func Reduce ¶ added in v0.2.0

func Reduce[A any](in <-chan Try[A], n int, f func(A, A) (A, error)) (result A, hasResult bool, err error)

Reduce combines all items from the input stream into a single value using a binary function f. The function f is called for pairs of items, progressively reducing the stream contents until only one value remains.

As an unordered function, Reduce can apply f to any pair of items in any order, which requires f to be:

Associative: f(a, f(b, c)) == f(f(a, b), c)
Commutative: f(a, b) == f(b, a)

The hasResult return flag is set to false if the stream was empty, otherwise it is set to true.

Reduce is a blocking unordered function that processes items concurrently using n goroutines. The case when n = 1 is optimized: it does not spawn additional goroutines and processes items sequentially, making the function ordered. This also removes the need for the function f to be commutative.

See the package documentation for more information on blocking unordered functions and error handling.

Example ¶

package main

import (
	"fmt"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Sum all numbers
	sum, ok, err := rill.Reduce(numbers, 3, func(a, b int) (int, error) {
		return a + b, nil
	})

	fmt.Println("Result:", sum, ok)
	fmt.Println("Error:", err)
}

Output:

func Split2 ¶

func Split2[A any](in <-chan Try[A], n int, f func(A) (bool, error)) (outTrue <-chan Try[A], outFalse <-chan Try[A])

Split2 divides the input stream into two output streams based on the predicate function f: The splitting behavior is determined by the boolean return value of f. When f returns true, the item is sent to the outTrue stream, otherwise it is sent to the outFalse stream. In case of any error, the item is sent to one of the output streams in a non-deterministic way.

This is a non-blocking unordered function that processes items concurrently using n goroutines. An ordered version of this function, OrderedSplit2, is also available.

See the package documentation for more information on non-blocking unordered functions and error handling.

func ToChans ¶

func ToChans[A any](in <-chan Try[A]) (<-chan A, <-chan error)

ToChans splits an input stream into two channels: one for values and one for errors. It's an inverse of FromChans. Returns two nil channels if the input is nil.

func ToSlice ¶

func ToSlice[A any](in <-chan Try[A]) ([]A, error)

ToSlice converts an input stream into a slice.

This is a blocking ordered function that processes items sequentially. See the package documentation for more information on blocking ordered functions and error handling.

Example ¶

package main

import (
	"fmt"

	"github.com/destel/rill"
)

func main() {
	numbers := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Square each number
	// Concurrency = 3; Ordered
	squares := rill.OrderedMap(numbers, 3, func(x int) (int, error) {
		return x * x, nil
	})

	squaresSlice, err := rill.ToSlice(squares)

	fmt.Println("Result:", squaresSlice)
	fmt.Println("Error:", err)
}

Output:

func Unbatch ¶

func Unbatch[A any](in <-chan Try[[]A]) <-chan Try[A]

Unbatch is the inverse of Batch. It takes a stream of batches and returns a stream of individual items.

This is a non-blocking ordered function that processes items sequentially. See the package documentation for more information on non-blocking ordered functions and error handling.

Example ¶

package main

import (
	"fmt"

	"github.com/destel/rill"
)

func main() {
	batches := rill.FromSlice([][]int{
		{1, 2, 3},
		{4, 5},
		{6, 7, 8, 9},
		{10},
	}, nil)

	numbers := rill.Unbatch(batches)

	printStream(numbers)
}

// printStream prints all items from a stream (one per line) and an error if any.
func printStream[A any](stream <-chan rill.Try[A]) {
	fmt.Println("Result:")
	err := rill.ForEach(stream, 1, func(x A) error {
		fmt.Printf("%+v\n", x)
		return nil
	})
	fmt.Println("Error:", err)
}

Output:

Types ¶

type Try ¶

type Try[A any] struct {
	Value A
	Error error
}

Try is a container holding a value of type A or an error

func Wrap ¶

func Wrap[A any](value A, err error) Try[A]

Wrap converts a value and/or error into a Try container. It's a convenience function to avoid creating a Try container manually and benefit from type inference.

Such function signature also allows concise wrapping of functions that return a value and an error:

item := rill.Wrap(strconv.ParseInt("42"))

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
internal
core
ringbuffer
th Package th provides basic test helpers.	Package th provides basic test helpers.

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL

README ¶

Rill

Key Features

Motivation

Example Usage

Installation

Testing Strategy

Design Philosophy

Batching

Fan-in and Fan-out

Errors, Termination and Contexts

Order Preservation

Performance

Documentation ¶

Overview ¶

Streams and Try Containers ¶

Non-blocking functions ¶

Blocking functions ¶

Unordered functions ¶

Ordered functions ¶

Error handling ¶

Index ¶

Examples ¶

Constants ¶

Variables ¶

Functions ¶

func All ¶ added in v0.2.0

func Any ¶ added in v0.2.0

func Batch ¶

func Buffer ¶

func Catch ¶

func Drain ¶

func DrainNB ¶

func Err ¶ added in v0.2.0

func Filter ¶

func FilterMap ¶ added in v0.3.0

func First ¶ added in v0.2.0

func FlatMap ¶

func ForEach ¶

func FromChan ¶

func FromChans ¶

func FromSlice ¶

func Map ¶

func MapReduce ¶ added in v0.2.0

func Merge ¶

func OrderedCatch ¶

func OrderedFilter ¶

func OrderedFilterMap ¶ added in v0.3.0

func OrderedFlatMap ¶

func OrderedMap ¶

func OrderedSplit2 ¶

func Reduce ¶ added in v0.2.0

func Split2 ¶

func ToChans ¶

func ToSlice ¶

func Unbatch ¶

Types ¶

type Try ¶

func Wrap ¶

Source Files ¶

Directories ¶