ro

https://pkg.go.dev/github.com/samber/ro
Admin
ro is a Go implementation of the ReactiveX specification for streams and reactive programming,...
Tokens:431,191
Snippets:3,189
Trust Score:10
Update:7 months ago
Show doc for...
Tokens:
Raw
### Install ro/plugins/regexp

Source: https://pkg.go.dev/github.com/samber/ro/plugins/regexp%40v0.0

Installs the regexp plugin for the ro package using go get.

```go
go get github.com/samber/ro/plugins/regexp

```

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

### Install ro stdio plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/stdio%40v0.0

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

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

```

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

### Install rostrconv String Conversion Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/strconv

Instructions to install the string conversion plugin for the rostrconv package using go get.

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

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

### Quick Start: Reactive WebSocket Client (Go)

Source: https://pkg.go.dev/github.com/samber/ro/plugins/websocket/client%40v0.0

A Go example demonstrating how to create a reactive WebSocket client using the Ro plugin. It shows how to establish a connection, subscribe to messages, handle errors, and send messages. This requires the Ro reactive programming library and gorilla/websocket.

```go
package main

import (
	"encoding/json"
	"fmt"
	
	"github.com/samber/ro"
	"github.com/samber/ro/plugins/websocket/client"
)

func main() {
	// Create a WebSocket subject
	ws := rowebsocket.NewWebsocketSubject(rowebsocket.WebsocketSubjectConfig[string, string]{
		URL: "ws://localhost:8080/ws",
		Serializer: func(msg string) ([]byte, error) {
			return json.Marshal(msg)
		},
		Deserializer: func(data []byte) (string, error) {
			var msg string
			err := json.Unmarshal(data, &msg)
			return msg, err
		},
	})
	
	// Subscribe to incoming messages
	subscription := ws.Subscribe(
		ro.NewObserver(
			func(message string) {
				fmt.Printf("Received: %s\n", message)
			},
			func(err error) {
				fmt.Printf("Error: %v\n", err)
			},
			func() {
				fmt.Println("WebSocket closed")
			},
		),
	)
	defer subscription.Unsubscribe()
	
	// Send a message
	ws.Next("Hello, WebSocket!")
	
	// Keep the application running
	select {}
}

```

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

### Install ro/plugins/io

Source: https://pkg.go.dev/github.com/samber/ro/plugins/io

Installs the IO plugin for the ro package.

```go
go get github.com/samber/ro/plugins/io

```

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

### Install rosort Plugin for Go

Source: https://pkg.go.dev/github.com/samber/ro/plugins/sort%40v0.0

Installs the rosort plugin for Go using the go get command. This is the first step to using the sorting operators.

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

```

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

### Install Zerolog Plugin for ro

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/zerolog%40v0.0

Installs the Zerolog plugin for ro using go get. This command fetches and installs the necessary package to enable Zerolog integration within your ro observables.

```go
go get github.com/samber/ro/plugins/observability/zerolog

```

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

### Quick Start: Ro WebSocket Client Example

Source: https://pkg.go.dev/github.com/samber/ro/plugins/websocket/client

Demonstrates a basic usage of the Ro WebSocket client. It shows how to create a WebSocket subject, subscribe to messages, send a message, and handle connection events. Requires Ro reactive programming library and gorilla/websocket.

```go
package main

import (
	"encoding/json"
	"fmt"
	
	"github.com/samber/ro"
	"github.com/samber/ro/plugins/websocket/client"
)

func main() {
	// Create a WebSocket subject
	ws := rowebsocket.NewWebsocketSubject(rowebsocket.WebsocketSubjectConfig[string, string]{
		URL: "ws://localhost:8080/ws",
		Serializer: func(msg string) ([]byte, error) {
			return json.Marshal(msg)
		},
		Deserializer: func(data []byte) (string, error) {
			var msg string
			err := json.Unmarshal(data, &msg)
			return msg, err
		},
	})
	
	// Subscribe to incoming messages
	subscription := ws.Subscribe(
		ro.NewObserver(
			func(message string) {
				fmt.Printf("Received: %s\n", message)
			},
			func(err error) {
				fmt.Printf("Error: %v\n", err)
			},
			func() {
				fmt.Println("WebSocket closed")
			},
		),
	)
	sdefer subscription.Unsubscribe()
	
	// Send a message
	ws.Next("Hello, WebSocket!")
	
	// Keep the application running
	select {}
}

```

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

