ro

https://ro.samber.dev/docs/
Admin
ro is a Go library for streams and reactive programming that enables composable asynchronous data...
Tokens:262,114
Snippets:1,616
Trust Score:4.8
Update:7 months ago
Show doc for...
Tokens:
Raw
### Install samber/ro Library

Source: https://ro.samber.dev/docs/getting-started

Installs the samber/ro reactive programming library for Go. Requires Go 1.18+.

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

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

### Install ro/plugins/regexp

Source: https://ro.samber.dev/docs/plugins/regexp

Installs the regexp sub-package for the ro library using go get.

```go
go get -u github.com/samber/ro/plugins/regepx
```

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

### Install ro/plugins/proc

Source: https://ro.samber.dev/docs/plugins/proc

Installs the proc sub-package for the ro library using go get. This command fetches and installs the specified package and its dependencies.

```bash
go get -u github.com/samber/ro/plugins/proc
```

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

### Install ro/plugins/template

Source: https://ro.samber.dev/docs/plugins/template

Instructions to install the template sub-package for ro using go get.

```go
go get -u github.com/samber/ro/plugins/template  

```

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

### Install ro/plugins/bytes

Source: https://ro.samber.dev/docs/plugins/bytes

Installs the bytes sub-package of the ro library using go get. This command fetches and installs the specified package and its dependencies.

```go
go get -u github.com/samber/ro/plugins/bytes
```

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

### Hot Observables with ro Framework

Source: https://ro.samber.dev/docs/getting-started

Demonstrates the concept of hot observables in the ro framework, where multiple subscribers share a single stream. This example shows how to create a hot observable from a cold one using `Connectable` and how subscribers share the same sequence.

```go
// Create a hot observable from a cold one  
hot := ro.Connectable(ro.Just(1, 2, 3))

// Both subscribers share the same sequence simultaneously  
sub1 := hot.Subscribe(ro.OnNext(func(x int) { fmt.Println("Sub1:", x) }))
sub2 := hot.Subscribe(ro.OnNext(func(x int) { fmt.Println("Sub2:", x) }))

// Start subscription  
subscription := connectable.Connect()
```

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

### Install ro/plugins/observability/log

Source: https://ro.samber.dev/docs/plugins/logger-log

Installs the ro observability log sub-package using go get.

```bash
go get -u github.com/samber/ro/plugins/observability/log
```

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

### Install ratelimit/ulule Plugin

Source: https://ro.samber.dev/docs/plugins/ratelimit-ulule

Instructions to install the ratelimit/ulule sub-package using go get. This is the first step to using the rate limiting operators provided by this plugin.

```go
go get -u github.com/samber/ro/plugins/ratelimit/ulule
```

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

### Resource Management with ro Framework Subscriptions

Source: https://ro.samber.dev/docs/getting-started

Highlights the best practice of managing resources by cleaning them up to prevent memory leaks. This example shows how to use `defer subscription.Unsubscribe()` to ensure resources are released when the stream processing is complete.

```go
// Clean up resources when done  
subscription := stream.Subscribe(observer)
defer subscription.Unsubscribe()
```

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

### Install ro/plugins/ratelimit/native

Source: https://ro.samber.dev/docs/plugins/ratelimit-native

Installs the native ratelimit plugin for the ro library using the go get command. This is the first step to using the native ratelimit operators.

```bash
go get -u github.com/samber/ro/plugins/ratelimit/native
```

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

### Install ro/plugins/stdio

Source: https://ro.samber.dev/docs/plugins/stdio

Imports the stdio sub-package into your Go project using the go get command. This is the first step before using any functionalities from this plugin.

```go
go get -u github.com/samber/ro/plugins/stdio
```

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

### Install ro/plugins/strconv

Source: https://ro.samber.dev/docs/plugins/strconv

Import the strconv sub-package into your Go project using the go get command.

```go
go get -u github.com/samber/ro/plugins/strconv
```

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

### Install ro/plugins/strings

Source: https://ro.samber.dev/docs/plugins/strings

Instructions to import the 'strings' sub-package into your Go project using 'go get'. This is the first step to using any of the string operators.

```go
go get -u github.com/samber/ro/plugins/strings  

```

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

### Install ro/plugins/cron

Source: https://ro.samber.dev/docs/plugins/cron

Installs the cron sub-package for the ro library using go get. This is the first step to using cron-related functionalities.

