ro

https://pkg.go.dev/github.com/samber/ro@v0.3.0
Admin
ro is a Go implementation of the ReactiveX specification for streams and reactive programming,...
Tokens:184,563
Snippets:1,195
Trust Score:10
Update:7 months ago
Show doc for...
Tokens:
Raw
### Install the ro Package in Go

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Provides the command to install the 'ro' package, a reactive programming library for Go, using the Go module system. This is the standard way to add external dependencies to a Go project.

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

```

--------------------------------

### Start Observable from Callback (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

Creates an observable that executes a given callback function and emits its return value. This is useful for integrating synchronous operations into an observable stream.

```Go
func Start[T any](cb func() T) Observable[T]
```

--------------------------------

### Start Operator

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

The Start operator creates an observable that executes a given function and emits its return value.

```APIDOC
## Start Operator

### Description
Creates an observable that executes a given function and emits its return value.

### Method
N/A (Functional operator)

### Endpoint
N/A

### Parameters
- **cb** (func() T) - The function to execute.

### Request Example
```go
startObs := Start(func() string { return "started" })
```

### Response
#### Success Response
- **Observable[T]** (Observable) - An observable sequence emitting the return value of the function.

#### Response Example
```go
// For Start(func() string { return "started" }), emits "started".
```
```

--------------------------------

### Start

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Creates an Observable that emits lazily a single value.

```APIDOC
## Start

### Description
Creates an Observable that emits lazily a single value.

### Method
N/A (Creation Operator)

### Endpoint
N/A

### Parameters
#### Path Parameters
N/A

#### Query Parameters
N/A

#### Request Body
N/A

### Request Example
N/A

### Response
#### Success Response (Observable emits item T)
- **value** (T) - The lazily emitted value.

#### Response Example
```
Start!
Next: 42
Completed
Start!
Next: 42
Completed
```
```

--------------------------------

### Create Observables from Callbacks and Errors

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Functions to create observables from a callback function or to emit a specific error. 'Start' executes a function and emits its result, while 'Throw' emits an error.

```Go
func Start[T any](cb func() T) Observable[T]
func Throw[T any](err error) Observable[T]
```

--------------------------------

### Observable Creation and Transformation Functions

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/index

This section details functions for creating and transforming Observables, including various Pipe operations, Race, RandFloat64, RandIntN, Range, Repeat, Start, Throw, Timer, and Zip operations.

```APIDOC
## Observable Creation and Transformation Functions

### Description
Provides functions for creating, transforming, and combining Observables. This includes a variety of `Pipe` functions for chaining operations, as well as functions for generating random numbers, ranges, repeating items, starting asynchronous operations, throwing errors, timers, and zipping multiple Observables.

### Functions

*   **`Pipe20[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[U]`**: Chains 20 operators.
*   **`Pipe21[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[V]`**: Chains 21 operators.
*   **`Pipe22[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[W]`**: Chains 22 operators.
*   **`Pipe23[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W, X any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[X]`**: Chains 23 operators.
*   **`Pipe24[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W, X, Y any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[Y]`**: Chains 24 operators.
*   **`Pipe25[...](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[Z]`**: Chains 25 operators.
*   **`Race[T any](sources ...Observable[T]) Observable[T]`**: Returns the Observable that first emits an item.
*   **`RandFloat64(count int) Observable[float64]`**: Emits `count` random float64 numbers.
*   **`RandIntN(n, count int) Observable[int]`**: Emits `count` random integers between 0 and `n` (exclusive).
*   **`Range(start, end int64) Observable[int64]`**: Emits a sequence of integers from `start` to `end` (inclusive).
*   **`RangeWithInterval(start, end int64, interval time.Duration) Observable[int64]`**: Emits a sequence of integers from `start` to `end` (inclusive) with a specified interval.
*   **`RangeWithStep(start, end, step float64) Observable[float64]`**: Emits a sequence of float64 numbers from `start` to `end` with a specified step.
*   **`RangeWithStepAndInterval(start, end, step float64, interval time.Duration) Observable[float64]`**: Emits a sequence of float64 numbers from `start` to `end` with a specified step and interval.
*   **`Repeat[T any](item T, count int64) Observable[T]`**: Repeats a given item `count` times.
*   **`RepeatWithInterval[T any](item T, count int64, interval time.Duration) Observable[T]`**: Repeats a given item `count` times with a specified interval.
*   **`Start[T any](cb func() T) Observable[T]`**: Creates an Observable that executes a function and emits its result.
*   **`Throw[T any](err error) Observable[T]`**: Creates an Observable that immediately emits an error.
*   **`Timer(duration time.Duration) Observable[time.Duration]`**: Creates an Observable that emits a `duration` after a specified delay.
*   **`Zip[T any](sources ...Observable[T]) Observable[[]T]`**: Combines multiple Observables by emitting an array of their latest values when all source Observables have emitted at least one value.
*   **`Zip2[A, B any](obsA Observable[A], obsB Observable[B]) Observable[lo.Tuple2[A, B]]`**: Combines two Observables into a tuple.
*   **`Zip3[A, B, C any](obsA Observable[A], obsB Observable[B], obsC Observable[C]) Observable[lo.Tuple3[A, B, C]]`**: Combines three Observables into a tuple.
*   **`Zip4[A, B, C, D any](obsA Observable[A], obsB Observable[B], obsC Observable[C], obsD Observable[D]) Observable[lo.Tuple4[A, B, C, D]]`**: Combines four Observables into a tuple.
*   **`Zip5[A, B, C, D, E any](obsA Observable[A], obsB Observable[B], obsC Observable[C], obsD Observable[D], ...) Observable[lo.Tuple5[A, B, C, D, E]]`**: Combines five Observables into a tuple.
*   **`Zip6[A, B, C, D, E, F any](obsA Observable[A], obsB Observable[B], obsC Observable[C], obsD Observable[D], ...) Observable[lo.Tuple6[A, B, C, D, E, F]]`**: Combines six Observables into a tuple.
```

--------------------------------

### Create Integer Range Observable with RxGo

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Range creates an Observable that emits a sequence of integers within a specified range [start:end). The `start` value is included, but the `end` value is excluded. It handles cases where start equals end (empty Observable) or start is greater than end (descending order).

```go
func Range(start, end int64) Observable[int64]
```

--------------------------------

### Configure and Create Connectable Observables in Go

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

Provides types and functions for creating and managing connectable observables. Connectable observables allow deferring subscription until a `Connect` method is called, enabling shared subscriptions.

```go
type ConnectableConfig struct {
	Connector func() Subject[T]
	ResetOnDisconnect bool
}

