An implementation of IO, Stream, Fiber for Golang, inspired by awesome Scala libraries cats and fs2

goio

A lightweight toolkit for data source extraction, transformation, and loading (ETL)

# go-etl
[![LICENSE][license-img]][license]
[![Language][lang-img]][lang]
[![Build][ci-img]][ci]
[![Go Report Card][report-img]][report]
[![GitHub release][release-img]][release]
[![GitHub release date][release-date-img]][release-date]
[![Coverage Status][cov-img]][cov]
[![GoDoc][doc-img]][doc]
[![Mentioned in Awesome Go][awesome-img]][awesome]
[![Docker Version][docker-img]][docker]

English | [简体中文](README_zh-CN.md)

go-etl is a toolset for extracting, transforming, and loading data sources, providing powerful data synchronization capabilities.

go-etl will provide the following ETL capabilities:

- The ability to extract and load data from mainstream databases is implemented in the storage package
- The ability to extract and load data from data streams in a two-dimensional table-like format is implemented in the stream package
- Similar data synchronization capabilities to datax, implemented in the datax package

Since I have limited energy, everyone is welcome to submit issues to discuss go-etl, let's make progress together!


## Data Synchronization Tool

This data synchronization tool has the synchronization capability for the following data sources.

| Type         | Data Source             | Reader | Writer| Document                                                        |
| ------------ | ------------------ | ------------ | ---------- | ------------------------------------------------------------ |
| Relational Database | MySQL/Mariadb/Tidb/TDSQL MySQL | √            | √          | [Read](datax/plugin/reader/mysql/README.md)、[Write](datax/plugin/writer/mysql/README.md) |
|              | Postgres/Greenplum | √            | √          | [Read](datax/plugin/reader/postgres/README.md)、[Write](datax/plugin/writer/postgres/README.md) |
|              | DB2 LUW            | √            | √          | [Read](datax/plugin/reader/db2/README.md)、[Write](datax/plugin/writer/db2/README.md) |
|              | SQL Server            | √            | √          | [Read](datax/plugin/reader/sqlserver/README.md)、[Write](datax/plugin/writer/sqlserver/README.md) |
|              | Oracle            | √            | √          | [Read](datax/plugin/reader/oracle/README.md)、[Write](datax/plugin/writer/oracle/README.md) |
|              | Sqlite3            | √            | √          | [Read](datax/plugin/reader/sqlite3/README.md)、[Write](datax/plugin/writer/sqlite3/README.md) |
|              | Dameng            | √            | √          | [Read](datax/plugin/reader/dm/README.md)、[Write](datax/plugin/writer/dm/README.md) |
| Unstructured Data Stream    | CSV                | √            | √          | [Read](datax/plugin/reader/csv/README.md)、[Write](datax/plugin/writer/csv/README.md) |
|              | XLSX（excel）      | √            | √          | [Read](datax/plugin/reader/xlsx/README.md)、[Write](datax/plugin/writer/xlsx/README.md) |

### Getting Started

#### Quick Start (3 Minutes)

##### Start from Binary Program