```go
go get -u github.com/samber/ro/plugins/cron
```

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

### Install ro/plugins/observability/sentry

Source: https://ro.samber.dev/docs/plugins/logger-sentry

Installs the Sentry observability plugin for the ro library using go get.

```bash
go get -u github.com/samber/ro/plugins/observability/sentry
```

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

### Install ro/plugins/observability/slog in Go

Source: https://ro.samber.dev/docs/plugins/logger-slog

This snippet shows how to install the ro/plugins/observability/slog sub-package using the go get command. This is the first step to using the structured logging features provided by the package.

```go
go get -u github.com/samber/ro/plugins/observability/slog
```

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

### Using Pipeline Operators for Composable Code in ro

Source: https://ro.samber.dev/docs/getting-started

Illustrates the best practice of using pipeline operators in the ro framework for creating clean and reusable code. This example shows how to chain operators like `Filter`, `Map`, and `Retry` into a composable pipeline.

```go
// Good: Composable pipeline  
pipeline := ro.Pipe3(
	source,
	ro.Filter(predicate),
	ro.Map(transformer),
	ro.Retry(3),
)

// Reusable pipeline  
result1 := pipeline(stream1)
result2 := pipeline(stream2)
```

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

### Install ro http/client plugin

Source: https://ro.samber.dev/docs/plugins/http-client

Installs the http/client sub-package for the ro library using go get. This is the first step to using the HTTP client operators.

```go
go get -u github.com/samber/ro/plugins/http/client
```

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

### Install ozzo-validation plugin for ro

Source: https://ro.samber.dev/docs/plugins/ozzo-validation

Installs the ozzo-validation sub-package for the ro library using go get. This command fetches and installs the necessary package to enable ozzo-validation capabilities within your ro project.

```go
go get -u github.com/samber/ro/plugins/ozzo/ozzo-validation
```

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

### Install encoding/yaml Sub-package (Go)

Source: https://ro.samber.dev/docs/plugins/encoding-yaml

This snippet shows how to install the encoding/yaml sub-package using the go get command. This is the first step to using the YAML operators in your Go project.

```go
go get -u github.com/samber/ro/plugins/yaml  

```

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

### BufferWithCount: Practical example for batch processing (Go)

Source: https://ro.samber.dev/docs/operator/transformation

This example demonstrates using BufferWithCount to create batches of items, which are then processed. Each batch is mapped to a single value (the sum of its elements), showcasing a common use case for batch processing.

```Go
package main

import (
	"fmt"
	"time"

	"github.com/reactivex/rxjs/go/rxjs"
)

func main() {
	obs := rxjs.Pipe[int, int](
	 rxjs.Just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10),
	 rxjs.BufferWithCount[int](4),
	 rxjs.Map(func(batch []int) int {
		// Process batch of items
		sum := 0
		for _, item := range batch {
			sum += item
		}
		return sum
	}),
	)

	ssub := obs.Subscribe(rxjs.PrintObserver[int]())
	defer sub.Unsubscribe()

	// Next: 10 (1+2+3+4)
	// Next: 22 (5+6+7+8)
	// Next: 19 (9+10)
	// Completed
}

```

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

### Install exp/simd sub-package

Source: https://ro.samber.dev/docs/plugins/exp-simd

Installs the exp/simd sub-package for use in your Go project. This is the first step to utilizing SIMD operators.

```go
go get -u github.com/samber/ro/plugins/exp/simd
```

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

### Cold Observables in ro Framework

Source: https://ro.samber.dev/docs/getting-started

Explains and demonstrates cold observables in the ro framework. Cold observables, which are the default, provide each subscriber with an independent stream, meaning consumption starts upon subscription for each individual subscriber.

```go
cold := ro.Just(1, 2, 3)

// Each subscriber sees the same values independently. Consumption starts on subscription.
cold.Subscribe(ro.OnNext(func(x int) { fmt.Println("Sub1:", x) }))
cold.Subscribe(ro.OnNext(func(x int) { fmt.Println("Sub2:", x) }))
```

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

### Install ro/plugins/encoding/base64

Source: https://ro.samber.dev/docs/plugins/encoding-base64

Installs the base64 sub-package for the ro library. This is the first step before using any encoding or decoding operators.

```go
go get -u github.com/samber/ro/plugins/encoding/base64
```

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

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

Source: https://ro.samber.dev/docs/getting-started