type ConnectableObservable interface {
	Connect() Subscription
	ConnectWithContext(ctx context.Context) Subscription
}

func ConnectableWithConfig[T any](source Observable[T], config ConnectableConfig[T]) ConnectableObservable[T]
func Connectable[T any](source Observable[T]) ConnectableObservable[T]
func NewConnectableObservableWithConfigAndContext[T any](subscribe func(ctx context.Context, destination Observer[T]) Teardown, ...)
func NewConnectableObservableWithConfig[T any](subscribe func(destination Observer[T]) Teardown, config ConnectableConfig[T]) ConnectableObservable[T]
func NewConnectableObservableWithContext[T any](subscribe func(ctx context.Context, destination Observer[T]) Teardown) ConnectableObservable[T]
func NewConnectableObservable[T any](subscribe func(destination Observer[T]) Teardown) ConnectableObservable[T]
```

--------------------------------

### ShareReplayWithConfig Observable Multicast with Replay and Config (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Creates a new Observable that multicasts the original and replays items to future subscribers, with customizable behavior. The source is subscribed to as long as there's at least one subscriber. Unsubscribes from the source when all subscribers have unsubscribed. Configuration options include bufferSize, ResetOnRefCountZero.

```go
func ShareReplayWithConfig[T any](bufferSize int, config ShareReplayConfig) func(Observable[T]) Observable[T]
```

--------------------------------

### Example Output for Zip Operations (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Demonstrates the output of ZipAll and ZipWith operations, showing 'Next' emissions with combined values and 'Completed' or 'Error' notifications. The examples illustrate successful data combination and error handling scenarios.

```text
Output:

Error: assert.AnError general error for testing

Next: [1 10 100]
Next: [2 11 101]
Completed

Next: {1 10 20}
Next: {2 11 21}
Next: {3 12 22}
Completed

Next: {1 10}
Next: {2 11}
Next: {3 12}
Completed

Next: {1 10 20}
Next: {2 11 21}
Completed

Next: {1 10 20 30}
Completed
```

--------------------------------

### Observer Creation Functions

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

Utility functions for creating `Observer` instances with specific handlers.

```APIDOC
## Observer Creation Functions

### Description
Provides convenient functions to create `Observer` instances, allowing you to specify handlers for `OnNext`, `OnError`, and `OnComplete` events, with or without context.

### Method
`NewObserver`, `NewObserverWithContext`, `OnNext`, `OnNextWithContext`, `OnError`, `OnErrorWithContext`, `OnComplete`, `OnCompleteWithContext`, `NoopObserver`, `PrintObserver`

### Endpoint
N/A (Function calls)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```go
// Using OnNext, OnError, OnComplete
observer1 := rxgo.NewObserver(
    rxgo.OnNext(func(v interface{}) { /* handle next */ }),
    rxgo.OnError(func(err error) { /* handle error */ }),
    rxgo.OnComplete(func() { /* handle complete */ }),
)

// Using context-aware handlers
observer2 := rxgo.NewObserverWithContext(
    rxgo.OnNextWithContext(func(ctx context.Context, v interface{}) { /* handle next with context */ }),
    rxgo.OnErrorWithContext(func(ctx context.Context, err error) { /* handle error with context */ }),
    rxgo.OnCompleteWithContext(func(ctx context.Context) { /* handle complete with context */ }),
)

// No-operation observer
noop := rxgo.NoopObserver[int]()

// Print observer (logs events)
printObs := rxgo.PrintObserver[string]()
```

### Response
#### Success Response (200)
An `Observer[T]` instance.