### Install roproc Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/proc

Installs the roproc plugin for system resource and process monitoring. This command fetches the necessary package using the go get command.

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

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

### Install ro/plugins/io

Source: https://pkg.go.dev/github.com/samber/ro/plugins/io%40v0.0

Installs the IO plugin for the ro library. This is a prerequisite for using any of the IO operators.

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

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

### Run the Distributed Websocket Gateway (Go)

Source: https://pkg.go.dev/github.com/samber/ro/examples/distributed-websocket-gateway%40v0.0

This Go command starts the distributed websocket gateway. Ensure you have Go installed. The command executes the main package in the current directory.

```go
go run *.go
```

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

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

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/slog%40v0.0

Installs the slog plugin for the 'ro' library using 'go get'. This is the first step to integrating structured logging with Go's 'log/slog' package.

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

```

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

### Install rohttpclient HTTP Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/http/client

Installs the HTTP plugin for the ro package using go get. This is the first step to using HTTP request functionalities.

```go
go get github.com/samber/ro/plugins/http

```

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

### Install Ozzo Validation Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/ozzo/ozzo-validation%40v0.0

Installs the ozzo-validation plugin for reactive streams. This is the initial setup step before using any validation operators.

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

```

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

### Install rotime Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/time%40v0.0

This command installs the rotime plugin for use in your Go project. It utilizes the go get command to fetch the package from its repository.

```go
go get github.com/samber/ro/plugins/time

```

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

### Install rosignal Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/signal%40v0.0

Installs the rosignal plugin for handling operating system signals. This is the first step to using the signal catching functionalities.

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

```

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

### Install roproc Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/proc%40v0.0

Installs the roproc plugin for system resource and process monitoring. This is the initial step before utilizing any of the monitoring functionalities provided by the plugin.

```go
go get github.com/samber/ro/plugins/proc

```

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

### Install Ro WebSocket Client Plugin (Go)

Source: https://pkg.go.dev/github.com/samber/ro/plugins/websocket/client%40v0.0

This command installs the Ro WebSocket client plugin using Go modules. Ensure you have Go 1.18 or later installed. This plugin depends on the Ro reactive programming library and gorilla/websocket.

```go
go get github.com/samber/ro/plugins/websocket/client

```

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

### Install rostrings Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/strings%40v0.0

Installs the rostrings plugin for use in Go projects. This is the initial step before utilizing any string manipulation operators.

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

```

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

### Install Zap Plugin for ro

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/zap%40v0.0

Installs the Zap plugin for the ro library, enabling structured logging capabilities. This is a prerequisite for using the Zap operators.

```go
go get github.com/samber/ro/plugins/observability/zap

```

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

### Real-world Example: Logging API Requests

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/zerolog%40v0.0

A practical example demonstrating how to log API requests using the Zerolog plugin within a ro pipeline.

```APIDOC
## Real-world Example: Logging API Requests

### Description
This example demonstrates logging API requests in a production environment using Zerolog and the ro library.

### Method
`ro.Pipe2` with `rozerolog.LogWithNotification`

### Endpoint
N/A (Simulates API request processing)

### Parameters
None

### Request Example
```go
import (
    "context"
    "github.com/rs/zerolog"
    "github.com/rs/zerolog/log"
    "github.com/samber/ro"
    rozerolog "github.com/samber/ro/plugins/observability/zerolog"
)

// Create a logger for API requests
logger := log.With().
    Str("service", "api-gateway").
    Str("environment", "production").
    Logger()

// Process API requests with logging
pipeline := ro.Pipe2(
    // Source: API requests
    ro.Just("GET /users", "POST /users", "GET /users/123"),
    // Log each request
    rozerolog.LogWithNotification[string](&logger, zerolog.InfoLevel),
)