Demonstrates creating a simple observable stream using ro.Interval and ro.Take, transforming values with ro.Map, and subscribing to process emitted strings. The stream emits 5 ticks at 1-second intervals.

```go
package main

import (
    "fmt"
    "time"

    "github.com/samber/ro"
)

func main() {
    // Create a simple stream
    observable := ro.Pipe2(
        ro.Interval(1 * time.Second),
        ro.Take[int64](5),
        ro.Map(func(x int64) string {
            return fmt.Sprintf("Tick: %d", x)
        }),
    )

    // Subscribe and print values
    subscription := observable.Subscribe(ro.OnNext(func(s string) {
        fmt.Println(s)
    }))

    // Wait for completion
    subscription.Wait()
}
```

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

### Install encoding/json/v2 Plugin

Source: https://ro.samber.dev/docs/plugins/encoding-json-v2

This snippet shows how to import the encoding/json/v2 sub-package into your Go project using the go get command. This is the first step to using the JSON v2 operators.

```go
go get -u github.com/samber/ro/plugins/encoding/json/v2
```

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

### Manage Subscriptions with Observer Functions in Go

Source: https://ro.samber.dev/docs/getting-started

Illustrates how to subscribe to an observable in Go using `ro.NewObserver` to handle next, error, and completed events. It also shows how to cancel a subscription using `Unsubscribe()`.

```go
subscription := observable.Subscribe(ro.NewObserver(
    func(value int) {           // OnNext
        fmt.Println("Received:", value)
    },
    func(err error) {           // OnError
        fmt.Println("Error:", err)
    },
    func() {                    // OnCompleted
        fmt.Println("Done!")
    },
))

// Cancel subscription if needed
subscription.Unsubscribe()
```

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

### Create Observables from Values and Slices in Go

Source: https://ro.samber.dev/docs/getting-started

Shows how to create Observables in Go using the samber/ro library. `ro.Just` creates an observable from individual values, while `ro.FromSlice` creates one from a slice of elements.

```go
// Create from values
numbers := ro.Just(1, 2, 3, 4, 5)

// Create from a slice
letters := ro.FromSlice([]string{"a", "b", "c"})
```

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

### Install ro/plugins/encoding/json

Source: https://ro.samber.dev/docs/plugins/encoding-json

Installs the encoding/json sub-package for use in your Go project. This is the first step before utilizing its operators.

```go
go get -u github.com/samber/ro/plugins/encoding/json
```

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

### Install testify plugin for ro

Source: https://ro.samber.dev/docs/plugins/testify

This command installs the testify sub-package for the ro library, which provides operators for creating test assertions for observables.

```go
go get -u github.com/samber/ro/plugins/testify
```

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

### Install samber/psi Plugin

Source: https://ro.samber.dev/docs/plugins/samber-psi

Command to install the samber/psi sub-package for ro. This is the first step to using PSI operators in your project.

```bash
go get -u github.com/samber/ro/plugins/samber/psi
```

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

### Install samber/hot Plugin

Source: https://ro.samber.dev/docs/plugins/samber-hot

Installs the `samber/hot` plugin for use with the `ro` package. This is the initial step before utilizing any of the hot operators.

```go
go get -u github.com/samber/ro/plugins/samber/hot
```

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

### Real-time Data Streaming Example

Source: https://ro.samber.dev/docs/operator/sink

Demonstrates how to simulate and consume real-time data streams using Observables and channels.

```APIDOC
## GET /api/stream/realtime

### Description
Simulates and streams real-time sensor data. The stream is limited to a specific number of readings and consumed via a channel.

### Method
GET

### Endpoint
/api/stream/realtime

### Parameters
#### Query Parameters
- **count** (int) - Optional - The number of readings to take from the stream. Defaults to 5.
- **bufferSize** (int) - Optional - The buffer size for the channel. Defaults to 1.

### Request Example
```
GET /api/stream/realtime?count=5&bufferSize=1
```

### Response
#### Success Response (200)
- **stream** (<-chan Event) - A read-only channel emitting simulated sensor events.

#### Response Example
```json
{
  "stream": "channel_reference"
}
```

### Event Structure
```json
{
  "SensorID": "string",
  "Temperature": "float64",
  "Humidity": "float64",
  "Timestamp": "time.Time"
}
```
```

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

### ShareReplayWithConfig with Buffer Management (Go)