#### Response Example
N/A (Function returns an interface)
```

--------------------------------

### Create Float Range Observable with Step using RxGo

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

RangeWithStep creates an Observable that emits a sequence of float64 values within a range [start:end) with a specified step. The `start` value is included, but the `end` value is excluded. The step must be greater than 0. It handles empty or descending ranges.

```go
func RangeWithStep(start, end, step float64) Observable[float64]
```

--------------------------------

### Subscription and Teardown

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/index

Documentation for Subscription and Teardown types.

```APIDOC
## Subscription and Teardown

### Description
Provides types for managing subscriptions and performing cleanup actions when a subscription is disposed.

### Types

*   **`Subscription`**: An interface representing a disposable resource. Calling `Unsubscribe()` on a Subscription cancels the execution of an Observable.
*   **`Teardown`**: A function type used for cleanup actions when a subscription is disposed.

### Functions

*   **`NewSubscription(teardown Teardown) Subscription`**: Creates a new Subscription that executes the provided `teardown` function when unsubscribed.
```

--------------------------------

### Range

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Creates an Observable that emits a range of integers from start (inclusive) to end (exclusive).

```APIDOC
## Range

### Description
Creates an Observable that emits a range of integers. The range is [start:end), so `start` is emitted but not `end`. If `start` is equal to `end`, an empty Observable is returned. If `start` is greater than `end`, the emitted values are in descending order. The step is 1.

### Method
N/A (Creation Operator)

### Endpoint
N/A

### Parameters
#### Path Parameters
N/A

#### Query Parameters
N/A

#### Request Body
N/A

### Request Example
N/A

### Response
#### Success Response (Observable emits int64)
- **value** (int64) - An integer within the specified range.

#### Response Example
```
Next: 0
Next: 1
Next: 2
Next: 3
Next: 4
Completed
```
```

--------------------------------

### Create Observables from Various Sources in Go

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

Provides functions to create observables from different sources, including slices, channels, and factories. These functions simplify the process of converting existing data structures or asynchronous operations into observable streams.

```go
func FromChannel[T any](in <-chan T) Observable[T]
func FromSlice[T any](collections ...[]T) Observable[T]
func Future[T any](factory func() (T, error)) Observable[T]
func Interval(interval time.Duration) Observable[int64]
func IntervalWithInitial(initial, interval time.Duration) Observable[int64]
func Just[T any](values ...T) Observable[T]
func Of[T any](values ...T) Observable[T]
```

--------------------------------

### Create and Consume an Observable Stream in Go

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Demonstrates creating an observable stream with filtering and mapping operators, then consuming it via subscription or collecting all values. This showcases the core reactive programming paradigm in Go using the 'ro' package.

```go
package main

import (
	"fmt"
	"time"

	"github.com/samber/ro"
)

func main() {
	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!") }
	))

	// or:

	values, err := ro.Collect(observable)
	if err != nil {
		fmt.Println(err.Error())
	}
	fmt.Printf("%v\n", values)
}

```

--------------------------------

### Interval Observable (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Creates an observable that emits sequential numbers starting from 0, at a specified time interval.

```Go
func Interval(interval time.Duration) Observable[int64]
```

--------------------------------

### Concat Observables (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Concatenates multiple observables sequentially. An observable will only start emitting values after the previous one has completed.

```Go
func Concat[T any](obs ...Observable[T]) Observable[T]
```

--------------------------------

### Create and Subscribe to an Observable Stream in Go

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/index

Demonstrates creating an observable stream using ro.Pipe, applying transformations like filtering and mapping, and subscribing to consume emitted values. It shows how to handle values, errors, and completion signals. The stream generates even numbers with a delay and formats them as strings.

```go
package main

import (
	"fmt"
	"time"

	"github.com/samber/ro"
)

func main() {
	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>

}
```

--------------------------------

### Subscription Management

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

The Subscription interface represents a disposable resource, typically used to cancel an ongoing operation or release resources. It provides methods to add teardown logic or other Unsubscribable objects, and to check if the subscription has been closed.

```go
type Subscription interface {
	IsClosed() bool
	Add(teardown Teardown)
	AddUnsubscribable(unsubscribable Unsubscribable)
	Wait()
}

func NewSubscription(teardown Teardown) Subscription
```

--------------------------------

### Interval

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/index

The Interval function creates an Observable that emits sequential integers starting from 0, at a specified time interval. The Observable never completes on its own.

```APIDOC
## Interval

### Description
Creates an Observable that emits sequential integers starting from 0, at a specified time interval. This Observable never completes.

### Method
N/A (Function Signature)

### Endpoint
N/A

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```go
// Example usage (conceptual)
import "time"

intervalObservable := rxgo.Interval(1 * time.Second)

intervalObservable.Subscribe(func(ctx context.Context, i interface{}) { fmt.Printf("Tick: %v\n", i) })

// This will print "Tick: 0", "Tick: 1", "Tick: 2", etc. every second.
// To stop it, you would typically use a context or other cancellation mechanism.
```

### Response
#### Success Response (200)
N/A (Function returns an Observable)