You can download the 64-bit binary executable for Windows or Linux operating systems from the [latest releases](https://github.com/Breeze0806/go-etl/releases).

Start data synchronization with the [go-etl Data Synchronization User Manual](README_USER.md).

##### Start from Docker Image

**Pull Docker Image**
```bash
docker pull go-etl:v0.2.3
```

**Start Container**
```bash
docker run -d -p 6080:6080 --name etl -v /data:/usr/local/go-etl/data go-etl:v0.2.3
```

**Enter Container**
```bash
docker exec -it etl bash
```

**Execute Sync in Container**
```bash
docker exec -it etl release/bin/go-etl -c data/config.json
```

#### Start from Performance Testing

If you want to directly obtain performance-related information, you can deploy according to the [Prometheus Monitoring Deployment Manual](docker/README.md) to acquire relevant performance metrics and even performance visualization charts.

#### Start from Source Code

##### Linux

###### Compilation Dependencies

1. golang 1.20 and later versions
2. gcc 4.8 and later versions

###### Build

```bash
cd ${GO_PATH}/src
git clone https://github.com/Breeze0806/go-etl.git "github.com/Breeze0806/go-etl"
cd github.com/Breeze0806/go-etl
make dependencies
make release
```

###### Removing DB2 Dependency

Before compilation, it is necessary to use `export IGNORE_PACKAGES=db2` 

```bash
export IGNORE_PACKAGES=db2
cd ${GO_PATH}/src
git clone https://github.com/Breeze0806/go-etl.git "github.com/Breeze0806/go-etl"
cd github.com/Breeze0806/go-etl
make dependencies
make release
```

##### Windows

###### Compilation Dependencies

1. A MinGW-w64 environment with GCC 7.2.0 or higher is required for compilation.
2. golang 1.20 and later versions
3. The minimum compilation environment is Windows 7.

###### Build

```bash
cd ${GO_PATH}\src
git clone https://github.com/Breeze0806/go-etl.git "github.com/Breeze0806/go-etl"
cd github.com/Breeze0806/go-etl
release.bat
```

###### Removing DB2 Dependency

Before compilation, it is necessary to use `set IGNORE_PACKAGES=db2`

```bash
cd ${GO_PATH}\src
git clone https://github.com/Breeze0806/go-etl.git "github.com/Breeze0806/go-etl"
cd github.com/Breeze0806/go-etl
set IGNORE_PACKAGES=db2
release.bat
```

##### Compilation Output

```
    +---datax---|---plugin---+---reader--mysql---|--README.md
    |                        | .......
    |                        |
    |                        |---writer--mysql---|--README.md
    |                        | .......
    |
    +---bin----go-etl
    +---exampales---+---csvpostgres----config.json
    |               |---db2------------config.json
    |               | .......
    |
    +---README_USER.md
```

+ The datax/plugin directory contains the documentation for various plugins.
+ The bin directory houses the data synchronization program, named go-etl.
+ The examples directory includes configuration files for data synchronization in different scenarios.
+ README_USER.md is the user manual or guide in English.

#### Start from Compiled Docker Image

Use the following commands to get the `go-etl` project (version `v0.2.3`):

```bash
git clone https://github.com/Breeze0806/go-etl.git
cd go-etl
git describe --abbrev=0 --tags
```

Build the Docker image with the following command:

```bash
docker build . -t go-etl:v0.2.3
```

Start the container:

```bash
docker run -d -p 6080:6080 --name etl -v /data:/usr/local/go-etl/data go-etl:v0.2.3
```

Enter the container:

```bash
docker exec -it etl bash
```

Note that currently, sqlite3, DB2, and Oracle are not directly supported and require downloading the corresponding ODBC and configuring environment variables.

#### Batch Sync

Use a wizard CSV file to batch sync multiple tables.

**1. Create data source config `config.json`** - same as single sync

**2. Create wizard file `wizard.csv`** - each row defines a source-target table pair:
```csv
source_table,target_table
table1,table1_copy
table2,table2_copy
```

**3. Generate batch configs and run script:**

**Linux:**
```bash
./go-etl -c config.json -w wizard.csv; ./run.sh
```

**Windows:**
```powershell
.\go-etl.exe -c config.json -w wizard.csv; run.bat
```

**Docker:**
```bash
docker exec -it etl release/bin/go-etl -c data/config.json -w data/wizard.csv; docker exec -it etl bash run.sh
```

### Data Synchronization Development Guide

Refer to the [go-etl Data Synchronization Developer Documentation](datax/README.md) to assist with your development.

## Module Introduction

### datax

This package provides an interface similar to Alibaba's [DataX](https://github.com/alibaba/DataX) to implement an offline data synchronization framework in Go.

```
readerPlugin(reader)—> Framework(Exchanger+Transformer) ->writerPlugin(writer)  
```

Built using a Framework + plugin architecture. Data source reading and writing are abstracted into Reader/Writer plugins and integrated into the overall synchronization framework.

+ Reader: The Reader is the data acquisition module, responsible for collecting data from the data source and sending it to the Framework.
+ Writer: The Writer is the data writing module, responsible for continuously fetching data from the Framework and writing it to the destination.
+ Framework: The Framework connects the reader and writer, serving as a data transmission channel, and handles core technical aspects such as buffering, flow control, concurrency, and data transformation.

For detailed information, please refer to the [go-etl Data Synchronization Developer Documentation](datax/README.md).

### element

The data types and data type conversions in go-etl have been implemented. For more information, please refer to the [go-etl Data Type Descriptions](element/README.md).

### storage

#### database

Basic integration for databases has been implemented, abstracting the database dialect (Dialect) interface. For specific implementation details, please refer to the [Database Storage Developer Guide](storage/database/README.md).

#### stream

Primarily used for parsing byte streams, such as files, message queues, Elasticsearch, etc. The byte stream format can be CSV, JSON, XML, etc.

##### file

Focused on file parsing, including CSV, Excel, etc. It abstracts the InputStream and OutputStream interfaces. For specific implementation details, refer to the [Developer Guide for Tabular File Storage](storage/stream/file/README.md).

### tools

A collection of utilities for compilation, adding licenses, etc.

#### datax

##### build

```bash
go generate ./...
```

Release command used to register developer-created reader and writer plugins into the program's code.

Additionally, this command inserts compilation information such as software version, Git version, Go compiler version, and compilation time into the command line.

##### plugin

A plugin template creation tool for data sources. It's used to create a new reader or writer template, in conjunction with the release command, to reduce the developer's workload.

##### release

A packaging tool for the data synchronization program and user documentation.

#### license

Automatically adds a license to Go code files and formats the code using `gofmt -s -w`.

```bash
go run tools/license/main.go
```

## Contributing

Contributions are welcome! Please read our [Contributing Guide](CONTRIBUTING.md) for details on how to contribute to this project.

### Ways to Contribute

- Report bugs and issues
- Suggest new features
- Submit pull requests
- Improve documentation
- Share your use cases

### Getting Help

- Check the [Documentation](README.md)
- Check the [User Manual](README_USER.md)
- Submit a GitHub Issue for discussion

[lang-img]:https://img.shields.io/badge/Language-Go-blue.svg
[lang]:https://golang.org/
[report-img]:https://goreportcard.com/badge/github.com/Breeze0806/go-etl
[report]:https://goreportcard.com/report/github.com/Breeze0806/go-etl
[doc-img]:https://godoc.org/github.com/Breeze0806/go-etl?status.svg
[doc]:https://godoc.org/github.com/Breeze0806/go-etl
[license-img]: https://img.shields.io/badge/License-Apache%202.0-blue.svg
[license]: https://github.com/Breeze0806/go-etl/blob/main/LICENSE
[ci-img]: https://github.com/Breeze0806/go-etl/actions/workflows/Build.yml/badge.svg
[ci]: https://github.com/Breeze0806/go-etl/actions/workflows/Build.yml
[release-img]: https://img.shields.io/github/tag/Breeze0806/go-etl.svg?label=release
[release]: https://github.com/Breeze0806/go-etl/releases
[release-date-img]: https://img.shields.io/github/release-date/Breeze0806/go-etl.svg
[release-date]: https://github.com/Breeze0806/go-etl/releases
[cov-img]: https://codecov.io/gh/Breeze0806/go-etl/branch/main/graph/badge.svg?token=UGb27Nysga
[cov]: https://codecov.io/gh/Breeze0806/go-etl
[awesome-img]:https://awesome.re/mentioned-badge.svg
[awesome]:https://github.com/avelino/awesome-go
[docker-img]:https://img.shields.io/docker/v/breeze0806/go-etl?sort=semver&logo=docker&logoColor=white&label=Docker&color=blue
[docker]:https://hub.docker.com/r/breeze0806/go-etl

go-etl

# go-streams
[![Build](https://github.com/reugn/go-streams/actions/workflows/build.yml/badge.svg)](https://github.com/reugn/go-streams/actions/workflows/build.yml)
[![PkgGoDev](https://pkg.go.dev/badge/github.com/reugn/go-streams)](https://pkg.go.dev/github.com/reugn/go-streams)
[![Go Report Card](https://goreportcard.com/badge/github.com/reugn/go-streams)](https://goreportcard.com/report/github.com/reugn/go-streams)
[![codecov](https://codecov.io/gh/reugn/go-streams/branch/master/graph/badge.svg)](https://codecov.io/gh/reugn/go-streams)

`go-streams` provides a lightweight and efficient stream processing framework for Go. Its concise DSL allows
for easy definition of declarative data pipelines using composable sources, flows, and sinks.

![pipeline-architecture-example](./docs/images/pipeline-architecture-example.png)

> [Wiki](https://en.wikipedia.org/wiki/Pipeline_(computing))  
> In computing, a pipeline, also known as a data pipeline, is a set of data processing elements connected in series,
> where the output of one element is the input of the next one. The elements of a pipeline are often executed in
> parallel or in time-sliced fashion. Some amount of buffer storage is often inserted between elements.

## Overview
The core module has no external dependencies and provides three key [components](./streams.go)
for constructing stream processing pipelines:

- **Source:** The entry point of a pipeline, emitting data into the stream. (One open output)
- **Flow:** A processing unit, transforming data as it moves through the pipeline. (One open input, one open output)
- **Sink:** The termination point of a pipeline, consuming processed data and often acting as a subscriber. (One
  open input)

### Flows
The [flow](flow) package provides a collection of `Flow` implementations for common stream
processing operations. These building blocks can be used to transform and manipulate data within pipelines.

- **Map:** Transforms each element in the stream.
- **FlatMap:** Transforms each element into a stream of slices of zero or more elements.
- **Filter:** Selects elements from the stream based on a condition.
- **Fold:** Combines elements of the stream with the last folded value and emits the new value.
  Requires an initial value.
- **Reduce:** Combines elements of the stream with the last reduced value and emits the new value.
  Does not require an initial value.
- **PassThrough:** Passes elements through unchanged.
- **Split<sup>1</sup>:** Divides the stream into two streams based on a boolean predicate.
- **FanOut<sup>1</sup>:** Duplicates the stream to multiple outputs for parallel processing.
- **RoundRobin<sup>1</sup>:** Distributes elements evenly across multiple outputs.
- **Merge<sup>1</sup>:** Combines multiple streams into a single stream.
- **ZipWith<sup>1</sup>:** Combines elements from multiple streams using a function.
- **Flatten<sup>1</sup>:** Flattens a stream of slices of elements into a stream of elements.
- **Batch:** Breaks a stream of elements into batches based on size or timing.
- **Throttler:** Limits the rate at which elements are processed.
- **AdaptiveThrottler:** Limits the rate at which elements are processed based on the current system resource utilization (CPU and memory usage).
- **SlidingWindow:** Creates overlapping windows of elements.
- **TumblingWindow:** Creates non-overlapping, fixed-size windows of elements.
- **SessionWindow:** Creates windows based on periods of activity and inactivity.
- **Keyed:** Groups elements by key for parallel processing of related data.

**<sup>1</sup>** Utility Flows

### Connectors
Standard `Source` and `Sink` implementations are located in the [extension](extension) package.

* Go channel inbound and outbound connector
* File inbound and outbound connector
* Standard Go `io.Reader` Source and `io.Writer` Sink connectors
* `os.Stdout` and `Discard` Sink connectors (useful for development and debugging)

The following connectors are available as separate modules and have their own dependencies.
* [Aerospike](https://www.aerospike.com/)
* [Apache Kafka](https://kafka.apache.org/)
* [Apache Pulsar](https://pulsar.apache.org/)
* AWS ([S3](https://aws.amazon.com/s3/))
* Azure ([Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs/))
* GCP ([Storage](https://cloud.google.com/storage/))
* [NATS](https://nats.io/)
* [Redis](https://redis.io/)
* [WebSocket](https://en.wikipedia.org/wiki/WebSocket)

## Usage Examples
See the [examples](examples) directory for practical code samples demonstrating how to build
complete stream processing pipelines, covering various use cases and integration scenarios.

## License
Licensed under the MIT License.


go-streams

Type-safe stream processing library inspired by the Java Streams API

# Go Streams API

[![Go Reference](https://pkg.go.dev/badge/github.com/mariomac/gostream.svg)](https://pkg.go.dev/github.com/mariomac/gostream)
[![Go Report Card](https://goreportcard.com/badge/github.com/mariomac/gostream)](https://goreportcard.com/report/github.com/mariomac/gostream)
[![codecov](https://codecov.io/github/mariomac/gostream/graph/badge.svg?token=D3C6Y3OCXZ)](https://codecov.io/github/mariomac/gostream)

Type safe Stream processing library inspired in the [Java Streams API](https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html).

## Table of contents

* [Requirements](#requirements)
* [Usage examples](#usage-examples)
* [Limitations](#limitations)
* [Performance](#performance)
* [Completion status](#completion-status)
* [Extra credits](#extra-credits)

## Requirements

* Go 1.24 or higher

## Usage examples

For more details about the API, and until [pkg.go.dev](https://pkg.go.dev/github.com/mariomac/gostream), 
is able to parse documentation for functions and types using generics, you can have a quick look
to the generated [go doc text descriptions](./docs), or just let that the embedded document browser of your
IDE does the job.

### Example 1: basic creation, transformation and iteration

1. Creates a literal stream containing all the integers from 1 to 11.
2. From the Stream, selects all the integers that are prime
3. Iterates the stream. For each filtered int, prints a message.

```go
import (
  "fmt"
  "github.com/mariomac/gostream/stream"
)

func main() {
  numbers := stream.Of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11)

  for n := range numbers.Filter(isPrime).Seq() {
    fmt.Printf("%d is a prime number\n", n)
  }
}

func isPrime(n int) bool {
  for i := 2; i <= n/2; i++ {
    if n%i == 0 {
      return false
    }
  }
  return true
}
```

Output: 
```
1 is a prime number
2 is a prime number
3 is a prime number
5 is a prime number
7 is a prime number
11 is a prime number
```

Alternatively, you can use the `ForEach` method to iterate the stream in a functional way:

```go
func main() {
  stream.Of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11).
    Filter(isPrime).
    ForEach(func(n int) {
      fmt.Printf("%d is a prime number\n", n)
    })
}
```
### Example 2: generation, map, limit and slice conversion

1. Creates an **infinite** stream of random integers (no problem, streams are evaluated lazily!)
2. Divides the random integer to get a number between 1 and 6
3. Limits the infinite stream to 5 elements.
4. Collects the stream items as a slice.

```go
rnd := rand.New(rand.NewSource(time.Now().UnixMilli()))
fmt.Println("let me throw 5 times a dice for you")

results := stream.Generate(rnd.Int).
    Map(func(n int) int {
        return n%6 + 1
    }).
    Limit(5).
    ToSlice()

fmt.Printf("results: %v\n", results)
```

Output:
```
let me throw 5 times a dice for you
results: [3 5 2 1 3]
```

### Example 3: Generation from an iterator, Map to a different type

1. Generates an infinite stream composed by `1`, `double(1)`, `double(double(1))`, etc...
   and cut it to 6 elements.
2. Maps the numbers' stream to a strings' stream. Because, at the moment,
   [go does not allow type parameters in methods](https://github.com/golang/go/issues/49085),
   we need to invoke the `stream.Map` function instead of the `numbers.Map` method
   because the contained type of the output stream (`string`) is different than the type of
   the input stream (`int`).
3. Converts the words stream to a slice and prints it.


```go
func main() {
    numbers := stream.Iterate(1, double).Limit(6)
    words := stream.Map(numbers, asWord).ToSlice()
    fmt.Println(words)
}

func double(n int) int {
    return 2 * n
}

func asWord(n int) string {
    if n < 10 {
        return []string{"zero", "one", "two", "three", "four", "five",
            "six", "seven", "eight", "nine"}[n]
    } else {
        return "many"
    }
}
```

Output:
```
[one two four eight many many]
```

### Example 4: deduplication of elements

Following example requires to compare the elements of the Stream, so the Stream needs to be
composed by `comparable` elements (this is, accepted by the the `==` and `!=` operators):

1. Instantiate a `Stream` of `comparable` items.
2. Pass it to the `Distinct` method, that will return a copy of the original Stream without
   duplicates
3. Operating as any other stream.

```go
words := stream.Distinct(
  stream.Of("hello", "hello", "!", "ho", "ho", "ho", "!"),
).ToSlice()

fmt.Printf("Deduplicated words: %v\n", words)
```

Output:

```
Deduplicated words: [hello ! ho]
```

### Example 5: sorting from higher to lower

1. Generate a stream of uint32 numbers.
2. Picking up 5 elements.
3. Sorting them by the inverse natural order (from higher to lower)
   - It's **important** to limit the number of elements, avoiding invoking
     `Sorted` over an infinite stream (otherwise it would panic).

```go
fmt.Println("picking up 5 random numbers from higher to lower")
stream.Generate(rand.Uint32).
    Limit(5).
    Sorted(order.Inverse(cmp.Compare[uint32])).
    ForEach(func(n uint32) {
        fmt.Println(n)
    })
```

Output:

```
picking up 5 random numbers from higher to lower
4039455774
2854263694
2596996162
1879968118
1823804162
```

### Example 6: Reduce and helper functions

1. Generate an infinite incremental Stream (1, 2, 3, 4...) using the `stream.Iterate`
   instantiator and the `item.Increment` helper function.
2. Limit the generated to 8 elements
3. Reduce all the elements multiplying them using the item.Multiply helper function

```go
fac8, _ := stream.Iterate(1, item.Increment[int]).
    Limit(8).
    Reduce(item.Multiply[int])
fmt.Println("The factorial of 8 is", fac8)
```

Output: 

```
The factorial of 8 is 40320
```

### Other examples: interact with Go's `iter` package

Check out this blog post: [Go streams meet standard Go iterators!](https://macias.info/entry/202508160000_gostream_meets_goiter.md)

## Limitations

Due to the initial limitations of Go generics, the API has the following limitations.
We will work on overcome them as long as new features are added to the Go type parameters
specification.

* You can use `Map` and `FlatMap` as method as long as the output element has the same type of the input.
  If you need to map to a different type, you need to use `stream.Map` or `stream.FlatMap` as functions.
* There is no `Distinct` method. There is only a `stream.Distinct` function.
* There is no `ToMap` method. There is only a `stream.ToMap` function.

## Performance

You might want to check: [Performance comparison of Go functional stream libraries](https://macias.info/entry/202212020000_go_streams.md).

Streams aren't the fastest option. They are aimed for complex workflows where you can
sacrifice few microseconds for the sake of code organization and readability. Also disclaimer:
functional streams don't have to always be the most readable option.

The following results show the difference in performance for an arbitrary set of operations
in an imperative form versus the functional form using streams (see
[stream/benchs_test.go file](stream/benchs_test.go)):

```
$ gotip test -bench=. -benchmem  ./...
goos: darwin
goarch: amd64
pkg: github.com/mariomac/gostream/stream
cpu: Intel(R) Core(TM) i5-5257U CPU @ 2.70GHz
BenchmarkImperative-4            2098518               550.6 ns/op          1016 B/op          7 allocs/op
BenchmarkFunctional-4             293095              3653 ns/op            2440 B/op         23 allocs/op
```

If you want a more performant, parallelizable alternative to create data processing pipelines (following
a programming model focused on Extract-Transform-Load, ETL), you
could give a try to my alternative project: [PIPES: Processing In Pipeline-Embedded Stages](https://github.com/mariomac/pipes).

## Completion status

* Stream instantiation functions
  - [X] Comparable
  - [X] Concat
  - [X] Empty
  - [X] Generate
  - [X] Iterate
  - [X] Of
  - [X] OfMap
  - [x] OfSlice
  - [X] OfChannel
* Stream transformers
  - [X] Distinct
  - [X] Filter
  - [X] FlatMap
  - [X] Limit
  - [X] Map
  - [X] Peek
  - [X] Skip
  - [X] Sorted
* Collectors/Terminals
  - [X] ToMap
  - [X] ToSlice
  - [X] AllMatch
  - [X] AnyMatch
  - [X] Count
  - [X] FindFirst
  - [X] ForEach
  - [X] Max
  - [X] Min
  - [X] NoneMatch
  - [X] Reduce
  - [X] Iter
* Auxiliary Functions
  - [X] Add (for numbers)
  - [X] Increment (for numbers)
  - [X] IsZero
  - [X] Multiply (for numbers)
  - [X] Neg (for numbers)
  - [X] Not (for bools)
* Future
  - [ ] Collectors for future standard generic data structures
    - E.g. [ ] Join (for strings)
  - [ ] Allow users implement their own Comparable or Ordered types
  - [ ] More operations inspired in the Kafka Streams API
  - [ ] Parallel streams 
    - [ ] FindAny


## Extra credits

The Stream processing and aggregation functions are heavily inspired in the
[Java Stream Specification](https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html).

Stream code documentation also used 
[Stream Javadoc](https://docs.oracle.com/javase/8/docs/api/java/util/stream/Stream.html) as an
essential reference and might contain citations from it.



gostream

Go library for writing and generating stream workers with built in metrics and traceability

[![Go](https://github.com/whitaker-io/machine/actions/workflows/go.yml/badge.svg)](https://github.com/whitaker-io/machine/actions/workflows/go.yml)
[![CodeQL](https://github.com/whitaker-io/machine/actions/workflows/github-code-scanning/codeql/badge.svg)](https://github.com/whitaker-io/machine/actions/workflows/github-code-scanning/codeql)
[![PkgGoDev](https://pkg.go.dev/badge/github.com/whitaker-io/machine)](https://pkg.go.dev/github.com/whitaker-io/machine)
[![GoDoc](https://godoc.org/github.com/whitaker-io/machine?status.svg)](https://godoc.org/github.com/whitaker-io/machine)
[![Go Report Card](https://goreportcard.com/badge/github.com/whitaker-io/machine)](https://goreportcard.com/report/github.com/whitaker-io/machine)
[![Codacy Badge](https://app.codacy.com/project/badge/Grade/aa8efa7beb3f4e66a5dc0247e25557b5)](https://app.codacy.com/gh/whitaker-io/machine/dashboard?utm_source=gh&utm_medium=referral&utm_content=&utm_campaign=Badge_grade)
[![Codacy Badge](https://app.codacy.com/project/badge/Coverage/aa8efa7beb3f4e66a5dc0247e25557b5)](https://app.codacy.com/gh/whitaker-io/machine/dashboard?utm_source=gh&utm_medium=referral&utm_content=&utm_campaign=Badge_coverage)
[![Version Badge](https://img.shields.io/github/v/tag/whitaker-io/machine)](https://img.shields.io/github/v/tag/whitaker-io/machine)

<p align="center">
    <img alt="Machine" height="125" src="https://raw.githubusercontent.com/whitaker-io/machine/master/docs/static/Black-No-BG.png">
</p>

`Machine` is a library for creating data workflows. These workflows can be either very concise or quite complex, even allowing for cycles for flows that need retry or self healing mechanisms.



------

### **Installation**

Add the primary library to your project
```bash
  go get github.com/whitaker-io/machine/v3
```

------

The two function types are:

```golang
// Monad is a function that is applied to payload and used for transformations
type Monad[T any] func(d T) T

// Filter is a function that can be used to filter the payload.
type Filter[T any] func(d T) bool

```

These are used in the `Machine` for functional operations

```golang
// New is a function for creating a new Machine.
//
// name string
// input chan T
// option ...Option
//
// Call the startFn returned by New to start the Machine once built.
func New[T any](name string, input chan T, options ...Option) (startFn func(context.Context), x Machine[T])

// Transform is a function for converting the type of the Machine. Cannot be used inside a loop
// until I figure out how to do it without some kind of run time error or overly complex
// tracking method that isn't type safe. I really wish method level generics were a thing.
func Transform[T, U any](m Machine[T], fn func(d T) U) (Machine[U], error)

// Machine is the interface provided for creating a data processing stream.
type Machine[T any] interface {
	// Name returns the name of the Machine path. Useful for debugging or reasoning about the path.
	Name() string
	
	// Then apply a mutation to each individual element of the payload.
	Then(a Monad[T]) Machine[T]

	// Recurse applies a recursive function to the payload through a Y Combinator.
	// f is a function used by the Y Combinator to perform a recursion
	// on the payload.
	// Example:
	//
	//	func(f Monad[int]) Monad[int] {
	//		 return func(x int) int {
	//			 if x <= 0 {
	//				 return 1
	//			 } else {
	//				 return x * f(x-1)
	//			 }
	//		 }
	//	}
	Recurse(x Monad[Monad[T]]) Machine[T]

	// Memoize applies a recursive function to the payload through a Y Combinator
	// and memoizes the results based on the index func.
	// f is a function used by the Y Combinator to perform a recursion
	// on the payload.
	// Example:
	//
	//	func(f Monad[int]) Monad[int] {
	//		 return func(x int) int {
	//			 if x <= 0 {
	//				 return 1
	//			 } else {
	//				 return x * f(x-1)
	//			 }
	//		 }
	//	}
	Memoize(x Monad[Monad[T]], index func(T) string) Machine[T]

	// Or runs all of the functions until one succeeds or sends the payload to the right branch
	Or(x ...Filter[T]) (Machine[T], Machine[T])

	// And runs all of the functions and if one doesnt succeed sends the payload to the right branch
	And(x ...Filter[T]) (Machine[T], Machine[T])

	// Filter splits the data into multiple stream branches
	If(f Filter[T]) (Machine[T], Machine[T])

	// Select applies a series of Filters to the payload and returns a list of Builders
	// the last one being for any unmatched payloads.
	Select(fns ...Filter[T]) []Machine[T]

	// Tee duplicates the data into multiple stream branches.
	Tee(func(T) (a, b T)) (Machine[T], Machine[T])

	// While creates a loop in the stream based on the filter
	While(x Filter[T]) (loop, out Machine[T])

	// Drop terminates the data from further processing without passing it on
	Drop()

	// Distribute is a function used for fanout
	Distribute(Edge[T]) Machine[T]
	
	// Output provided channel
	Output() chan T
}
```

`Distribute` is a special method used for fan-out operations. It takes an instance of `Edge[T]` and can be used most typically to distribute work via a Pub/Sub or it can be used in a commandline utility to handle user input or a similiar blocking process. 


The `Edge[T]` interface is as follows:

```golang
// Edge is an interface that is used for transferring data between vertices
type Edge[T any] interface {
	Output() chan T
	Send(payload T)
}
```

The `Send` method is used for data leaving the associated vertex and the `Output` method is used by the following vertex to receive data from the channel.

------

Confirguration is done using the `Option` helper

```golang
// Option is used to configure the machine
type Option interface

// OptionFIF0 controls the processing order of the payloads
// If set to true the system will wait for one payload
// to be processed before starting the next.
var OptionFIF0 Option

// OptionBufferSize sets the buffer size on the edge channels between the
// vertices, this setting can be useful when processing large amounts
// of data with FIFO turned on.
func OptionBufferSize(size int) Option

// OptionAttributes apply the slog.Attr's to the machine metrics and spans
// Do not override the "name", "type", "duration", "error", or "value" attributes
func OptionAttributes(attributes ...slog.Attr) Option

// OptionFlush attempts to send all data to the flushFN before exiting after the gracePeriod has expired
// Im looking for a good way to make this type specific, but want to avoid having to add separate option
// settings for the Transform function.
func OptionFlush(gracePeriod time.Duration, flushFN func(vertexName string, payload any)) Option
```

`Machine` supports collecting metrics and traces through a `log/slog` wrapper that sends 
the telemetry to the provided [OpenTelemetry](https://github.com/open-telemetry/opentelemetry-go) `Meter` and `Tracer`

```golang
// import "github.com/whitaker-io/machine/telemetry"

// Make your slog handler however you please
yourSlogHandler := slog.Default().Handler()

// wrap your handler and provide your tracer and meter
telemetryHandler := telemetry.New(
	yourSlogHandler,
	meterProvider.Meter("your_meter"), // Your otel metric.Meter
	tracerProvider.Tracer("your_tracer"), // Your otel trace.Tracer
	false, // Log Metrics and Traces to logs as well (useful for debugging)
)

slog.SetDefault(slog.New(telemetryHandler))
```

Examples of `Edge` implentations can be found in the edge directory and can be used as follows

```golang
// import "github.com/whitaker-io/machine/edge/pubsub"

func New[T any](
	ctx context.Context,
	subscription *pubsub.Subscription,
	publisher *pubsub.Topic,
	to func(T) *pubsub.Message,
	from func(context.Context, *pubsub.Message) T,
) machine.Edge[T]

// import "github.com/whitaker-io/machine/edge/http"

func New[T any](c http.Client, fn func(context.Context, T) *http.Request) machine.Edge[T]
```

------

***
## 🤝 Contributing

Contributions, issues and feature requests are welcome.<br />
Feel free to check [issues page](https://github.com/whitaker-io/machine/issues) if you want to contribute.<br />
[Check the contributing guide](./CONTRIBUTING.md).<br />

## Author

👤 **Jonathan Whitaker**

- Twitter: [@io_whitaker](https://twitter.com/io_whitaker)
- Github: [@jonathan-whitaker](https://github.com/jonathan-whitaker)

## Show your support

Please ⭐️ this repository if this project helped you!

***
## [License](#license)

Machine is provided under the [MIT License](https://github.com/whitaker-io/machine/blob/master/LICENSE).

```text
The MIT License (MIT)

Copyright (c) 2020 Jonathan Whitaker
```


machine

A lightweight package for micro batch processing

<p align="center"><img src="https://github.com/user-attachments/assets/1b34c21a-8031-43d3-a172-44e039b58190" alt="nibbler gopher" width="256px"/></p>

[![](https://github.com/naughtygopher/nibbler/actions/workflows/go.yml/badge.svg?branch=main)](https://github.com/naughtygopher/nibbler/actions)
[![Go Reference](https://pkg.go.dev/badge/github.com/naughtygopher/nibbler.svg)](https://pkg.go.dev/github.com/naughtygopher/nibbler)
[![Go Report Card](https://goreportcard.com/badge/github.com/naughtygopher/nibbler?cache_invalidate=1)](https://goreportcard.com/report/github.com/naughtygopher/nibbler)
[![Coverage Status](https://coveralls.io/repos/github/naughtygopher/nibbler/badge.svg?branch=main&cache_invalidate=1)](https://coveralls.io/github/naughtygopher/nibbler?branch=main)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://github.com/creativecreature/sturdyc/blob/master/LICENSE)
[![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go#stream-processing)

# Nibbler

Nibbler is a resilient & minimal package which helps you implement micro-batch processing, _within an application_. Nibbler remains minimal with its 0 external dependencies and remains resilient within the context of the application by gracefully handling errors and panics.

IMPORTANT: This is not a general purpose _*distributed*_ task queue.

## What is Micro-batch Processing?

Micro-batch processing is a way to handle data by breaking a big task into smaller pieces and processing them one piece at a time. This method is useful in processing realtime data or stream processing, where the incoming data is split into "micro-batches" and processed quickly, rather than waiting to collect all data at once or process one by one.

The same concept can also be extended to handle events processing. So, we have a queue subscriber, and instead of processing the events individually, we create micro batches and process them.

The processing of a single micro batch can be triggered in two ways, based on a time ticker or if the micro batch size is full. i.e. process a non empty batch if duration X has passed or if the batch size is full

<p align="center">
<img src="https://github.com/user-attachments/assets/0a7df1c0-2d23-475e-9cc3-205f3f9bf4c4" alt="nibbler" width="384px"/>
</p>

## Why use nibbler?

High-throughput event and stream processing systems benefit significantly from batch processing rather than handling events individually. Well-implemented batching optimizes resource utilization across downstream dependencies like databases, APIs, and external systems-by dramatically reducing [IOPS](https://en.wikipedia.org/wiki/IOPS). Batch operations enable more efficient use of network round-trips, connection pools, and database transaction overhead. For example, a single bulk INSERT of 100 rows is substantially cheaper than 100 individual INSERT statements, both in terms of I/O operations and connection utilization.

However, batching introduces a latency tradeoff. A naive implementation that waits for a batch to reach full capacity before processing will delay event handling indefinitely during low-traffic periods. If your batch size is 100 and events arrive at 1 per second, you'd wait over a minute and a half before any processing occurs-unacceptable for systems requiring near-realtime responsiveness.

Nibbler addresses this with a dual-trigger flush mechanism: batches are processed either when they reach a configured size threshold or when a time interval elapses; whichever occurs first. This bounded-latency approach ensures:

1. **High throughput scenarios**: Batches fill quickly and flush at capacity, maximizing I/O efficiency

2. **Low throughput scenarios**: Partial batches flush at the configured interval, guaranteeing a maximum processing delay

3. **Variable load patterns**: The system self-adjusts without manual intervention, processing full batches during traffic spikes while maintaining responsiveness during quiet periods

The configurable flush interval (TickerDuration) represents your maximum acceptable latency. The worst-case delay between event arrival and processing. The batch size (Size) determines your maximum I/O optimization. Together, these parameters let you tune the latency-efficiency tradeoff to match your specific requirements.

### Config

```golang
type BatchProcessor[T any] func(ctx context.Context, trigger trigger, batch []T) error

type Config[T any] struct {
	// ProcessingTimeout is context timeout for processing a single batch. If less than 1ms, defaults to 1s
	ProcessingTimeout time.Duration
	// TickerDuration is the duration after which a non-empty batch would be flushed. If less than 1ms, defaults to 1s
	TickerDuration time.Duration

	// Size is the micro batch size. If 0, defaults to 100
	Size uint

	// Processor is a required configuration, it is called when a batch process is initiated either by
	// ticker or when the batch is full. The 'batch' slice received must not be changed, if you have to
	// process the batch asynchronously, copy the batch to a new slice prior to sending it to a Go routine.
	Processor BatchProcessor[T]

	// ResumeAfterErr if true will continue listening and keep processing if the processor returns
	// an error, or if processor panics. In both cases, ProcessorErr would be executed
	ResumeAfterErr bool
	// ProcessorErr is the function which is executed if processor encounters an error or panic
	ProcessorErr func(failedBatch []T, err error)
}
```

## How to use nibbler?

Below is an example showing how batching is used for a "banking" app which bulk processes account statements.

```golang
package main

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

	"github.com/naughtygopher/nibbler"
)

type db struct {
	data         sync.Map
	totalBalance int
}

func (d *db) BulkAddAccountsAndBalance(pp []AccStatement) error {
	// assume we are doing a bulk insert/update into the database instead of inserting one by one.
	// Bulk operations reduce the number of I/O required between your application and the database.
	// Thereby making it better in most cases.
	for _, p := range pp {
		d.data.Store(p.AccountID, p.Balance)
		d.totalBalance += p.Balance
	}
	return nil
}

type Bank struct {
	db *db
}

func (bnk *Bank) ProcessAccountsBatch(
	ctx context.Context,
	trigger nibbler.Trigger,
	batch []AccStatement,
) error {
	err := bnk.db.BulkAddAccountsAndBalance(batch)
	if err != nil {
		return err
	}

	return nil
}

func (bnk *Bank) TotalBalance() int {
	return bnk.db.totalBalance
}

func (bnk *Bank) TotalAccounts() int {
	counter := 0
	bnk.db.data.Range(func(key, value any) bool {
		counter++
		return true
	})
	return counter
}

type AccStatement struct {
	AccountID string
	Balance   int
}

func main() {
	bnk := Bank{
		db: &db{
			data: sync.Map{},
		},
	}

	nib, err := nibbler.Start(&nibbler.Config[AccStatement]{
		Size:           10,
		TickerDuration: time.Second,
		Processor:      bnk.ProcessAccountsBatch,
	})
	if err != nil {
		panic(err)
	}

	receiver := nib.Receiver()
	for i := range 100 {
		accID := fmt.Sprintf("account_id_%d", i)
		receiver <- AccStatement{
			AccountID: accID,
			Balance:   50000 / (i + 1),
		}
	}

	// wait for batches to be processed. Ideally this wouldn't be required as our application
	// would not exit, instead just keep listening to the events stream.
	time.Sleep(time.Second)

	fmt.Printf(
		"Number of accounts %d, total balance: %d\n",
		bnk.TotalAccounts(),
		bnk.TotalBalance(),
	)
}
```

You can find all usage details in the tests.

## The gopher

The gopher used here was created using [Gopherize.me](https://gopherize.me/). Nibbler is out there eating your events/streams
one chunky bite at a time.


nibbler

Reactive Programming: declarative and composable API for event-driven applications


# ro - Streams & reactive programming for Go

[![tag](https://img.shields.io/github/tag/samber/ro.svg)](https://github.com/samber/ro/releases)
![Go Version](https://img.shields.io/badge/Go-%3E%3D%201.18-%23007d9c)
[![GoDoc](https://godoc.org/github.com/samber/ro?status.svg)](https://pkg.go.dev/github.com/samber/ro)
![Build Status](https://github.com/samber/ro/actions/workflows/test.yml/badge.svg)
[![Go report](https://goreportcard.com/badge/github.com/samber/ro)](https://goreportcard.com/report/github.com/samber/ro)
[![Coverage](https://img.shields.io/codecov/c/github/samber/ro)](https://app.codecov.io/gh/samber/ro)
[![Contributors](https://img.shields.io/github/contributors/samber/ro)](https://github.com/samber/ro/graphs/contributors)

> A *Go* implementation of the [ReactiveX](https://reactivex.io/) spec.

The purpose of Reactive Programming is to simplify the development of event-driven and asynchronous applications by providing a declarative and composable way to handle streams of data or events.

----

<div align="center">
  <sup><b>💖 Sponsored by:</b></sup>
  <br>
  <a href="https://www.dbos.dev/?utm_campaign=gh-smbr">
    <div>
	  <img width="200" alt="dbos" src="https://github.com/user-attachments/assets/d583cb62-7735-4d3c-beb7-e6ef1a5faf49" />
    </div>
    <div>
      DBOS - Durable workflow orchestration library for Go
    </div>
  </a>
</div>

----

![cover](/docs/static/img/cover.png)

**See also:**

- [samber/lo](https://github.com/samber/lo): A Lodash-style Go library based on Go 1.18+ Generics
- [samber/do](https://github.com/samber/do): A dependency injection toolkit based on Go 1.18+ Generics
- [samber/mo](https://github.com/samber/mo): Monads based on Go 1.18+ Generics (Option, Result, Either...)
- [samber/cc-skills-golang](https://github.com/samber/cc-skills-golang): AI Agent Skills for Golang

What makes it different from **samber/lo**?
- lo: synchronous helpers across finite sequences (maps, slices...)
- ro: processing of infinite data streams for event-driven scenarios

## The Reactive Programming paradigm

Reactive Programming is focused on handling asynchronous data streams where values (like user input, API responses, or sensor data) are emitted over time. Instead of pulling data or waiting for events manually, you react to changes as they occur using `Observable`, `Observer`, and `Operator`. This approach simplifies building systems that are responsive, resilient, and scalable, especially in event-driven or real-time applications.

```go
observable := ro.Pipe(
    ro.RangeWithInterval(0, 5, 1*time.Second),
    ro.Filter(func(x int) bool {
        return x%2 == 0
    }),
    ro.Map(func(x int) string {
        return fmt.Sprintf("even-%d", x)
    }),
)

// Start consuming on subscription
observable.Subscribe(ro.NewObserver(
    func(s string) { fmt.Println(s) },
    func(err error) { fmt.Println(err.Error()) },
    func() { fmt.Println("Completed!") }
))
// Output:
//   "even-0"
//   "even-2"
//   "even-4"
//   "Completed!"

// or:

values, err := ro.Collect(observable)
// []string{"even-0", "even-2", "even-4"}
// <nil>
```

Now you discovered the paradigm, follow the documentation and turn reactive: [🚀 Getting started](https://ro.samber.dev/docs/getting-started)


## 🚀 Install

```sh
go get github.com/samber/ro

# AI Agent Skill
npx skills add https://github.com/samber/cc-skills-golang --skill golang-samber-ro
```

This library is v0 and follows SemVer strictly.

Some breaking changes might be made to exported APIs before v1.0.0. Experimental packages under `exp/` are considered as unstable.

This library has minimal dependencies outside the Go standard library.

## Core package

[Full documentation here](https://ro.samber.dev/docs/operator).

The `ro` library provides all basic operators:
- **Creation operators**: The data source, usually the first argument of `ro.Pipe`
- **Chainable operators**: They filter, validate, transform, enrich... messages
  - **Transforming operators**: They transform items emitted by an `Observable`
  - **Filtering operators**: They selectively emit items from a source `Observable`
  - **Conditional operators**: Boolean operators
  - **Math and aggregation operators**: They perform basic math operations
  - **Error handling operators**: They help to recover from error notifications from an `Observable`
  - **Combining operators**: Combine multiple `Observable` into one
  - **Connectable operators**: Convert cold into hot `Observable`
  - **Other**: manipulation of context, utility, async scheduling...
- **Plugins**: External operators (mostly IOs and library wrappers)

## Plugins

The `ro` library provides a rich ecosystem of plugins for various use cases:

[Full documentation here](https://ro.samber.dev/docs/plugins).

### Data Manipulation
- **Bytes** (`plugins/bytes`) - String and byte slice manipulation operators
- **Strings** (`plugins/strings`) - String manipulation operators
- **Sort** (`plugins/sort`) - Sorting operators
- **Type Conversion** (`plugins/strconv`) - String conversion operators
- **SIMD** (`plugins/exp/simd`) - SIMD-accelerated transformations

### Encoding & Serialization
- **JSON** (`plugins/encoding/json`) - JSON marshaling and unmarshaling
- **CSV** (`plugins/encoding/csv`) - CSV reading and writing
- **Base64** (`plugins/encoding/base64`) - Base64 encoding and decoding
- **Gob** (`plugins/encoding/gob`) - Go binary serialization

### Scheduling & Timing
- **Cron** (`plugins/cron`) - Schedule jobs using cron expressions or duration intervals
- **ICS** (`plugins/ics`) - Read and parse ICS/iCal calendars

### Network & I/O
- **HTTP** (`plugins/http`) - HTTP request operators
- **I/O** (`plugins/io`) - File and stream I/O operators
- **File System** (`plugins/fsnotify`) - File system monitoring operators

### Observability & Logging
- **Log** (`plugins/observability/log`) - Standard logging operators
- **Zap** (`plugins/observability/zap`) - Structured logging with zap
- **Logrus** (`plugins/observability/logrus`) - Structured logging with logrus
- **Slog** (`plugins/observability/slog`) - Structured logging with slog
- **Zerolog** (`plugins/observability/zerolog`) - Structured logging with zerolog
- **Sentry** (`plugins/observability/sentry`) - Error tracking with Sentry
- **Oops** (`plugins/samber/oops`) - Structured error handling

### Rate Limiting
- **Native** (`plugins/ratelimit/native`) - Native rate limiting operators
- **Ulule** (`plugins/ratelimit/ulule`) - Rate limiting with ulule/limiter

### Text Processing
- **Regular Expressions** (`plugins/regexp`) - Regular expression operators
- **Templates** (`plugins/template`) - Template processing operators

### System Integration
- **Process** (`plugins/proc`) - Process execution operators
- **Signal** (`plugins/signal`) - Signal handling operators
- **Iterators** (`plugins/iter`) - Iterator operators
- **PSI** (`plugins/samber/psi`) - Starvation notifier

### Data Validation
- **Validation** (`plugins/ozzo/ozzo-validation`) - Data validation operators

### Testing
- **Testing** (`plugins/testify`) - Testing utilities

### Utilities
- **HyperLogLog** (`plugins/hyperloglog`) - Cardinality estimation operators
- **Hot** (`plugins/samber/hot`) - In-memory cache

## 📚 Documentation

- [Documentation](https://ro.samber.dev) - Official doc
- [Godoc](https://pkg.go.dev/github.com/samber/ro) - API Reference
- [Plugins](./plugins) - Individual plugin documentation
- [Examples](./examples) - Working examples

## 👀 Examples

See the [examples](./examples) directory for complete working examples:

- [Stocker price enrichment](./examples/stock-price-enrichment/) - Demonstrate a websocket client with data enrichment
- [Connectable](./examples/connectable) - Demonstrates connectable observables
- [Distributed WebSocket Gateway](./examples/distributed-websocket-gateway) - Shows how to build a distributed WebSocket gateway
- [Parallel API Requests](./examples/parallel-api-requests) - Demonstrates concurrent HTTP requests
- [SQL to CSV](./examples/sql-to-csv) - Shows how to process database results to CSV
- [ICS to CSV](./examples/ics-to-csv) - Shows how to process ICS calendar files to CSV format
- [Enterprise Edition Examples](./examples/ee-*) - Examples using enterprise features

## 🤝 Contributing

Check the [contribution guide](https://ro.samber.dev/docs/contributing).

- Ping me on Twitter [@samuelberthe](https://twitter.com/samuelberthe) (DMs, mentions, whatever :))
- Fork the [project](https://github.com/samber/ro)
- Fix [open issues](https://github.com/samber/ro/issues) or request new features

Don't hesitate ;)

## 👤 Contributors

![Contributors](https://contrib.rocks/image?repo=samber/ro)

## 💫 Show your support

Give a ⭐️ if this project helped you!

[![GitHub Sponsors](https://img.shields.io/github/sponsors/samber?style=for-the-badge)](https://github.com/sponsors/samber)

## License

Copyright © 2025 [Samuel Berthe](https://github.com/samber).

This project is licensed under the Apache 2.0 License - see the [LICENSE](LICENSE) file for details.

**Note**: The `ee/` directory contains the Enterprise Edition of the library, which is subject to a custom license. Please refer to the [ee/LICENSE.md](ee/LICENSE.md) file for the specific terms and conditions applicable to the Enterprise Edition.


Type-safe reactive state management inspired by Angular Signals with computed values, effects, and dependency tracking

# Signals

> **Type-safe reactive state management for Go, inspired by Angular Signals**

[![Release](https://img.shields.io/github/v/release/coregx/signals?style=flat-square&logo=github)](https://github.com/coregx/signals/releases)
[![Go Version](https://img.shields.io/github/go-mod/go-version/coregx/signals?style=flat-square&logo=go)](https://go.dev)
[![Go Report Card](https://goreportcard.com/badge/github.com/coregx/signals?style=flat-square)](https://goreportcard.com/report/github.com/coregx/signals)
[![GoDoc](https://img.shields.io/badge/godoc-reference-blue?style=flat-square&logo=go)](https://pkg.go.dev/github.com/coregx/signals)
[![CI](https://img.shields.io/github/actions/workflow/status/coregx/signals/test.yml?branch=main&style=flat-square&logo=github&label=tests)](https://github.com/coregx/signals/actions)
[![codecov](https://codecov.io/gh/coregx/signals/graph/badge.svg)](https://codecov.io/gh/coregx/signals)
[![License](https://img.shields.io/badge/license-MIT-blue?style=flat-square)](LICENSE)
[![Stars](https://img.shields.io/github/stars/coregx/signals?style=flat-square&logo=github)](https://github.com/coregx/signals/stargazers)

A modern, production-grade reactive programming library for Go 1.25+ that brings Angular's powerful signals pattern to the Go ecosystem with full type safety, zero allocations in hot paths, and comprehensive concurrency support.

---

## Features

- **Pure Go** - No dependencies, works everywhere Go works
- **Type-Safe** - Full generic support with Go 1.25+ type parameters
- **Thread-Safe** - Built-in synchronization for concurrent access
- **Zero Allocations** - Hot paths designed for zero heap allocations
- **Angular-Inspired** - API design inspired by Angular Signals
- **Fine-Grained Reactivity** - Only re-compute what changed
- **Glitch-Free** - Atomic updates prevent intermediate states
- **Lazy Evaluation** - Computed values calculate only when needed
- **Effect Cleanup** - Automatic resource management with cleanup callbacks
- **Production Ready** - 55 tests, 94.5% coverage, 28 benchmarks

---

## Quick Start

### Installation

```bash
go get github.com/coregx/signals
```

Requires Go 1.25 or later.

### Basic Usage

```go
package main

import (
    "fmt"
    "github.com/coregx/signals"
)

func main() {
    // Create a reactive signal
    count := signals.New(0)

    // Create computed value with explicit dependencies
    doubled := signals.Computed(func() int {
        return count.Get() * 2
    }, count.AsReadonly())

    // Create effect that runs immediately and on dependency changes
    eff := signals.Effect(func() {
        fmt.Printf("Count: %d, Doubled: %d\n", count.Get(), doubled.Get())
    }, count.AsReadonly(), doubled)
    defer eff.Stop()
    // Output: Count: 0, Doubled: 0

    // Update signal - effect automatically re-runs
    count.Set(5)
    // Output: Count: 5, Doubled: 10

    count.Set(10)
    // Output: Count: 10, Doubled: 20
}
```

### Advanced Example

```go
// Multiple dependencies
firstName := signals.New("John")
lastName := signals.New("Doe")

fullName := signals.Computed(func() string {
    return firstName.Get() + " " + lastName.Get()
}, firstName.AsReadonly(), lastName.AsReadonly())

// Effect with cleanup
eff := signals.EffectWithCleanup(func() func() {
    fmt.Println("Full name:", fullName.Get())

    // Return cleanup function (runs before next execution and on Stop)
    return func() {
        fmt.Println("Effect cleaned up")
    }
}, fullName)

firstName.Set("Jane")  // Cleanup runs, then effect re-runs
lastName.Set("Smith")   // Cleanup runs, then effect re-runs

eff.Stop()  // Final cleanup runs, effect stops
```

[More examples →](cmd/example/)

---

## Documentation

### Reference
- **[API Reference](https://pkg.go.dev/github.com/coregx/signals)** - Complete API documentation
- **[Examples](cmd/example/)** - Working code examples

### Advanced
- **[Architecture Overview](docs/dev/ARCHITECTURE.md)** - How it works internally
- **[Implementation Guide](docs/dev/IMPLEMENTATION_GUIDE.md)** - Development guide
- **[Angular Signals Analysis](docs/dev/ANGULAR_SIGNALS_ANALYSIS.md)** - Comparison with Angular

---

## Current Status

**Version**: v0.1.0 (Stable - Production-ready!)

**Production Readiness: Core functionality complete and stable!**

**[See detailed roadmap →](ROADMAP.md)**

### Fully Implemented

#### Phase 1: Core Signal[T]
- Signal creation and basic operations
- Thread-safe read/write with RWMutex
- Subscription system with automatic unsubscribe
- Context-based auto-cancellation
- Panic recovery with custom handlers
- Read-only view support
- Custom equality functions
- Comprehensive test coverage (19 tests)
- Zero allocations in hot paths (verified by benchmarks)

#### Phase 2: Computed[T]
- Lazy evaluation with automatic caching
- Explicit dependency tracking and invalidation
- Fine-grained reactivity
- Glitch-free execution
- Thread-safe recomputation
- Comprehensive test coverage (17 tests)
- Optimized performance (minimal allocations)

#### Phase 3: Effect
- Explicit dependency tracking
- Cleanup function support (runs before re-execution and on Stop)
- Panic recovery with custom handlers
- Immediate execution on creation (Angular pattern)
- Comprehensive test coverage (16 tests)
- Concurrent effect management

### Test Coverage

| Package | Tests | Coverage | Benchmarks |
|---------|-------|----------|------------|
| signals | 55    | 94.5%    | 28         |

**Key Metrics**:
- 19 Signal tests
- 17 Computed tests
- 16 Effect tests
- 3 Internal tests (type erasure, reflection fallback)
- Zero allocations in signal read/write hot paths
- Race detector clean (all tests pass with `-race`)

### Performance Characteristics

```
BenchmarkSignal_Get            46738254     27.72 ns/op    0 B/op    0 allocs/op
BenchmarkSignal_Set            20975720     52.42 ns/op    0 B/op    0 allocs/op
BenchmarkComputed_Get_Clean    79329402     19.84 ns/op    0 B/op    0 allocs/op
BenchmarkEffect_Execute         8391343    139.6  ns/op    0 B/op    0 allocs/op
```

*Zero allocations in hot paths ensure minimal GC pressure*

### Remaining Work

#### Phase 4: Documentation (In Progress)
- User guides and tutorials
- API documentation examples
- Migration guides
- Best practices guide

#### Phase 5: Advanced Features (Planned v0.2.0)
- Resource tracking and lifecycle management
- Advanced batching strategies
- Performance monitoring and debugging tools
- Additional utility functions

See [ROADMAP.md](ROADMAP.md) for detailed timeline.

---

## Development

### Requirements
- Go 1.25 or later
- golangci-lint (for linting)
- No external runtime dependencies

### Building

```bash
# Clone repository
git clone https://github.com/coregx/signals.git
cd signals

# Run tests
make test

# Run tests with race detector
make test-race

# Run benchmarks
make benchmark

# Run linter
make lint
```

### Testing

```bash
# Run all tests
go test ./...

# Run with coverage
go test -coverprofile=coverage.out ./...
go tool cover -html=coverage.out

# Run with race detector
go test -race ./...

# Run benchmarks
go test -bench=. -benchmem ./...
```

### Code Quality

This project maintains high code quality standards:

```bash
# Format code
make fmt

# Run linter (zero issues required)
make lint

# Run all pre-commit checks
make pre-commit
```

---

## Contributing

Contributions are welcome! This is an early-stage project and we'd love your help.

**Before contributing**:
1. Read [CONTRIBUTING.md](CONTRIBUTING.md) - Development workflow and guidelines
2. Check [open issues](https://github.com/coregx/signals/issues)
3. Review the [Architecture Overview](docs/dev/ARCHITECTURE.md)

**Ways to contribute**:
- Report bugs
- Suggest features
- Improve documentation
- Submit pull requests
- Star the project

---

## Comparison with Other Libraries

| Feature | Signals | RxGo | Reactor |
|---------|---------|------|---------|
| Type-Safe Generics | Yes (Go 1.25+) | Limited | No |
| Zero Allocations | Yes (hot paths) | No | No |
| Thread-Safe | Yes (built-in) | Yes | Partial |
| Angular-Inspired | Yes | No | No |
| Fine-Grained Reactivity | Yes | Observable-based | Stream-based |
| Dependencies | Zero | Multiple | Multiple |
| Learning Curve | Low (if you know Angular) | Medium | Medium |

---

## Angular Signals Compatibility

This library is designed to be conceptually compatible with Angular Signals:

| Angular Signals | Go Signals | Status |
|----------------|------------|--------|
| `signal(T)` | `New[T](value)` | Complete |
| `computed(() => T)` | `Computed[T](fn, deps...)` | Complete |
| `effect(() => {})` | `Effect(fn, deps...)` | Complete |
| `signal.set(value)` | `signal.Set(value)` | Complete |
| `signal()` | `signal.Get()` | Complete |
| `signal.update(fn)` | `signal.Update(fn)` | Complete |
| `signal.asReadonly()` | `signal.AsReadonly()` | Complete |
| Automatic tracking | Explicit dependencies | Adapted for Go |
| Glitch-free | Glitch-free | Complete |
| Lazy computed | Lazy computed | Complete |

---

## License

This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.

---

## Acknowledgments

- The Angular team for the signals design pattern
- The Go team for generics and type parameters
- All contributors to this project

---

## Support

- [API Documentation](https://pkg.go.dev/github.com/coregx/signals)
- [Issue Tracker](https://github.com/coregx/signals/issues)
- [Discussions](https://github.com/coregx/signals/discussions)

---

**Status**: Stable - Production-ready!
**Version**: v0.1.0

---

*Built with care for the Go community*


signals

Go Stream, like Java 8 Stream: Filter/Map/FlatMap/Peek/Sorted/ForEach/Reduce..

# Go Stream

[![PkgGoDev](https://pkg.go.dev/badge/github.com/youthlin/stream)](https://pkg.go.dev/github.com/youthlin/stream)
[![Go Report Card](https://goreportcard.com/badge/github.com/youthlin/stream)](https://goreportcard.com/report/github.com/youthlin/stream)
[![Build Status](https://travis-ci.org/youthlin/stream.svg?branch=main)](https://travis-ci.org/youthlin/stream)
[![codecov](https://codecov.io/gh/youthlin/stream/branch/main/graph/badge.svg?token=1CqmLWbsYL)](https://codecov.io/gh/youthlin/stream)

Go Stream, like Java 8 Stream.

Blog Post: https://youthlin.com/?p=1755

## How to get

```shell script
go get github.com/youthlin/stream
```

国内镜像: https://gitee.com/youthlin/stream  
在 `go.mod` 中引入模块路径 `github.com/youthlin/stream` 及版本后，  
再添加 replace 即可：

```go
// go.mod

require github.com/youthlin/stream latest

replace github.com/youthlin/stream latest => gitee.com/youthlin/stream latest

```

## Play online

https://play.golang.org/p/nPQJYqA3-Jr

```go
package main

import (
	"fmt"

	"github.com/youthlin/stream"
	"github.com/youthlin/stream/types"
)

func main() {
	m := stream.IntRange(0, 10).
		Filter(func(e types.T) bool {
			return e.(int)%2 == 0
		}).
		Map(func(e types.T) types.R {
			return e.(int) * 2
		}).
		ReduceWith(map[int]string{}, func(acc types.R, e types.T) types.R {
			m := acc.(map[int]string)
			m[e.(int)] = fmt.Sprintf("<%d>", e)
			return m
		})
	fmt.Println(m)
	// Output:
	// map[0:<0> 4:<4> 8:<8> 12:<12> 16:<16>]
}
```

## Examples

```go

type Stream interface {
	// stateless operate 无状态操作

	Filter(types.Predicate) Stream         // 过滤
	Map(types.Function) Stream             // 转换
	FlatMap(func(t types.T) Stream) Stream // 打平
	Peek(types.Consumer) Stream            // peek 每个元素

	// stateful operate 有状态操作

	Distinct(types.IntFunction) Stream // 去重
	Sorted(types.Comparator) Stream    // 排序
	Limit(int64) Stream                // 限制个数
	Skip(int64) Stream                 // 跳过个数

	// terminal operate 终止操作

	// 遍历
	ForEach(types.Consumer)
	// return []T 转为切片
	ToSlice() []types.T
	// return []X which X is the type of some
	ToElementSlice(some types.T) types.R
	// return []X which X is same as the `typ` representation
	ToSliceOf(typ reflect.Type) types.R
	// 测试是否所有元素满足条件
	AllMatch(types.Predicate) bool
	// 测试是否没有元素满足条件
	NoneMatch(types.Predicate) bool
	// 测试是否有任意元素满足条件
	AnyMatch(types.Predicate) bool
	// Reduce return optional.Empty if no element. calculate result by (T, T) -> T from first element
	Reduce(accumulator types.BinaryOperator) optional.Optional
	// type of initValue is same as element.  (T, T) -> T
	ReduceFrom(initValue types.T, accumulator types.BinaryOperator) types.T
	// type of initValue is different from element. (R, T) -> R
	ReduceWith(initValue types.R, accumulator func(types.R, types.T) types.R) types.R
	FindFirst() optional.Optional
	// 返回元素个数
	Count() int64
}


func ExampleOf() {
	fmt.Println(stream.Of().Count())
	fmt.Println(stream.Of(1).Count())
	fmt.Println(stream.Of("a", "b").Count())
	var s = []int{1, 2, 3, 4}
	stream.Of(stream.Slice(s)...).ForEach(func(t types.T) {
		fmt.Printf("%d,", t)
	})
	// Output:
	// 0
	// 1
	// 2
	// 1,2,3,4,
}

func ExampleOfSlice() {
	var intArr = []int{1, 2, 3, 4}
	stream.OfSlice(intArr).ForEach(func(e types.T) {
		fmt.Printf("%d,", e)
	})
	var nilArr []int
	stream.OfSlice(nilArr).ForEach(func(e types.T) {
		fmt.Printf("should not print")
	})
	var strArr = []string{"a", "b"}
	stream.OfSlice(strArr).
		Map(func(e types.T) types.R {
			return fmt.Sprintf("<%s>", e)
		}).
		ForEach(func(e types.T) {
			fmt.Printf("%s,", e)
		})
	// Output:
	// 1,2,3,4,<a>,<b>,
}

func ExampleOfMap() {
	var m1 = map[int]string{
		3: "c",
		2: "b",
		1: "a",
	}
	s := stream.OfMap(m1).
		Map(func(e types.T) types.R {
			p := e.(types.Pair)
			p.First, p.Second = p.Second, p.First
			return p
		}).
		Sorted(func(left types.T, right types.T) int {
			p1 := left.(types.Pair)
			p2 := right.(types.Pair)
			return p1.Second.(int) - p2.Second.(int)
		}).
		ToSlice()
	fmt.Println(s)
	stream.OfMap(nil).ForEach(func(e types.T) {
		fmt.Println("not print")
	})
	// Output:
	// [{a 1} {b 2} {c 3}]
}

func ExampleStream_Filter() {
	stream.Of(0, 1, 2, 3, 4, 5, 6, 7, 8, 9).
		Filter(func(e types.T) bool {
			return e.(int)%3 == 0
		}).
		ForEach(func(e types.T) {
			fmt.Println(e)
		})
	// Output:
	// 0
	// 3
	// 6
	// 9
}
func ExampleStream_Map() {
	stream.IntRange(0, 5).
		Map(func(t types.T) types.R {
			return fmt.Sprintf("<%d>", t.(int))
		}).
		ForEach(func(t types.T) {
			fmt.Printf("%v", t)
		})
	// Output:
	// <0><1><2><3><4>
}
func ExampleStream_FlatMap() {
	stream.Of([]int{0, 2, 4, 6, 8}, []int{1, 3, 5, 7, 9}).
		FlatMap(func(t types.T) stream.Stream {
			return stream.Of(stream.Slice(t)...)
		}).
		ForEach(func(t types.T) {
			fmt.Printf("%d", t)
		})
	// Output:
	// 0246813579
}
func ExampleStream_Sorted() {
	stream.IntRange(1, 10).
		Sorted(types.ReverseOrder(types.IntComparator)).
		ForEach(func(t types.T) {
			fmt.Printf("%d,", t)
		})
	// Output:
	// 9,8,7,6,5,4,3,2,1,
}
func TestToMap(t *testing.T) {
	m := stream.IntRange(0, 10).ReduceWith(make(map[int]int), func(acc types.R, t types.T) types.R {
		acc.(map[int]int)[t.(int)] = t.(int) * 10
		return acc
	})
	t.Log(m)
	// Output:
	// map[0:0 1:10 2:20 3:30 4:40 5:50 6:60 7:70 8:80 9:90]
}

```

## Change Log

- v0.0.3 2020-12-08 add factory method: OfInts, OfInt64s, OfFloat32s, OfFloat64s, OfStrings;  
  add Stream method: ReduceBy
- v0.0.2 2020-12-07 add factory method: OfSlice, OfMap
- v0.0.1 2020-11-12 first version

## Todo

- [ ] add Benchmark test
- [ ] support parallel stream


stream

A lightweight streaming SQL engine for real-time data processing

The Best Go Libraries For Stream Processing (10)