Source: https://ro.samber.dev/docs/operator/connectable

This example illustrates ShareReplayWithConfig's buffer management capabilities. It creates a data stream with varying rates and simulates periodic subscribers, showing how each batch replays the last specified number of values from when they subscribed.

```Go
package main

import (
	"fmt"
	"time"
	ro "github.com/reactivex/rxgo/v2"
)

func main() {
	config := ro.ShareReplayConfig{
		ResetOnRefCountZero: true,
	}

	// Stream with varying data rates
	dataStream := ro.Pipe[int64, int64](
		ro.Interval(100 * time.Millisecond),
		ro.Take[int64](20),
		ro.ShareReplayWithConfig[int64](5, config), // Keep last 5 values
	)

	// Simulate periodic subscribers
	for i := 0; i < 4; i++ {
		go func(batch int) {
			time.Sleep(time.Duration(batch) * 300 * time.Millisecond)
			fmt.Printf("Batch %d subscribing:\n", batch+1)
			sub := dataStream.Subscribe(ro.OnNext(func(value int64) {
				fmt.Printf("  B%d: %d\n", batch+1, value)
			}))
			time.Sleep(400 * time.Millisecond)
			sub.Unsubscribe()
			fmt.Printf("Batch %d done\n", batch+1)
		}(i)
	}

	time.Sleep(1500 * time.Millisecond)

	// Shows how each batch gets replayed last 5 values
	// from when they subscribed
}

```

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

### Creating Custom Operators with ro Framework

Source: https://ro.samber.dev/docs/getting-started

Shows how to create reusable custom operators in Go using the ro framework. This example defines a `Square` operator that squares integer values in an observable stream, demonstrating encapsulation of common transformations.

```go
// Custom operator that squares numbers  
func Square[T constraints.Integer](observable ro.Observable[T]) ro.Observable[T] {
	return ro.Map(func(x T) T {
		return x * x
	})(observable)
}

func main() {
	result := Square(ro.Just(1, 2, 3, 4, 5))

	result.Subscribe(ro.OnNext(func(x int) {
		fmt.Println(x)  // 1, 4, 9, 16, 25
	}))
```

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

### Install ro signal plugin

Source: https://ro.samber.dev/docs/plugins/signal

This command installs the signal sub-package for the ro library using go get.

```bash
go get -u github.com/samber/ro/plugins/signal
```

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

### Install Hyperloglog Plugin

Source: https://ro.samber.dev/docs/plugins/hyperloglog

This snippet shows how to install the hyperloglog sub-package for the ro library using go get.

```go
go get -u github.com/samber/ro/plugins/hyperloglog
```

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

### Memory and CPU Profiling with Go Tooling

Source: https://ro.samber.dev/docs/troubleshooting/debugging

Shows how to generate CPU and memory profiles for Go applications and tests using the 'go test' command. It also includes instructions on how to analyze these profiles using the 'go tool pprof' command to identify performance bottlenecks.

```bash
# Generate CPU profile
go test -cpuprofile=cpu.prof -bench=.

# Generate memory profile
go test -memprofile=mem.prof -bench=.

# Analyze profiles
go tool pprof cpu.prof
go tool pprof mem.prof
```

```go
func BenchmarkYourOperator(b *testing.B) {
    source := ro.Just(1, 2, 3, 4, 5)
    operator := YourOperator()

    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        values, err := ro.Collect(operator(source))
        if err != nil {
            b.Fatal(err)
        }
        _ = values
    }
}
```

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

### NewPrompt

Source: https://ro.samber.dev/docs/plugins/stdio

Creates an observable that prompts the user for input.

```APIDOC
## NewPrompt

### Description
Creates an observable that prompts the user for input.

### Method
N/A (Function)

### Endpoint
N/A

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

#### Query Parameters
N/A

#### Request Body
N/A

### Request Example
```go
import (
    "github.com/samber/ro"
    rostdio "github.com/samber/ro/plugins/stdio"
)

obs := rostdio.NewPrompt("Enter your name: ")

sub := obs.Subscribe(ro.NewObserver(
    func(input []byte) {
        println("You entered:", string(input))
    },
    func(err error) {
        println("Error:", err.Error())
    },
    func() {
        println("Input completed")
    },
))
def sub.Unsubscribe()