#### Response Example
```json
// Emits 0, 1, 2, 3, ... at the specified interval.
```
```

--------------------------------

### Create Specialized Observables in Go

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

Offers functions to create specialized observables such as `Amb`, `Concat`, `Defer`, `Empty`, `Merge`, and `Never`. These cover various stream manipulation and creation scenarios, including handling multiple sources and deferred execution.

```go
func Amb[T any](sources ...Observable[T]) Observable[T]
func Concat[T any](obs ...Observable[T]) Observable[T]
func Defer[T any](factory func() Observable[T]) Observable[T]
func Empty[T any]() Observable[T]
func Merge[T any](sources ...Observable[T]) Observable[T]
func Never() Observable[struct{}]
```

--------------------------------

### Interval with Initial Delay Observable (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Creates an observable that emits sequential numbers starting from 0, after an initial delay, and then at a specified time interval.

```Go
func IntervalWithInitial(initial, interval time.Duration) Observable[int64]
```

--------------------------------

### Manage Observable Stream Sharing and Replay (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

Operators for sharing a single observable subscription among multiple observers and replaying emitted items. `Share` and `ShareReplay` provide mechanisms for efficient stream consumption.

```Go
func ShareReplayWithConfig[T any](bufferSize int, config ShareReplayConfig) func(Observable[T]) Observable[T]
func ShareReplay[T any](bufferSize int) func(Observable[T]) Observable[T]
func ShareWithConfig[T any](config ShareConfig[T]) func(Observable[T]) Observable[T]
func Share[T any]() func(Observable[T]) Observable[T]
```

--------------------------------

### IntervalWithInitial

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/index

The IntervalWithInitial function creates an Observable that emits sequential integers starting from 0, with an initial delay and then at a specified time interval. The Observable never completes on its own.

```APIDOC
## IntervalWithInitial

### Description
Creates an Observable that emits sequential integers starting from 0, with an initial delay and then at a specified time interval. This Observable never completes.

### Method
N/A (Function Signature)

### Endpoint
N/A

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```go
// Example usage (conceptual)
import "time"

intervalObservable := rxgo.IntervalWithInitial(2 * time.Second, 1 * time.Second)

intervalObservable.Subscribe(func(ctx context.Context, i interface{}) { fmt.Printf("Tick: %v\n", i) })

// This will wait 2 seconds, then print "Tick: 0", then print "Tick: 1" after another second, and so on.
// To stop it, you would typically use a context or other cancellation mechanism.
```

### Response
#### Success Response (200)
N/A (Function returns an Observable)

#### Response Example
```json
// Emits 0 after initial delay, then 1, 2, 3, ... at the specified interval.
```
```

--------------------------------

### Create Observables with ZipWith Functionality in Go

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

These functions allow zipping multiple observables together, creating a new observable that emits tuples of values from the source observables. They are generic and support zipping up to 5 additional observables.

```go
func ZipWith1[A, B any](obsB Observable[B]) func(Observable[A]) Observable[lo.Tuple2[A, B]]
func ZipWith2[A, B, C any](obsB Observable[B], obsC Observable[C]) func(Observable[A]) Observable[lo.Tuple3[A, B, C]]
func ZipWith3[A, B, C, D any](obsB Observable[B], obsC Observable[C], obsD Observable[D]) func(Observable[A]) Observable[lo.Tuple4[A, B, C, D]]
func ZipWith4[A, B, C, D, E any](obsB Observable[B], obsC Observable[C], obsD Observable[D], obsE Observable[E]) func(Observable[A]) Observable[lo.Tuple5[A, B, C, D, E]]
func ZipWith5[A, B, C, D, E, F any](obsB Observable[B], obsC Observable[C], obsD Observable[D], obsE Observable[E], ...) func(Observable[A]) Observable[lo.Tuple6[A, B, C, D, E, F]]
func ZipWith[A, B any](obsB Observable[B]) func(Observable[A]) Observable[lo.Tuple2[A, B]]
```

--------------------------------

### StartWith: Emit prefix values before source Observable values (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

StartWith emits the given values before emitting the values from the source Observable. This operator is useful for prepending data to a stream.

```Go
func StartWith[T any](prefixes ...T) func(Observable[T]) Observable[T]

// Example (Error):
// Output:
// Next: 1
// Next: 2
// Next: 3
// Error: assert.AnError general error for testing

// Example (Ok):
// Output:
// Next: 1
// Next: 2
// Next: 3
// Next: 4
// Next: 5
// Next: 6
// Completed