subscription := pipeline.Subscribe(
    ro.NewObserver(
        func(request string) {
            // Process the request
        },
        func(err error) {
            // Handle error
        },
        func() {
            // Handle completion
        },
    ),
)
deper subscription.Unsubscribe()
```

### Response
#### Success Response (200)
API requests are processed, and each request is logged with structured information.

#### Response Example
```json
{"level":"info","service":"api-gateway","environment":"production","message":"ro.Next: GET /users"}
{"level":"info","service":"api-gateway","environment":"production","message":"ro.Next: POST /users"}
{"level":"info","service":"api-gateway","environment":"production","message":"ro.Next: GET /users/123"}
{"level":"info","service":"api-gateway","environment":"production","message":"ro.Complete"}
```
```

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

### Install rocron Cron Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/cron%40v0.0

Installs the rocron cron plugin using go get. This is the first step to enable cron scheduling in your Go project.

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

```

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

### Install robase64 Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/encoding/base64%40v0.0

Installs the robase64 plugin for encoding and decoding data using Base64. This is the initial step before using the plugin's operators.

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

```

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

### Real-world Base64 Encoding Example

Source: https://pkg.go.dev/github.com/samber/ro/plugins/encoding/base64%40v0.0

A practical example demonstrating the use of Base64 encoding for processing binary data, such as generating Basic Authentication headers. It shows encoding credentials and then formatting them.

```go
import (
    "encoding/base64"
    "github.com/samber/ro"
    robase64 "github.com/samber/ro/plugins/encoding/base64"
)

// Process binary data with base64 encoding
pipeline := ro.Pipe3(
    // Simulate binary data
    ro.Just(
        []byte("user:password"),
        []byte("api:key123"),
        []byte("token:secret"),
    ),
    // Encode as base64
    robase64.Encode[[]byte](base64.StdEncoding),
    // Process encoded strings
    ro.Map(func(encoded string) string {
        return "Basic " + encoded
    }),
)

subscription := pipeline.Subscribe(ro.PrintObserver[string]())
def subscription.Unsubscribe()

// Output:
// Next: Basic dXNlcjpwYXNzd29yZA==
// Next: Basic YXBpOmtleTEyMw==
// Next: Basic dG9rZW46c2VjcmV0
// Completed

```

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

### Real-world Example: Log Monitoring

Source: https://pkg.go.dev/github.com/samber/ro/plugins/fsnotify%40v0.0

A practical example demonstrating how to monitor a log directory for new log files and process them.

```APIDOC
## POST /websites/pkg_go_dev_github_com_samber_ro/plugins/fsnotify/log_monitor

### Description
Monitors a specified directory for new log files (.log extension) and provides a pipeline to process these files.

### Method
POST

### Endpoint
/websites/pkg_go_dev_github_com_samber_ro/plugins/fsnotify/log_monitor

### Parameters
#### Query Parameters
- **logDirectory** (string) - Required - The directory to monitor for log files.

### Request Body
```json
{
  "logDirectory": "/var/log"
}
```

### Response
#### Success Response (200)
- **Observable[string]** (object) - An observable that emits the paths of newly created log files.

#### Response Example
```json
{
  "new_log_file": "/var/log/app.log"
}
```
```

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

### Install Testify Dependency for ro.go

Source: https://pkg.go.dev/github.com/samber/ro/plugins/testify%40v0.0

Provides the command to install the testify library, which is a required dependency for using the rotestify testing utilities.

```bash
go get github.com/stretchr/testify
```

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

### Install Native Rate Limiter Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/ratelimit/native%40v0.0

Installs the native rate limiter plugin for the ro package. This is the first step to using the rate limiting functionalities.

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

```

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

### Zerolog LogWithNotification Function Example

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/zerolog%40v0.0

This example demonstrates the `LogWithNotification` function from the `rozerolog` package. It shows how to log observable events with additional notification details at the debug level.

```go
import (
    "github.com/rs/zerolog"
    "github.com/samber/ro"
    rozerolog "github.com/samber/ro/plugins/observability/zerolog"
)

// Assume logger is already initialized
// var logger zerolog.Logger