// (User sees prompt: "Enter your name: ")
// User types: "Alice"
// You entered: Alice
```

### Response
#### Success Response (200)
N/A (Observable emits byte slices upon user input)

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

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

### Runtime Tracing in Go

Source: https://ro.samber.dev/docs/troubleshooting/debugging

Demonstrates how to enable and use Go's runtime tracing capabilities to observe the execution of reactive code. The 'trace.Start' and 'trace.Stop' functions are used to capture trace data, which can then be analyzed to understand program behavior.

```go
func EnableTracing() {
    // Enable runtime tracing
    trace.Start(os.Stdout)
    defer trace.Stop()

    // Your reactive code here
    source := ro.Just(1, 2, 3, 4, 5)
    obs := yourOperator()(source)

    values, err := ro.Collect(obs)
    if err != nil {
        log.Printf("Error: %v", err)
    }
    log.Printf("Result: %v", values)
}
```

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

### Install ro sort plugin

Source: https://ro.samber.dev/docs/plugins/sort

Installs the sort sub-package for the ro library using go get. This is the first step to using any of the sort operators.

```go
go get -u github.com/samber/ro/plugins/sort  

```

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

### Watch Swap Device Statistics with ro/plugins/proc

Source: https://ro.samber.dev/docs/plugins/proc

Creates a watcher for swap device statistics that updates at a specified interval. It subscribes to the watcher and prints swap device statistics using ro.PrintObserver. Requires the 'github.com/shirou/gopsutil/v4/mem' package.

```go
import (
    "time"

    "github.com/samber/ro"
    roproc "github.com/samber/ro/plugins/proc"
    "github.com/shirou/gopsutil/v4/mem"
)

obs := roproc.NewSwapDeviceWatcher(2 * time.Second)

sub := obs.Subscribe(ro.PrintObserver[*mem.SwapDevice]())
defer sub.Unsubscribe()

// Next: &{Name: /dev/sda1, Used: 1073741824, Free: 3221225472, ...}
// Next: &{Name: /dev/sda2, Used: 2147483648, Free: 2147483648, ...}
// ...
```

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

### Install fsnotify Plugin for Go

Source: https://ro.samber.dev/docs/plugins/fsnotify

Installs the fsnotify sub-package for the ro project using go get. This is the initial step to enable file system monitoring capabilities.

```go
go get -u github.com/samber/ro/plugins/fsnotify
```

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

### Install ICS Plugin for ro

Source: https://ro.samber.dev/docs/plugins/ics

Installs the ICS sub-package for the ro library using go get. This is the first step to using ICS operators in your project.

```bash
go get -u github.com/samber/ro/plugins/ics
```

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

### Creating a Complete Observer in Go

Source: https://ro.samber.dev/docs/core/observer

Demonstrates how to create a fully functional Observer in Go using `ro.NewObserver`, providing callbacks for value reception, error handling, and completion. This is recommended for production code.

```Go
// Create a full Observer with all callbacks
observer := ro.NewObserver(
    func(value int) {
        fmt.Println("Received:", value)
    },
    func(err error) {
        fmt.Println("Error:", err)
    },
    func() {
        fmt.Println("Completed")
    },
)

// Use with an Observable
observable := ro.Just(1, 2, 3)
observable.Subscribe(observer)
// Output:
// Received: 1
// Received: 2
// Received: 3
// Completed
```

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

### Create Observable for User Input Prompt (Go)

Source: https://ro.samber.dev/docs/plugins/stdio

Creates an observable that prompts the user for input. It takes a prompt string and returns an observable that emits the user's input as a byte slice. The observable completes after the user provides input.

```go
import (
    "github.com/samber/ro"
    rostdio "github.com/samber/ro/plugins/stdio"
)

obs := rostdio.NewPrompt("Enter your name: ")

sub := obs.Subscribe(ro.NewObserver(
    func(input []byte) {
        println("You entered:", string(input))
    },
    func(err error) {
        println("Error:", err.Error())
    },
    func() {
        println("Input completed")
    },
))
def sub.Unsubscribe()

// (User sees prompt: "Enter your name: ")
// User types: "Alice"
// You entered: Alice
```

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

### Collect and Pipe1 Example in Go

Source: https://ro.samber.dev/docs/contributing

Demonstrates the usage of `ro.Collect` and `ro.Pipe1` for processing collections in Go. It shows how to combine operations like `Just` and `Flatten` and verifies the results and error handling using `is.Equal` and `is.NoError`.

```go
values, err := Collect(
    Pipe1(
        Just([]int{1, 2, 3}, []int{4, 5, 6}),
        Flatten[int](),
    ),
)
is.Equal([]int{1, 2, 3, 4, 5, 6}, values)
is.NoError(err)