```

--------------------------------

### Create Connectable Observables in Go

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Functions for creating connectable observables, which defer subscription until a `Connect()` method is called. This allows for multicasting and controlling when the underlying observable starts emitting.

```go
func Connectable[T any](source Observable[T]) ConnectableObservable[T]
func ConnectableWithConfig[T any](source Observable[T], config ConnectableConfig[T]) ConnectableObservable[T]
func NewConnectableObservable[T any](subscribe func(destination Observer[T]) Teardown) ConnectableObservable[T]
func NewConnectableObservableWithConfig[T any](subscribe func(destination Observer[T]) Teardown, config ConnectableConfig[T]) ConnectableObservable[T]
func NewConnectableObservableWithConfigAndContext[T any](subscribe func(ctx context.Context, destination Observer[T]) Teardown, ...) ConnectableObservable[T]
func NewConnectableObservableWithContext[T any](subscribe func(ctx context.Context, destination Observer[T]) Teardown) ConnectableObservable[T]
```

--------------------------------

### Advanced Observable Stream Operations (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

Miscellaneous operators for advanced stream manipulation. This includes `SequenceEqual` for comparing sequences, `StartWith` for prepending elements, `SubscribeOn` for controlling subscription context, `Tail` for emitting the last element, `ThrowIfEmpty` for error handling, `ThrowOnContextCancel` for cancellation handling, and `WindowWhen` for grouping emissions.

```Go
func SequenceEqual[T comparable](obsB Observable[T]) func(Observable[T]) Observable[bool]
func StartWith[T any](prefixes ...T) func(Observable[T]) Observable[T]
func SubscribeOn[T any](bufferSize int) func(Observable[T]) Observable[T]
func Tail[T any]() func(Observable[T]) Observable[T]
func ThrowIfEmpty[T any](throw func() error) func(Observable[T]) Observable[T]
func ThrowOnContextCancel[T any]() func(Observable[T]) Observable[T]
func WindowWhen[T, B any](boundary Observable[B]) func(Observable[T]) Observable[Observable[T]]
```

--------------------------------

### Buffer observable items by time duration (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

BufferWithTime buffers items emitted by an Observable for a specified duration. It emits the buffer and starts a new one. Buffering stops if the source Observable completes or errors.

```Go
func BufferWithTime[T any](duration time.Duration) func(Observable[T]) Observable[[]T]
```

--------------------------------

### Observer Interface and Functions

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/index

Documentation for the Observer interface and related functions for creating and managing observers.

```APIDOC
## Observer Interface and Functions

### Description
Defines the Observer interface and provides functions to create and configure observers, which handle emitted values, errors, and completion signals from an Observable.

### Interface

*   **`Observer[T any]`**: An interface with methods `OnNext(value T)`, `OnError(err error)`, and `OnComplete()`. (Note: Context-aware versions are also available).

### Functions

*   **`NewObserver[T any](onNext func(value T), onError func(err error), onComplete func()) Observer[T]`**: Creates a new Observer with provided handlers.
*   **`NewObserverWithContext[T any](onNext func(ctx context.Context, value T), onError func(ctx context.Context, err error), onComplete func(ctx context.Context)) Observer[T]`**: Creates a new context-aware Observer.
*   **`NoopObserver[T any]() Observer[T]`**: Returns an Observer that does nothing.
*   **`OnComplete[T any](onComplete func()) Observer[T]`**: Creates an Observer that only handles the completion signal.
*   **`OnCompleteWithContext[T any](onComplete func(ctx context.Context)) Observer[T]`**: Creates a context-aware Observer that only handles the completion signal.
*   **`OnError[T any](onError func(err error)) Observer[T]`**: Creates an Observer that only handles error signals.
*   **`OnErrorWithContext[T any](onError func(ctx context.Context, err error)) Observer[T]`**: Creates a context-aware Observer that only handles error signals.
*   **`OnNext[T any](onNext func(value T)) Observer[T]`**: Creates an Observer that only handles next value signals.
*   **`OnNextWithContext[T any](onNext func(ctx context.Context, value T)) Observer[T]`**: Creates a context-aware Observer that only handles next value signals.
*   **`PrintObserver[T any]() Observer[T]`**: Creates an Observer that prints emitted values, errors, and completion signals.
```

--------------------------------

### Clamp Operator for RxGo Observables

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

The Clamp operator emits values within the inclusive lower and upper bounds. It ensures that emitted numbers stay within a defined range. The Play link provides an example.

```go
func Clamp[T constraints.Numeric](lower, upper T) func(Observable[T]) Observable[T]
// Play: https://go.dev/play/p/fu8O-BixXPM
```

--------------------------------

### ShareWithConfig Observable Multicast with Config (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Creates a new Observable that multicasts the original, with customizable behavior. The source is subscribed to as long as there's at least one subscriber. Unsubscribes from the source when all subscribers have unsubscribed. Configuration options include Connector, ResetOnError, ResetOnComplete, ResetOnRefCountZero.

```go
func ShareWithConfig[T any](config ShareConfig[T]) func(Observable[T]) Observable[T]
```

--------------------------------

### Buffer observable items by count (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

BufferWithCount buffers items emitted by an Observable until the buffer reaches a specified size. It then emits the buffer and starts a new one. Buffering stops if the source Observable completes or errors.

```Go
func BufferWithCount[T any](size int) func(Observable[T]) Observable[[]T]
```

--------------------------------

### Create and Manage Notifications in Go

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

Provides types and functions for creating and managing notifications emitted by observables. Notifications can represent a next value, an error, or a completion signal.

```go
type Notification struct {
	Err error
	Kind Kind
	Value T
}