// Example
observable := ro.Pipe1(
    ro.Just("hello", "world", "golang"),
    rozerolog.LogWithNotification[string](&logger, zerolog.DebugLevel),
)

// To run this example and see the output, you would typically subscribe to the observable:
// subscription := observable.Subscribe(ro.NewObserver(func(val string) { /* process val */ }))
// defer subscription.Unsubscribe()

```

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

### Install gopsutil Dependency

Source: https://pkg.go.dev/github.com/samber/ro/plugins/proc

This command installs the gopsutil library, which is a required dependency for the ro/plugins/proc package. gopsutil provides the underlying system information that the proc plugin wraps.

```bash
go get github.com/shirou/gopsutil/v4

```

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

### Duration Scheduling Examples with gocron (Go)

Source: https://pkg.go.dev/github.com/samber/ro/plugins/cron%40v0.0

Demonstrates how to use gocron.DurationJob for simple interval-based scheduling, showing examples for seconds, minutes, hours, and days.

```go
// Every 30 seconds
gocron.DurationJob(30 * time.Second)

// Every 5 minutes
gocron.DurationJob(5 * time.Minute)

// Every hour
gocron.DurationJob(1 * time.Hour)

// Every day
gocron.DurationJob(24 * time.Hour)

```

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

### Example Usage of roprometheus Collector

Source: https://pkg.go.dev/github.com/samber/ro/ee/plugins/prometheus

Demonstrates how to configure and use the roprometheus collector. This example shows setting up `CollectorConfig`, creating a stream with `ro.Just` and `ro.Map`, applying a `ro.Take` operator, and registering the collector with Prometheus.

```go
import (
    "github.com/samber/ro"
    roprometheus "github.com/samber/ro/ee/plugins/prometheus"
)

var obs, collector = roprometheus.Pipe3(
    roprometheus.CollectorConfig{
        Namespace: "example",
        Subsystem: "",
        ConstLabels: prometheus.Labels{
            "foo": "bar",
        },
    },
    ro.Just(1, 2, 3),
    ro.Map(func(v int64) int64 {
        return v*2
    }),
    ro.Take(2),
)

func main() {
    prometheus.MustRegister(collector)

    obs.Subscribe(
        ...
    )
}

```

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

### Real-world example: Email validation and extraction

Source: https://pkg.go.dev/github.com/samber/ro/plugins/regexp%40v0.0

A practical example demonstrating how to use the regexp plugin to filter and process valid email addresses from a stream of strings. It combines pattern compilation, filtering, and stream processing.

```go
import (
    "regexp"
    "github.com/samber/ro"
    roregexp "github.com/samber/ro/plugins/regexp"
)

// Email validation pattern
emailPattern := regexp.MustCompile(`^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$`)

// Process user data
pipeline := ro.Pipe2(
    // Source: User data
    ro.Just(
        "user1@example.com",
        "invalid-email",
        "user2@test.org",
        "not-an-email",
    ),
    // Filter valid emails
    roregexp.FilterMatchString[string](emailPattern),
)

subscription := pipeline.Subscribe(
    ro.NewObserver(
        func(email string) {
            // Process valid email
        },
        func(err error) {
            // Handle error
        },
        func() {
            // Handle completion
        },
    ),
)
def subscription.Unsubscribe()

```

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

### Install rohyperloglog Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/hyperloglog

Command to install the HyperLogLog plugin for the ro library using go get. This command fetches and installs the necessary package to integrate HyperLogLog functionality.

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

```

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

### Install rofsnotify Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/fsnotify

Installs the rofsnotify plugin for the ro package using go get. This command fetches and installs the necessary package to enable file system monitoring capabilities.

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

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

### Process File Line by Line with ro

Source: https://pkg.go.dev/github.com/samber/ro/plugins/io%40v0.0

This example demonstrates a practical use case of the ro library for processing a file line by line. It utilizes `NewIOReaderLine` to read lines and `ro.Map` to transform them, showing how to subscribe to the pipeline and handle results or errors.