```

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

### Synchronous Action with Start Observable

Source: https://ro.samber.dev/docs/operator/creation

Demonstrates creating an observable that executes a synchronous action for each subscriber using ro.Start. This is suitable for operations that need to be performed immediately upon subscription and should produce a single value.

```go
obs := ro.Start(func() int {
    // perform work such as HTTP request, database query...
    return 42
})

sub := obs.Subscribe(ro.PrintObserver[int]())
def sub.Unsubscribe()

// Next: 42
// Completed

```

```go
obs := ro.Start(func() string {
    time.Sleep(50 * time.Millisecond) // Simulate work
    return fmt.Sprintf("Result: %d", 21*2)
})

sub := obs.Subscribe(ro.PrintObserver[string]())
def sub.Unsubscribe()

// Next: "Result: 42" (after ~50ms)
// Completed

```

```go
obs := ro.Start(func() int {
    return rand.Intn(100) // Different random number each time
})

sub1 := obs.Subscribe(ro.PrintObserver[int]())
sub2 := obs.Subscribe(ro.PrintObserver[int]())

def sub1.Unsubscribe()
def sub2.Unsubscribe()

// Each subscriber gets a fresh execution
// sub1 might get: 73
// sub2 might get: 15

```

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

### IntervalWithInitial with Take Operator (Go)

Source: https://ro.samber.dev/docs/operator/creation

Demonstrates combining IntervalWithInitial with the Take operator to limit emissions after an initial delay. This example takes the first 5 values emitted by the observable, which then completes. The observable emits int64 values.

```Go
obs := ro.Pipe[int64, int64](
    ro.IntervalWithInitial(200*time.Millisecond, 100*time.Millisecond),
    ro.Take[int64](5),
)

sub := obs.Subscribe(ro.PrintObserver[int64]())
defer sub.Unsubscribe()

// Expected output:
// Next: 0 (after 200ms)
// Next: 1 (after 300ms)
// Next: 2 (after 400ms)
// Next: 3 (after 500ms)
// Next: 4 (after 600ms)
// Completed
```

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

### Install samber/oops Package in Go

Source: https://ro.samber.dev/docs/plugins/samber-oops

This snippet shows how to import the samber/oops sub-package into your Go project using the go get command. This is the first step to using the operators provided by the package.

```go
go get -u github.com/samber/ro/plugins/samber/oops
```

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

### Watch Disk Partition Statistics with ro/plugins/proc

Source: https://ro.samber.dev/docs/plugins/proc

Creates a watcher for disk partition statistics that updates at a specified interval. It subscribes to the watcher and prints partition statistics using ro.PrintObserver. Requires the 'github.com/shirou/gopsutil/v4/disk' package.

```go
import (
    "time"

    "github.com/samber/ro"
    roproc "github.com/samber/ro/plugins/proc"
    "github.com/shirou/gopsutil/v4/disk"
)

obs := roproc.NewDiskPartitionWatcher(5 * time.Second)

sub := obs.Subscribe(ro.PrintObserver[disk.PartitionStat]())
defer sub.Unsubscribe()

// Next: {Device: /dev/sda1, Mountpoint: /, Fstype: ext4, Opts: rw,relatime}
// Next: {Device: /dev/sda2, Mountpoint: /home, Fstype: ext4, Opts: rw,relatime}
// ...
```

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

### Transform Strings to Uppercase in Go

Source: https://ro.samber.dev/docs/getting-started

Transforms a stream of strings to their uppercase equivalents in Go using `ro.Map`. The `ro.Just` operator creates the initial stream of strings.

```go
obs := ro.Pipe1(
    ro.Just("apple", "banana", "cherry"),
    ro.Map(func(s string) string {
        return strings.ToUpper(s)
    }),
)
// Output: APPLE, BANANA, CHERRY
```

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

### BufferWithTime: Handle sparse emissions with time-based buffering (Go)

Source: https://ro.samber.dev/docs/operator/transformation

This example demonstrates BufferWithTime with sparse emissions. It shows that the buffering is strictly time-based, regardless of how many items are emitted within the time window. A delay is introduced to simulate sparse emissions.

```Go
package main