func NewNotificationComplete[T any]() Notification[T]
func NewNotificationError[T any](err error) Notification[T]
func NewNotificationNext[T any](value T) Notification[T]
func (n Notification[T]) String() string
```

--------------------------------

### Create Observer with Contextual OnComplete (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Creates a partial Observer that only implements the onComplete callback, which receives a context. Other notifications are ignored, and errors are silenced.

```go
func OnCompleteWithContext[T any](onComplete func(ctx context.Context)) Observer[T]
```

--------------------------------

### Do: Perform side effects on observable events (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

The Do operator is an alias for the Tap operator. It allows performing side effects for onNext, onError, and onComplete events in an observable stream. A Play link is provided for interactive examples.

```Go
func Do[T any](onNext func(value T), onError func(err error), onComplete func()) func(Observable[T]) Observable[T]

// Play: https://go.dev/play/p/s_BSHgxdjUR
```

--------------------------------

### Range and Interval Observables (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

These functions create observables that emit a sequence of numbers within a specified range. `Range` emits integers, while `RangeWithInterval` and `RangeWithStep` variants allow for custom intervals and steps.

```Go
func Range(start, end int64) Observable[int64]
func RangeWithInterval(start, end int64, interval time.Duration) Observable[int64]
func RangeWithStep(start, end, step float64) Observable[float64]
func RangeWithStepAndInterval(start, end, step float64, interval time.Duration) Observable[float64]
```

--------------------------------

### Emit Absolute Values of Observable Elements (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

The Abs function creates an Observable that emits the absolute values of the elements emitted by a source Observable of float64. It handles both successful emissions and errors, as demonstrated in the provided examples.

```go
func Abs() func(Observable[float64]) Observable[float64]

```

--------------------------------

### ShareReplay Configuration

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

Configuration options for sharing and replaying an Observable's emissions.

```APIDOC
## ShareReplay Configuration

### Description
The `ShareReplayConfig` struct is used to configure the behavior of sharing an Observable while replaying its emissions. It specifically includes an option to reset the replay behavior when the reference count drops to zero.

### Method
`ShareReplayConfig` struct fields (`ResetOnRefCountZero`)

### Endpoint
N/A (Struct definition)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```go
// Configure share replay behavior
shareReplayConfig := rxgo.ShareReplayConfig{
    ResetOnRefCountZero: true,
}

// Apply share replay logic to an observable (assuming a hypothetical .ShareReplay() operator)
// observable.ShareReplay(shareReplayConfig).Subscribe(...)
```

### Response
#### Success Response (200)
N/A (This is a struct definition, not an endpoint response)

#### Response Example
N/A
```

--------------------------------

### Buffer observable items by time or count (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

BufferWithTimeOrCount buffers items emitted by an Observable until a specified time or count is reached. It emits the buffer and starts a new one. Buffering stops if the source Observable completes or errors.

```Go
func BufferWithTimeOrCount[T any](size int, duration time.Duration) func(Observable[T]) Observable[[]T]
```

--------------------------------

### DistinctBy: Suppress duplicates based on key selector (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

The DistinctBy operator suppresses duplicate items in an Observable based on a key selector function. It takes a key selector that determines uniqueness. The provided examples illustrate scenarios with and without errors.

```Go
func DistinctBy[T any, K comparable](keySelector func(item T) K) func(Observable[T]) Observable[T]

// Example (Ok) Output:
// Next: {1 John}
// Next: {2 Jane}
// Next: {3 Jim}
// Completed

// Example (Error) Output:
// Next: {1 John}
// Next: {2 Jane}
// Error: assert.AnError general error for testing
```

--------------------------------

### Create Timed Integer Range Observable with RxGo

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

RangeWithInterval creates an Observable that emits integers within a range [start:end) at a specified time interval. The first value is emitted after the initial interval. It handles empty or descending ranges similar to `Range`.

```go
func RangeWithInterval(start, end int64, interval time.Duration) Observable[int64]
```

--------------------------------

### RWMutexWithLock

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3.0/internal/xsync

Implementation of an RWMutex using a standard Go read-write mutex.

```APIDOC
## Type RWMutexWithLock

### Description
RWMutexWithLock is a read-write mutex with a standard read-write mutex.

### Methods
- **NewRWMutexWithLock()** *RWMutexWithLock: Creates a new read-write mutex with a standard read-write mutex.
- **Lock()**: Locks the mutex for writing.
- **RLock()**: Locks the mutex for reading.
- **RUnlock()**: Unlocks the mutex for reading.
- **TryLock()** bool: Tries to lock the mutex for writing.
- **TryRLock()** bool: Tries to lock the mutex for reading.
- **Unlock()**: Unlocks the mutex for writing.
```

--------------------------------

### Buffer observable items based on a boundary observable (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

BufferWhen buffers items emitted by an Observable until a second Observable emits an item. It then emits the buffer and starts a new one. Buffering stops if the source Observable completes or errors, or if the boundary Observable completes.

```Go
func BufferWhen[T, B any](boundary Observable[B]) func(Observable[T]) Observable[[]T]
```

--------------------------------

### RWMutexWithoutLock Methods

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3.0/internal/xsync

Methods for the RWMutexWithoutLock type, including creation, locking, and unlocking.

```APIDOC
## RWMutexWithoutLock Methods

### Description
Represents a read-write mutex that does not perform any actual locking.