```go
import (
    "bytes"
    "os"
    "strings"
    "github.com/samber/ro"
    roio "github.com/samber/ro/plugins/io"
)

// Process a file line by line
pipeline := ro.Pipe2(
    // Read lines from file
    roio.NewIOReaderLine(strings.NewReader(`Line 1: Hello
Line 2: World
Line 3: Test`)),
    // Transform lines
    ro.Map(func(line []byte) []byte {
        // Convert to uppercase
        return bytes.ToUpper(line)
    }),
)

subscription := pipeline.Subscribe(
    ro.NewObserver(
        func(line []byte) {
            // Process transformed line
        },
        func(err error) {
            // Handle error
        },
        func() {
            // Handle completion
        },
    ),
)
deferr subscription.Unsubscribe()

```

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

### Install Gob Encoding Plugin for ro

Source: https://pkg.go.dev/github.com/samber/ro/plugins/encoding/gob%40v0.0

Installs the gob encoding plugin for the ro library using `go get`. This command fetches and installs the necessary package to enable gob encoding and decoding functionalities.

```bash
go get github.com/samber/ro/plugins/encoding/gob
```

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

### Process File Line by Line with Mapping

Source: https://pkg.go.dev/github.com/samber/ro/plugins/io

This example demonstrates a real-world use case of processing a file line by line. It uses roio.NewIOReaderLine to read lines, ro.Map to transform them (e.g., to uppercase), and ro.NewObserver to handle the processed lines, errors, and completion.

```go
import (
    "bytes"
    "os"
    "strings"
    "github.com/samber/ro"
    roio "github.com/samber/ro/plugins/io"
)

// Process a file line by line
pipeline := ro.Pipe2(
    // Read lines from file
    roio.NewIOReaderLine(strings.NewReader(`Line 1: Hello
Line 2: World
Line 3: Test`)),
    // Transform lines
    ro.Map(func(line []byte) []byte {
        // Convert to uppercase
        return bytes.ToUpper(line)
    }),
)

subscription := pipeline.Subscribe(
    ro.NewObserver(
        func(line []byte) {
            // Process transformed line
        },
        func(err error) {
            // Handle error
        },
        func() {
            // Handle completion
        },
    ),
)
deffer subscription.Unsubscribe()

```

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

### Install rolog Observability Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/log

Installs the log observability plugin for the ro package using go get. This command fetches and installs the specified package, making its functionalities available for use in your Go project.

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

```

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

### Install Robytes Plugin for Go

Source: https://pkg.go.dev/github.com/samber/ro/plugins/bytes%40v0.0

Installs the Robytes plugin for Go using the go get command. This is the first step to using the byte slice and string manipulation operators.

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

```

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

### Create an Observer with Contextual Completion Handler (Go)

Source: https://pkg.go.dev/github.com/samber/ro/index_tab=versions

Creates an Observer that only defines a contextual completion handler. Other event handlers (next, error) will be no-ops.

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

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

### Install Logrus Plugin for ro

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/logrus

Installs the Logrus plugin for the 'ro' observability package using go get. This is the initial step to integrate Logrus logging with 'ro' observables.

```go
go get github.com/samber/ro/plugins/observability/logrus

```

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

### Real-world Example: Logging API Requests with ro and Zerolog

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/zerolog%40v0.0

A practical example showing how to log API requests using ro and Zerolog. It sets up a logger for an API gateway and processes requests through a pipeline that logs each request with structured fields.

```go
import (
    "context"
    "github.com/rs/zerolog"
    "github.com/rs/zerolog/log"
    "github.com/samber/ro"
    rozerolog "github.com/samber/ro/plugins/observability/zerolog"
)

// Create a logger for API requests
logger := log.With().
    Str("service", "api-gateway").
    Str("environment", "production").
    Logger()

// Process API requests with logging
pipeline := ro.Pipe2(
    // Source: API requests
    ro.Just("GET /users", "POST /users", "GET /users/123"),
    // Log each request
    rozerolog.LogWithNotification[string](&logger, zerolog.InfoLevel),
)

subscription := pipeline.Subscribe(
    ro.NewObserver(
        func(request string) {
            // Process the request
        },
        func(err error) {
            // Handle error
        },
        func() {
            // Handle completion
        },
    ),
)
defer subscription.Unsubscribe()

```

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