import (
	"fmt"
	"time"

	"github.com/reactivex/rxjs/go/rxjs"
)

func main() {
	obs := rxjs.Pipe[int64, []int64](
	 rxjs.Interval(100*time.Millisecond),
	 rxjs.Take[int64](10),
	 // Add delay to make emissions sparse  
	 rxjs.Map(func(i int64) int64 {
		time.Sleep(50 * time.Millisecond)
		return i
	}),
	 rxjs.BufferWithTime[int64](250*time.Millisecond),
	)

	sub := obs.Subscribe(rxjs.PrintObserver[[]int64]())
	time.Sleep(2000 * time.Millisecond)
	sub.Unsubscribe()

	// Buffers based on time, not item count  
	// Next: [0] (after 250ms)
	// Next: [1, 2] (after 500ms)
	// Next: [3, 4] (after 750ms)
	// Next: [5] (after 1000ms)
}

```

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

### Demonstrate Multiple Independent Subscriptions in Go

Source: https://ro.samber.dev/docs/getting-started

Shows how multiple subscriptions to the same observable in Go create independent streams, each restarting from the beginning. This highlights the 'cold' observable behavior of `ro.Just`.

```go
source := ro.Just(1, 2, 3)

// First subscription
sub1 := source.Subscribe(ro.OnNext(func(x int) {
    fmt.Println("Subscriber 1:", x)
}))

// Second subscription - restarts from beginning
sub2 := source.Subscribe(ro.OnNext(func(x int) {
    fmt.Println("Subscriber 2:", x)
}))

// Output:
// Subscriber 1: 1
// Subscriber 1: 2
// Subscriber 1: 3
// Subscriber 2: 1
// Subscriber 2: 2
// Subscriber 2: 3
```

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

### Watch CPU Information with ro/plugins/proc

Source: https://ro.samber.dev/docs/plugins/proc

Creates a watcher for CPU information that updates at a specified interval. It subscribes to the watcher and prints CPU statistics using ro.PrintObserver. Requires the 'github.com/shirou/gopsutil/v3/cpu' package.

```go
import (
    "time"

    "github.com/samber/ro"
    roproc "github.com/samber/ro/plugins/proc"
    "github.com/shirou/gopsutil/v3/cpu"
)

obs := roproc.NewCPUInfoWatcher(2 * time.Second)

sub := obs.Subscribe(ro.PrintObserver[cpu.InfoStat]())
defer sub.Unsubscribe()

// Next: {CPU: 0, VendorID: GenuineIntel, Family: 6, Model: 142 ...}
// Next: {CPU: 1, VendorID: GenuineIntel, Family: 6, Model: 142 ...}
// ...
```

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

### Transform and Filter Streams with Operators in Go

Source: https://ro.samber.dev/docs/getting-started

Demonstrates chaining operators in Go using `ro.Pipe2` to filter and transform an observable stream. It filters even numbers from a range and maps them to strings.

```go
// Chain operators with ro.Pipe
result := ro.Pipe2(
    ro.Range(0, 10),
    ro.Filter(func(x int64) bool {
        return x%2 == 0  // Keep only even numbers
    }),
    ro.Map(func(x int64) string {
        return fmt.Sprintf("even-%d", x)  // Transform to string
    }),
)

result.Subscribe(ro.OnNext(func(s string) {
    fmt.Println(s)
}))
// "even-0", "even-2", "even-4", "even-6", "even-8"
```

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

### Basic Observable Subscription and Unsubscription (Go)

Source: https://ro.samber.dev/docs/core/subscription

Demonstrates how to create an observable, subscribe to it, and then unsubscribe to cancel the execution. This is a fundamental pattern for managing reactive streams and preventing resource leaks.

```Go
// Create an Observable
observable := ro.Interval(1 * time.Second)

// Subscribe and get the subscription
subscription := observable.Subscribe(ro.OnNext(func(tick int64) {
    fmt.Println("Tick:", tick)
}))

// Cancel the subscription
subscription.Unsubscribe()

```

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

### Handle and Recover from Errors in Go Streams

Source: https://ro.samber.dev/docs/getting-started

Demonstrates error handling and recovery for observable streams in Go using `ro.MapErr` and `ro.Catch`. If an error occurs during mapping, `ro.Catch` provides a fallback observable.

```go
riskyStream := ro.Pipe2(
    ro.Just(1, 2, 3, 4, 5),
    ro.MapErr(func(x int) (int, error) {
        if x == 3 {
            return 0, fmt.Errorf("error at %d", x)
        }
        return x * 2, nil
    }),
    ro.Catch(func(err error) ro.Observable[int] {
        fmt.Println("Recovered from error:", err)
        return ro.Just(42)  // Fallback value
    }),
)