### Methods

#### NewRWMutexWithoutLock

- **Description**: Creates a new instance of RWMutexWithoutLock.
- **Method**: (Constructor function, no explicit HTTP method)
- **Endpoint**: N/A

#### Lock

- **Description**: This method does nothing.
- **Method**: (Implicit receiver method, no explicit HTTP method)
- **Endpoint**: N/A

#### RLock

- **Description**: This method does nothing.
- **Method**: (Implicit receiver method, no explicit HTTP method)
- **Endpoint**: N/A

#### RUnlock

- **Description**: This method does nothing.
- **Method**: (Implicit receiver method, no explicit HTTP method)
- **Endpoint**: N/A

#### TryLock

- **Description**: Always returns true, indicating a successful lock acquisition (though no actual lock is held).
- **Method**: (Implicit receiver method, no explicit HTTP method)
- **Endpoint**: N/A

#### TryRLock

- **Description**: Always returns true, indicating a successful read lock acquisition (though no actual lock is held).
- **Method**: (Implicit receiver method, no explicit HTTP method)
- **Endpoint**: N/A

#### Unlock

- **Description**: This method does nothing.
- **Method**: (Implicit receiver method, no explicit HTTP method)
- **Endpoint**: N/A
```

--------------------------------

### TapOnSubscribeWithContext: Side effects on subscribe with context in Go

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

TapOnSubscribeWithContext enables side effects when the source Observable is subscribed to, including the context, without modifying emitted items. It mirrors the source Observable and forwards emissions to the provided observer, suitable for context-aware subscription setup.

```go
func TapOnSubscribeWithContext[T any](onSubscribe func(ctx context.Context)) func(Observable[T]) Observable[T]
```

--------------------------------

### FlatMapIWithContext

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/index

Transforms items into Observables with context, index, and flattens emissions.

```APIDOC
## FlatMapIWithContext

### Description
Transforms the items emitted by an Observable into Observables, then flattens the emissions from those into a single Observable, providing context, item, and index.

### Method
`func FlatMapIWithContext[T, R any](project func(ctx context.Context, item T, index int64) Observable[R]) func(Observable[T]) Observable[R]`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
None

#### Response Example
None
```

--------------------------------

### Create Timed Float Range Observable with Step using RxGo

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

RangeWithStepAndInterval creates an Observable that emits float64 values within a range [start:end) with a specified step and at a defined time interval. The first value is emitted after the initial interval. The step must be greater than 0.

```go
func RangeWithStepAndInterval(start, end, step float64, interval time.Duration) Observable[float64]
```

--------------------------------

### Repeat Item Observable (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

Creates an observable that repeatedly emits a given item a specified number of times. `Repeat` emits immediately, while `RepeatWithInterval` introduces a delay between emissions.

```Go
func Repeat[T any](item T, count int64) Observable[T]
func RepeatWithInterval[T any](item T, count int64, interval time.Duration) Observable[T]
```

--------------------------------

### ElementAt: Get nth element from Observable (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

ElementAt emits only the nth item emitted by an Observable. If the source Observable emits fewer than n items, ElementAt will emit an error. This function is useful for accessing a specific element in a sequence without processing the entire sequence if the element is found early.

```Go
func ElementAt[T any](nth int) func(Observable[T]) Observable[T]
```

--------------------------------

### ElementAtOrDefault: Get nth element or fallback from Observable (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

ElementAtOrDefault emits only the nth item emitted by an Observable. If the source Observable emits fewer than n items, ElementAtOrDefault will emit a fallback value. This is useful when you need to access a specific element but want to provide a default if it doesn't exist, preventing potential errors.

```Go
func ElementAtOrDefault[T any](nth int64, fallback T) func(Observable[T]) Observable[T]
```

--------------------------------

### Create Observer with Contextual OnNext (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Creates a partial Observer that only implements the onNext callback, which receives a context. Other notifications are ignored, and errors are silenced.

```go
func OnNextWithContext[T any](onNext func(ctx context.Context, value T)) Observer[T]
```

--------------------------------

### ShareReplay Observable Multicast with Replay (Go)

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3

Creates a new Observable that multicasts the original and replays a specified number of items to future subscribers. The source is subscribed to as long as there's at least one subscriber. Unsubscribes from the source when all subscribers have unsubscribed. This is an alias for ShareReplayWithConfig with default settings.

```go
func ShareReplay[T any](bufferSize int) func(Observable[T]) Observable[T]
```

--------------------------------

### Variables

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/index

Global variables and error definitions in RxGo.

```APIDOC
## Variables

### Error Variables

This section lists common error variables defined in RxGo, used for specific error conditions within operators.