### Install ozzo-validation Dependency

Source: https://pkg.go.dev/github.com/samber/ro/plugins/ozzo/ozzo-validation%40v0.0

This snippet shows how to add the ozzo-validation library as a dependency for the samber/ro package using go get. This is a prerequisite for utilizing the validation features.

```go
go get github.com/go-ozzo/ozzo-validation/v4

```

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

### Zerolog Log Function Example (WithContext)

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/zerolog%40v0.0

This example showcases the `Log` function with context awareness. It demonstrates logging events that carry context information, illustrating how context can be integrated with Zerolog logging in ro pipelines.

```go
import (
    "context"
    "github.com/rs/zerolog"
    "github.com/samber/ro"
    rozerolog "github.com/samber/ro/plugins/observability/zerolog"
)

// Assume logger is already initialized
// var logger zerolog.Logger

// Example (WithContext)
ctx := context.WithValue(context.Background(), "value", "context")
ctx = context.WithValue(ctx, "aware", "logging")

observable := ro.Pipe1(
    ro.Just(1, 2, 3),
    ro.MapWithContext(func(ctx context.Context, i int) (int, error) {
        return i,
    }),
    rozerolog.LogWithContext[int](&logger, zerolog.InfoLevel),
)

// To run this example and see the output, you would typically subscribe to the observable:
// subscription := observable.Subscribe(ro.NewObserver(func(val int) { /* process val */ }))
// defer subscription.Unsubscribe()

```

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

### Filter ICS Events by Date in Go

Source: https://pkg.go.dev/github.com/samber/ro/plugins/ics%40v0.0

Shows how to filter iCalendar events based on their start date using the ro.Filter operator and the roics plugin. This example filters events occurring in 2024 or later.

```go
import (
    "time"
    "github.com/samber/ro"
    roics "github.com/samber/ro/plugins/ics"
    ics "github.com/arran4/golang-ical"
)

observable := ro.Pipe2(
    roics.NewICSFileReader("calendar.ics"),
    ro.Filter(func(event *ics.VEvent) bool {
        start := event.GetProperty(ics.ComponentPropertyDtStart)
        if start == nil {
            return false
        }
        
        eventTime, err := time.Parse("20060102T150405Z", start.Value)
        if err != nil {
            return false
        }
        
        // Filter events from 2024 onwards
        return eventTime.Year() >= 2024
    }),
)

subscription := observable.Subscribe(ro.PrintObserver[*ics.VEvent]())
def subscription.Unsubscribe()


```

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

### Write to File using ro.Pipe1 and roio.NewIOWriter

Source: https://pkg.go.dev/github.com/samber/ro/plugins/io

This snippet demonstrates how to write data to a file using the ro.Pipe1 function and the roio.NewIOWriter plugin. It creates a file named 'output.txt', writes 'Hello, World!' to it, and ensures the file is closed using defer.

```go
import "os"

file, err := os.Create("output.txt")
if err != nil {
    // Handle error
}
deffer file.Close()

data := ro.Just([]byte("Hello, World!"))
observable := ro.Pipe1(
    data,
    roio.NewIOWriter(file),
)

```

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

### Fetch JSON String Response with roHTTPRequestJSON

Source: https://pkg.go.dev/github.com/samber/ro/plugins/http/client

This Go example demonstrates how to fetch a JSON response expected to be a string using `rohttp.HTTPRequestJSON`. It makes an HTTP GET request and subscribes to an observable that automatically decodes the JSON response into a Go string. Ensure the `github.com/samber/ro` and `github.com/samber/ro/plugins/http` packages are imported.

```go
import (
    "net/http"
    "github.com/samber/ro"
    rohttp "github.com/samber/ro/plugins/http"
)

req, _ := http.NewRequest("GET", "https://api.example.com/message", nil)

observable := rohttp.HTTPRequestJSON[string](req, nil)

subscription := observable.Subscribe(ro.PrintObserver[string]())
def subscription.Unsubscribe()


```

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