```

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

### Manual Backpressure with Go Channels (Go)

Source: https://ro.samber.dev/docs/core/backpressure

Illustrates manual backpressure management using Go channels with a fixed buffer size. It shows how to handle a full channel by either blocking, dropping values, or implementing other strategies.

```Go
// Traditional Go approach requires careful buffer management
ch := make(chan int, 100) // Fixed buffer size

go func() {
    for i := 0; i < 1000; i++ {
        select {
        case ch <- i:
            // OK - channel accepted value
        default:
            // Channel full - handle backpressure
            fmt.Println("Channel full, dropping value:", i)
            // Could block, drop, or use other strategy
        }
    }
    close(ch)
}()

for value := range ch {
    fmt.Println("Received:", value)
    time.Sleep(10 * time.Millisecond) // Slow processing
}

```

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

### Chain Context Transformations with ContextMap in Go

Source: https://ro.samber.dev/docs/operator/context

This example demonstrates chaining multiple `ro.ContextMap` operations to progressively add information to the context. It starts with `ro.ContextWithValue` and then uses `ro.ContextMap` to add a timestamp and a dynamically generated request ID.

```go
obs := ro.Pipe[string, string](
    ro.Just("data"),
    ro.ContextWithValue[string]("userID", 123),
    ro.ContextMap[string](func(ctx context.Context) context.Context {
        // Add timestamp to context
        return context.WithValue(ctx, "timestamp", time.Now())
    }),
    ro.ContextMap[string](func(ctx context.Context) context.Context {
        // Add request ID based on existing context
        userID := ctx.Value("userID")
        requestID := fmt.Sprintf("req-%v-%d", userID, time.Now().Unix())
        return context.WithValue(ctx, "requestID", requestID)
    }),
    ro.Map(func(s string) string {
        return fmt.Sprintf("Data: %s", s)
    }),
)

sub := obs.Subscribe(ro.NewObserver(
    func(value string) {
        userID := ctx.Value("userID")
        timestamp := ctx.Value("timestamp")
        requestID := ctx.Value("requestID")
        fmt.Printf("Next: %s (userID: %v, timestamp: %v, requestID: %v)\n",
            value, userID, timestamp, requestID)
    },
    func(err error) {
        fmt.Printf("Error: %v\n", err)
    },
    func() {
        fmt.Println("Completed")
    },
))
defer sub.Unsubscribe()

// Next: Data: data (userID: 123, timestamp: <current_time>, requestID: req-123-<timestamp>)
// Completed

```

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

### Recover Panics in ro Framework Operators

Source: https://ro.samber.dev/docs/getting-started

Demonstrates how the ro framework automatically catches panics within operators and converts them into errors. This ensures that unexpected panics do not crash the application but are handled as observable errors.

```go
package main

import (
	"fmt"
	"github.com/samber/ro"
)

func main() {
	// A stream with a potential panic in Map operator  
	stream := ro.Pipe1(
		ro.Just(1, 2, 3, 4, 5),
		ro.Map(func(x int) string {
			// This will panic when x == 3  
			if x == 3 {
				panic("something went wrong!")
			}
			return fmt.Sprintf("processed-%d", x)
		}),
	)

	// The framework automatically catches the panic and converts it to an error  
	ssubscription := stream.Subscribe(ro.NewObserver(
		func(value string) {
			fmt.Println("Received:", value)
		},
		func(err error) {
			fmt.Println("Error (automatically caught):", err)
		},
		func() {
			fmt.Println("Stream completed!")
		},
	))

	ssubscription.Wait()
}
```

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

### Combine Two Streams Sequentially in Go

Source: https://ro.samber.dev/docs/getting-started

Combines two observable streams sequentially in Go using `ro.Concat`. The output stream emits all values from the first stream followed by all values from the second stream.

```go
stream1 := ro.Just(1, 2, 3)
stream2 := ro.Just(4, 5, 6)

obs := ro.Concat(stream1, stream2)
// Output: 1, 2, 3, 4, 5, 6
```
Tokens:
Raw