- `ErrRangeWithStepWrongStep`: Indicates an invalid step value for `RangeWithStep`.
- `ErrRangeWithStepAndIntervalWrongStep`: Indicates an invalid step value for `RangeWithStepAndInterval`.
- `ErrFirstEmpty`: Indicates an empty observable when `First` is called.
- `ErrLastEmpty`: Indicates an empty observable when `Last` is called.
- `ErrHeadEmpty`: Alias for `ErrFirstEmpty`.
- `ErrTailEmpty`: Alias for `ErrLastEmpty`.
- `ErrTakeWrongCount`: Indicates an invalid count for `Take`.
- `ErrTakeLastWrongCount`: Indicates an invalid count for `TakeLast`.
- `ErrSkipWrongCount`: Indicates an invalid count for `Skip`.
- `ErrSkipLastWrongCount`: Indicates an invalid count for `SkipLast`.
- `ErrElementAtWrongNth`: Indicates an invalid index for `ElementAt`.
- `ErrElementAtNotFound`: Indicates the element was not found at the specified index.
- `ErrElementAtOrDefaultWrongNth`: Indicates an invalid index for `ElementAtOrDefault`.
- `ErrRepeatWrongCount`: Indicates an invalid count for `Repeat`.
- `ErrRepeatWithIntervalWrongCount`: Indicates an invalid count for `RepeatWithInterval`.
- `ErrRepeatWithWrongCount`: Indicates an invalid count for `RepeatWith`.
- `ErrBufferWithCountWrongSize`: Indicates an invalid size for `BufferWithCount`.
- `ErrBufferWithTimeWrongDuration`: Indicates an invalid duration for `BufferWithTime`.
- `ErrBufferWithTimeOrCountWrongSize`: Indicates an invalid size for `BufferWithTimeOrCount`.
- `ErrBufferWithTimeOrCountWrongDuration`: Indicates an invalid duration for `BufferWithTimeOrCount`.
- `ErrClampLowerLessThanUpper`: Indicates the lower bound is greater than the upper bound for `Clamp`.
- `ErrToChannelWrongSize`: Indicates an invalid buffer size for `ToChannel`.
- `ErrPoolWrongSize`: Indicates an invalid pool size for `Pool`.
- `ErrSubscribeOnWrongBufferSize`: Indicates an invalid buffer size for `SubscribeOn`.
- `ErrObserveOnWrongBufferSize`: Indicates an invalid buffer size for `ObserveOn`.
- `ErrDetachOnWrongMode`: Indicates an unexpected mode for `DetachOn`.
- `ErrUnicastSubjectConcurrent`: Indicates a `UnicastSubject` was subscribed to by multiple observers.
- `ErrConnectableObservableMissingConnectorFactory`: Indicates a missing connector factory for `ConnectableObservable`.

### Handler Variables

- `OnUnhandledError`: A function called when an error is emitted and no error handler is registered. Defaults to `IgnoreOnUnhandledError`.
- `OnDroppedNotification`: A function called when a notification is emitted and no notification handler is registered. Defaults to `IgnoreOnDroppedNotification`.
```

--------------------------------

### Subscriber Interface and Functions

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/index

Documentation for the Subscriber interface and functions for creating and managing subscribers.

```APIDOC
## Subscriber Interface and Functions

### Description
Subscribers are the consumers of events emitted by Observables. This section details the Subscriber interface and functions for creating different types of subscribers, including safe and unsafe versions.

### Interface

*   **`Subscriber[T any]`**: An interface that extends `Observer[T]` and `Subscription`.

### Functions

*   **`NewEventuallySafeSubscriber[T any](destination Observer[T]) Subscriber[T]`**: Creates a subscriber that ensures all callbacks are executed on the same goroutine.
*   **`NewSafeSubscriber[T any](destination Observer[T]) Subscriber[T]`**: Creates a subscriber that ensures callbacks are executed safely, preventing potential race conditions.
*   **`NewSubscriber[T any](destination Observer[T]) Subscriber[T]`**: Creates a basic subscriber.
*   **`NewSubscriberWithConcurrencyMode[T any](destination Observer[T], mode ConcurrencyMode) Subscriber[T]`**: Creates a subscriber with a specified concurrency mode.
*   **`NewUnsafeSubscriber[T any](destination Observer[T]) Subscriber[T]`**: Creates an unsafe subscriber that does not perform any safety checks, potentially leading to race conditions.
```

--------------------------------

### Share Configuration

Source: https://pkg.go.dev/github.com/samber/ro@v0.3.0/github.com/samber/ro%40v0.3_tab=versions

Configuration options for sharing an Observable among multiple subscribers.

```APIDOC
## Share Configuration

### Description
The `ShareConfig` struct provides options for configuring how an Observable is shared among subscribers. This includes defining a `Connector` function and specifying conditions under which the shared Observable should reset.

### Method
`ShareConfig` struct fields (`Connector`, `ResetOnComplete`, `ResetOnError`, `ResetOnRefCountZero`)

### Endpoint
N/A (Struct definition)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```go
// Configure sharing behavior
shareConfig := rxgo.ShareConfig{
    Connector: func() rxgo.Subject[int] { return rxgo.NewPublishSubject[int]() },
    ResetOnComplete: true,
    ResetOnError: false,
    ResetOnRefCountZero: true,
}

// Apply share logic to an observable (assuming a hypothetical .Share() operator)
// observable.Share(shareConfig).Subscribe(...)
```

### Response
#### Success Response (200)
N/A (This is a struct definition, not an endpoint response)

#### Response Example
N/A
```
Tokens:
Raw