### NewIOReader Example Output

Source: https://pkg.go.dev/github.com/samber/ro/plugins/io

Illustrates the expected output when using the NewIOReader function with a sample input. It shows the 'Next' event with the read bytes and the 'Completed' event.

```go
Output:

Next: [72 101 108 108 111 44 32 87 111 114 108 100 33 32 84 104 105 115 32 105 115 32 97 32 116 101 115 116 46]
Completed

```

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

### Create an Observer with Standard Completion Handler (Go)

Source: https://pkg.go.dev/github.com/samber/ro/index_tab=versions

Creates an Observer that only defines a standard completion handler. Other event handlers (next, error) will be no-ops.

```Go
func OnComplete[T any](onComplete func()) Observer[T]
```

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

### Use os.Open with NewIOReader in Go

Source: https://pkg.go.dev/github.com/samber/ro/plugins/stdio%40v0.0

Demonstrates using os.Open to open a file and then passing the file handle to NewIOReader to create an observable from file content. This allows processing file data as an input stream.

```go
import "os"

file, err := os.Open("data.txt")
if err != nil {
    // Handle error
}
def file.Close()

observable := rostdio.NewIOReader(file)


```

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

### Create Integer Range Observable with Rx Go

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

Range creates an Observable that emits a sequence of integers within a specified range [start:end). The `end` value is exclusive. It handles cases where start equals end or start is greater than end.

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

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

### Real-world Example: Logging API Requests with Zap and ro

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/zap%40v0.0

A practical example showing how to log API requests using Zap and the ro library. It sets up a logger, defines a pipeline to process requests, and logs each request using rozap.LogWithNotification.

```go
import (
    "context"
    "github.com/samber/ro"
    rozap "github.com/samber/ro/plugins/observability/zap"
    "go.uber.org/zap"
    "go.uber.org/zap/zapcore"
)

// Create a logger for API requests
logger, _ := zap.NewProduction()
defer logger.Sync()

// Process API requests with logging
pipeline := ro.Pipe2(
    // Source: API requests
    ro.Just("GET /users", "POST /users", "GET /users/123"),
    // Log each request
    rozap.LogWithNotification[string](logger, zapcore.InfoLevel),
)

subscription := pipeline.Subscribe(
    ro.NewObserver(
        func(request string) {
            // Process the request
        },
        func(err error) {
            // Handle error
        },
        func() {
            // Handle completion
        },
    ),
)
defer subscription.Unsubscribe()

```

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

### Start Trace on Subscription (Go)

Source: https://pkg.go.dev/github.com/samber/ro/ee/plugins/otel%40v0.0

A function that returns a pipe operator to start tracing when a subscription is created. It requires an otelCollector.

```go
func StartTraceOnSubscription[T any](collector *otelCollector) func(ro.Observable[T]) ro.Observable[T]
```

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

### Install roics Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/ics%40v0.0

This command installs the roics plugin for the ro library. It is a prerequisite for using the ICS file and URL reading functionalities.

```go
go get github.com/samber/ro/plugins/ics

```

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

### Real-world Example: Logging API Requests

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/zap

Demonstrates how to use the Rozap plugin to log incoming API requests within a pipeline.

```APIDOC
## Real-world Example: Logging API Requests

### Description

This example shows how to log API requests as they are processed through an observable pipeline using the `rozap.LogWithNotification` function.

### Code Example

```go
import (
    "context"
    "github.com/samber/ro"
    rozap "github.com/samber/ro/plugins/observability/zap"
    "go.uber.org/zap"
    "go.uber.org/zap/zapcore"
)

// Create a logger for API requests
logger, _ := zap.NewProduction()
defer logger.Sync()

// Process API requests with logging
pipeline := ro.Pipe2(
    // Source: API requests
    ro.Just("GET /users", "POST /users", "GET /users/123"),
    // Log each request
    rozap.LogWithNotification[string](logger, zapcore.InfoLevel),
)

subscription := pipeline.Subscribe(
    ro.NewObserver(
        func(request string) {
            // Process the request
        },
        func(err error) {
            // Handle error
        },
        func() {
            // Handle completion
        },
    ),
)
defer subscription.Unsubscribe()
```

### Explanation

1. A Zap logger is initialized.
2. An observable pipeline is created using `ro.Pipe2`.
3. `ro.Just` provides a stream of sample API requests.
4. `rozap.LogWithNotification` is used to log each request at the `InfoLevel`.
5. An observer is subscribed to the pipeline to handle the processed requests, errors, and completion signals.
```

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

### Real-world Example: Logging API Requests with ro and slog

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/slog%40v0.0

Demonstrates a practical application of the ro library with slog for logging API requests. It sets up a slog logger and uses `LogWithNotification` to log each request in the observable stream.

```go
import (
    "context"
    "log/slog"
    "os"
    "github.com/samber/ro"
    roslog "github.com/samber/ro/plugins/observability/slog"
)

// Create a logger for API requests
handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
    Level: slog.LevelInfo,
})
logger := slog.New(handler)

// Process API requests with logging
pipeline := ro.Pipe2(
    // Source: API requests
    ro.Just("GET /users", "POST /users", "GET /users/123"),
    // Log each request
    roslog.LogWithNotification[string](*logger, slog.LevelInfo),
)

subscription := pipeline.Subscribe(
    ro.NewObserver(
        func(request string) {
            // Process the request
        },
        func(err error) {
            // Handle error
        },
        func() {
            // Handle completion
        },
    ),
)
def subscription.Unsubscribe()

```

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

### Install fsnotify Plugin for ro

Source: https://pkg.go.dev/github.com/samber/ro/plugins/fsnotify%40v0.0

Installs the fsnotify plugin for the ro library, which is necessary for using file system monitoring functionalities.

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

```

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

### Zerolog Log Function Example (InPipeline)

Source: https://pkg.go.dev/github.com/samber/ro/plugins/observability/zerolog%40v0.0

This example demonstrates the `Log` function from the `rozerolog` package used within an ro pipeline. It shows how to log observable events at the info level, processing only even numbers.

```go
import (
    "github.com/rs/zerolog"
    "github.com/samber/ro"
    rozerolog "github.com/samber/ro/plugins/observability/zerolog"
)

// Assume logger is already initialized
// var logger zerolog.Logger

// Example (InPipeline)
observable := ro.Pipe1(
    ro.Just(1, 2, 3, 4, 5),
    ro.Filter(func(i int) bool { return i%2 == 0 }),
    rozerolog.Log[int](&logger, zerolog.InfoLevel),
)

// To run this example and see the output, you would typically subscribe to the observable:
// subscription := observable.Subscribe(ro.NewObserver(func(val int) { /* process val */ }))
// defer subscription.Unsubscribe()

```

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

### Configure Rate Limiter Stores (Go)

Source: https://pkg.go.dev/github.com/samber/ro/plugins/ratelimit/ulule%40v0.0

Illustrates how to configure different storage backends for the rate limiter. It shows examples for an in-memory store and a distributed Redis store.

```go
// Memory store (in-process)
store := memory.NewStore()

// Redis store (distributed)
redisClient := redis.NewClient(&redis.Options{
    Addr: "localhost:6379",
})
store := redis.NewStoreWithOptions(redisClient, limiter.StoreOptions{
    Prefix: "rate_limiter",
})

```

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

### Install Gob Encoding Plugin

Source: https://pkg.go.dev/github.com/samber/ro/plugins/encoding/gob

Installs the gob encoding plugin for the rogob package. This is the first step to using gob encoding and decoding functionalities.

```go
go get github.com/samber/ro/plugins/encoding/gob

```

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

### Create an Observer with Contextual Next Handler (Go)

Source: https://pkg.go.dev/github.com/samber/ro/index_tab=versions

Creates an Observer that only defines a contextual next value handler. Other event handlers (error, complete) will be no-ops.

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

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

### Install ro Package in Go

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

Provides the command to install the 'ro' package, a Go implementation of the ReactiveX spec, using the Go module system.

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

```
Tokens:
Raw