rootel

package module
v0.0.0-...-5128dcf Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 5, 2026 License: Apache-2.0 Imports: 18 Imported by: 3

README

OpenTelemetry (Enterprise Edition)

Configuration

The plugin uses a CollectorConfig struct to configure observability behavior:

type CollectorConfig struct {
    // Enable or disable observability features
    EnableTracing bool
    EnableMetrics bool
    EnableLogging bool

    // Use custom providers or global defaults
    TracerProvider trace.TracerProvider
    MetricProvider metric.MeterProvider
    LoggerProvider log.LoggerProvider

    // Custom attributes for all telemetry data
    TraceAttributes   []attribute.KeyValue
    MetricAttributes  []attribute.KeyValue
    LoggingAttributes []attribute.KeyValue

    // Histogram bucket configuration for metrics
    MetricHistogramObjectivesSeconds []float64

    // Log level configuration
    LogLevelSubscription log.Severity
    LogLevelNext        log.Severity
    LogLevelError       log.Severity
}

Example

Basic Usage
import (
    "github.com/samber/ro"
    rootel "github.com/samber/ro/ee/plugins/otel"
    "go.opentelemetry.io/otel/attribute"
)

var obs, collector = rootel.Pipe3(
    rootel.CollectorConfig{
        EnableTracing: true,
        EnableMetrics: true,
        EnableLogging: true,
        TraceAttributes: []attribute.KeyValue{
            attribute.String("service", "my-service"),
            attribute.String("environment", "production"),
        },
    },
    ro.Just(1, 2, 3),
    ro.Map(func(v int64) int64 {
        return v * 2
    }),
    ro.Take(2),
)

func main() {
    // Subscribe to the observable
    subscription := obs.Subscribe(
        ...
    )
    defer subscription.Unsubscribe()
}
Tracing Example
package main

import (
    "context"
    "log"
    "os"
    "time"

    "github.com/samber/ro"
    rolicense "github.com/samber/ro/ee/pkg/license"
    rootel "github.com/samber/ro/ee/plugins/otel"
    rocsv "github.com/samber/ro/plugins/encoding/csv"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.24.0"
)

type User struct {
    ID   string
    Name string
}

func getUsers(index int64) ([]User, error) {
    // Simulate database query
    return []User{
        {ID: "1", Name: "Alice"},
        {ID: "2", Name: "Bob"},
    }, nil
}

var pipeline, collector = rootel.Pipe7(
    rootel.CollectorConfig{
        EnableTracing: true,
        EnableMetrics: false,
        EnableLogging: false,
        TraceAttributes: []attribute.KeyValue{
            attribute.String("test-mode", "tracing"),
        },
    },
    ro.Range(0, 100),
    ro.MapErr(getUsers),
    ro.RetryWithConfig[[]User](ro.RetryConfig{
        MaxRetries: 2,
        Delay:      5 * time.Second,
    }),
    ro.TakeWhile(func(users []User) bool {
        return len(users) > 0
    }),
    ro.Flatten[User](),
    ro.Map(func(user User) []string {
        return []string{user.ID, user.Name}
    }),
    ro.StartWith([]string{"ID", "Name"}),
    rocsv.NewCSVWriter(csv.NewWriter(os.Stdout)),
)

func main() {
    // Set license
    err := rolicense.SetLicense("your-license-key")
    if err != nil {
        log.Fatalf("Failed to set license: %v", err)
    }

    ctx := context.Background()

    // Initialize OpenTelemetry trace exporter
    exp, err := otlptracegrpc.New(ctx,
        otlptracegrpc.WithInsecure(),
        otlptracegrpc.WithEndpoint("localhost:4317"),
    )
    if err != nil {
        log.Fatalf("Failed to create trace exporter: %v", err)
    }

    // Create resource with service information
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceName("ee-otel-tracing"),
            semconv.ServiceVersion("1.0.0"),
        ),
    )
    if err != nil {
        log.Fatalf("Failed to create resource: %v", err)
    }

    // Create trace provider
    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exp),
        sdktrace.WithResource(res),
    )
    otel.SetTracerProvider(tp)
    defer tp.Shutdown(ctx)

    // Subscribe to the pipeline
    subscription := pipeline.Subscribe(
        ...
    )
    defer subscription.Unsubscribe()

    log.Println("Processing completed!")
}
Metrics Example
package main

import (
    "context"
    "log"
    "net/http"

    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promhttp"
    "github.com/samber/ro"
    rolicense "github.com/samber/ro/ee/pkg/license"
    rootel "github.com/samber/ro/ee/plugins/otel"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    otelprometheus "go.opentelemetry.io/otel/exporters/prometheus"
    sdkmetric "go.opentelemetry.io/otel/sdk/metric"
    "go.opentelemetry.io/otel/sdk/resource"
    semconv "go.opentelemetry.io/otel/semconv/v1.24.0"
)

var pipeline, collector = rootel.Pipe3(
    rootel.CollectorConfig{
        EnableLogging: false,
        EnableMetrics: true,
        EnableTracing: false,
        MetricAttributes: []attribute.KeyValue{
            attribute.String("service", "ee-otel-metrics"),
            attribute.String("environment", "demo"),
        },
    },
    ro.Just(1, 2, 3, 4, 5),
    ro.Map(func(v int64) int64 {
        return v * 2
    }),
    ro.Take(3),
)

func main() {
    // Set license
    err := rolicense.SetLicense("your-license-key")
    if err != nil {
        log.Fatalf("Failed to set license: %v", err)
    }

    ctx := context.Background()

    // Create resource with service information
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceName("ee-otel-metrics"),
            semconv.ServiceVersion("1.0.0"),
        ),
    )
    if err != nil {
        log.Fatalf("Failed to create resource: %v", err)
    }

    // Create a Prometheus registry
    reg := prometheus.NewRegistry()

    // Initialize Prometheus exporter for metrics
    exporter, err := otelprometheus.New(otelprometheus.WithRegisterer(reg))
    if err != nil {
        log.Fatalf("Failed to create Prometheus exporter: %v", err)
    }

    // Create meter provider
    mp := sdkmetric.NewMeterProvider(
        sdkmetric.WithReader(exporter),
        sdkmetric.WithResource(res),
    )
    otel.SetMeterProvider(mp)
    defer mp.Shutdown(ctx)

    // Start HTTP server for metrics endpoint
    go func() {
        handler := promhttp.HandlerFor(reg, promhttp.HandlerOpts{})
        http.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) {
            handler.ServeHTTP(w, r)
        })
        log.Println("Starting OpenTelemetry metrics server on :8080")
        if err := http.ListenAndServe(":8080", nil); err != nil {
            log.Fatalf("Failed to start metrics server: %v", err)
        }
    }()

    // Subscribe to the pipeline
    subscription := pipeline.Subscribe(
        ...
    )
    defer subscription.Unsubscribe()

    log.Println("Processing completed! Metrics available at http://localhost:8080/metrics")

    // Keep the main goroutine alive to serve metrics
    select {}
}

Documentation

Index

Constants

View Source 
const (
	LabelNamePipePosition = "pipe.position"
)

Variables

View Source 
var DefaultHistogramObjectivesSeconds = []float64{
	0.0010,
	0.0025,
	0.0050,
	0.0075,

	0.010,
	0.025,
	0.050,
	0.075,

	0.10,
	0.25,
	0.50,
	0.75,

	1,
	2.5,
	5,
	10,
	15,
	30,

	60,
	120,
	300,
	600,
	1800,
	3600,
	7200,
	10800,
	21600,
}

Functions

func IncCounterOnComplete 

func IncCounterOnComplete[T any](counter metric.Int64Counter, attributes []attribute.KeyValue) func(ro.Observable[T]) ro.Observable[T]

IncCounterOnComplete is a pipe operator that increments a counter when a new Complete() notification is sent to the destination observer.

func IncCounterOnError 

func IncCounterOnError[T any](counter metric.Int64Counter, attributes []attribute.KeyValue) func(ro.Observable[T]) ro.Observable[T]

IncCounterOnError is a pipe operator that increments a counter when a new Error() notification is sent to the destination observer.

func IncCounterOnNext 

func IncCounterOnNext[T any](counter metric.Int64Counter, attributes []attribute.KeyValue) func(ro.Observable[T]) ro.Observable[T]

IncCounterOnNext is a pipe operator that increments a counter when a new Next() notification is sent to the destination observer.

func IncCounterOnSubscription 

func IncCounterOnSubscription[T any](counter metric.Int64Counter, attributes []attribute.KeyValue) func(ro.Observable[T]) ro.Observable[T]

IncCounterOnSubscription is a pipe operator that increments a counter when a new subscription is created.

func LogOnComplete 

func LogOnComplete[T any](logger log.Logger, severity log.Severity, attributes []log.KeyValue) func(ro.Observable[T]) ro.Observable[T]

LogOnComplete is a pipe operator that logs a message when a new Complete() notification is sent to the destination observer.

func LogOnError 

func LogOnError[T any](logger log.Logger, severity log.Severity, attributes []log.KeyValue) func(ro.Observable[T]) ro.Observable[T]

LogOnError is a pipe operator that logs a message when a new Error() notification is sent to the destination observer.

func LogOnNext 

func LogOnNext[T any](logger log.Logger, severity log.Severity, attributes []log.KeyValue) func(ro.Observable[T]) ro.Observable[T]

LogOnNext is a pipe operator that logs a message when a new Next() notification is sent to the destination observer.

func LogOnSubscription 

func LogOnSubscription[T any](logger log.Logger, severity log.Severity, attributes []log.KeyValue) func(ro.Observable[T]) ro.Observable[T]

LogOnSubscription is a pipe operator that logs a message when a new subscription is created.

func ObserveNextLag 

func ObserveNextLag[T any](tracer trace.Tracer, operatorName string, histogram metric.Float64Histogram, attributes []attribute.KeyValue) func(ro.Observable[T]) ro.Observable[T]

ObserveNextLag is a pipe operator that tracks the time it takes for a notification to traverse from the source observable to the destination observer. It mesures the time the source pauses while waiting for the destination to process the notification.

func Pipe1 

func Pipe1[A any, B any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
) (ro.Observable[B], *otelCollector)

Pipe1 is a typesafe 🎉 implementation of Pipe, that takes a source and 1 operator.

func Pipe2 

func Pipe2[A any, B any, C any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
) (ro.Observable[C], *otelCollector)

Pipe2 is a typesafe 🎉 implementation of Pipe, that takes a source and 2 operators.

func Pipe3 

func Pipe3[A any, B any, C any, D any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
) (ro.Observable[D], *otelCollector)

Pipe3 is a typesafe 🎉 implementation of Pipe, that takes a source and 3 operators.

func Pipe4 

func Pipe4[A any, B any, C any, D any, E any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
) (ro.Observable[E], *otelCollector)

Pipe4 is a typesafe 🎉 implementation of Pipe, that takes a source and 4 operators.

func Pipe5 

func Pipe5[A any, B any, C any, D any, E any, F any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
) (ro.Observable[F], *otelCollector)

Pipe5 is a typesafe 🎉 implementation of Pipe, that takes a source and 5 operators.

func Pipe6 

func Pipe6[A any, B any, C any, D any, E any, F any, G any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
) (ro.Observable[G], *otelCollector)

Pipe6 is a typesafe 🎉 implementation of Pipe, that takes a source and 6 operators.

func Pipe7 

func Pipe7[A any, B any, C any, D any, E any, F any, G any, H any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
) (ro.Observable[H], *otelCollector)

Pipe7 is a typesafe 🎉 implementation of Pipe, that takes a source and 7 operators.

func Pipe8 

func Pipe8[A any, B any, C any, D any, E any, F any, G any, H any, I any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
) (ro.Observable[I], *otelCollector)

Pipe8 is a typesafe 🎉 implementation of Pipe, that takes a source and 8 operators.

func Pipe9 

func Pipe9[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
) (ro.Observable[J], *otelCollector)

func Pipe10 

func Pipe10[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
) (ro.Observable[K], *otelCollector)

func Pipe11 

func Pipe11[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
) (ro.Observable[L], *otelCollector)

Pipe11 is a typesafe 🎉 implementation of Pipe, that takes a source and 11 operators.

func Pipe12 

func Pipe12[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
) (ro.Observable[M], *otelCollector)

Pipe12 is a typesafe 🎉 implementation of Pipe, that takes a source and 12 operators.

func Pipe13 

func Pipe13[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
) (ro.Observable[N], *otelCollector)

Pipe13 is a typesafe 🎉 implementation of Pipe, that takes a source and 13 operators.

func Pipe14 

func Pipe14[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any, O any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
	operator14 func(ro.Observable[N]) ro.Observable[O],
) (ro.Observable[O], *otelCollector)

Pipe14 is a typesafe 🎉 implementation of Pipe, that takes a source and 14 operators.

func Pipe15 

func Pipe15[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any, O any, P any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
	operator14 func(ro.Observable[N]) ro.Observable[O],
	operator15 func(ro.Observable[O]) ro.Observable[P],
) (ro.Observable[P], *otelCollector)

Pipe15 is a typesafe 🎉 implementation of Pipe, that takes a source and 15 operators.

func Pipe16 

func Pipe16[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any, O any, P any, Q any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
	operator14 func(ro.Observable[N]) ro.Observable[O],
	operator15 func(ro.Observable[O]) ro.Observable[P],
	operator16 func(ro.Observable[P]) ro.Observable[Q],
) (ro.Observable[Q], *otelCollector)

Pipe16 is a typesafe 🎉 implementation of Pipe, that takes a source and 16 operators.

func Pipe17 

func Pipe17[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any, O any, P any, Q any, R any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
	operator14 func(ro.Observable[N]) ro.Observable[O],
	operator15 func(ro.Observable[O]) ro.Observable[P],
	operator16 func(ro.Observable[P]) ro.Observable[Q],
	operator17 func(ro.Observable[Q]) ro.Observable[R],
) (ro.Observable[R], *otelCollector)

Pipe17 is a typesafe 🎉 implementation of Pipe, that takes a source and 17 operators.

func Pipe18 

func Pipe18[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any, O any, P any, Q any, R any, S any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
	operator14 func(ro.Observable[N]) ro.Observable[O],
	operator15 func(ro.Observable[O]) ro.Observable[P],
	operator16 func(ro.Observable[P]) ro.Observable[Q],
	operator17 func(ro.Observable[Q]) ro.Observable[R],
	operator18 func(ro.Observable[R]) ro.Observable[S],
) (ro.Observable[S], *otelCollector)

Pipe18 is a typesafe 🎉 implementation of Pipe, that takes a source and 18 operators.

func Pipe19 

func Pipe19[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any, O any, P any, Q any, R any, S any, T any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
	operator14 func(ro.Observable[N]) ro.Observable[O],
	operator15 func(ro.Observable[O]) ro.Observable[P],
	operator16 func(ro.Observable[P]) ro.Observable[Q],
	operator17 func(ro.Observable[Q]) ro.Observable[R],
	operator18 func(ro.Observable[R]) ro.Observable[S],
	operator19 func(ro.Observable[S]) ro.Observable[T],
) (ro.Observable[T], *otelCollector)

Pipe19 is a typesafe 🎉 implementation of Pipe, that takes a source and 19 operators.

func Pipe20 

func Pipe20[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any, O any, P any, Q any, R any, S any, T any, U any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
	operator14 func(ro.Observable[N]) ro.Observable[O],
	operator15 func(ro.Observable[O]) ro.Observable[P],
	operator16 func(ro.Observable[P]) ro.Observable[Q],
	operator17 func(ro.Observable[Q]) ro.Observable[R],
	operator18 func(ro.Observable[R]) ro.Observable[S],
	operator19 func(ro.Observable[S]) ro.Observable[T],
	operator20 func(ro.Observable[T]) ro.Observable[U],
) (ro.Observable[U], *otelCollector)

Pipe20 is a typesafe 🎉 implementation of Pipe, that takes a source and 20 operators.

func Pipe21 

func Pipe21[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any, O any, P any, Q any, R any, S any, T any, U any, V any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
	operator14 func(ro.Observable[N]) ro.Observable[O],
	operator15 func(ro.Observable[O]) ro.Observable[P],
	operator16 func(ro.Observable[P]) ro.Observable[Q],
	operator17 func(ro.Observable[Q]) ro.Observable[R],
	operator18 func(ro.Observable[R]) ro.Observable[S],
	operator19 func(ro.Observable[S]) ro.Observable[T],
	operator20 func(ro.Observable[T]) ro.Observable[U],
	operator21 func(ro.Observable[U]) ro.Observable[V],
) (ro.Observable[V], *otelCollector)

Pipe21 is a typesafe 🎉 implementation of Pipe, that takes a source and 21 operators.

func Pipe22 

func Pipe22[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any, O any, P any, Q any, R any, S any, T any, U any, V any, W any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
	operator14 func(ro.Observable[N]) ro.Observable[O],
	operator15 func(ro.Observable[O]) ro.Observable[P],
	operator16 func(ro.Observable[P]) ro.Observable[Q],
	operator17 func(ro.Observable[Q]) ro.Observable[R],
	operator18 func(ro.Observable[R]) ro.Observable[S],
	operator19 func(ro.Observable[S]) ro.Observable[T],
	operator20 func(ro.Observable[T]) ro.Observable[U],
	operator21 func(ro.Observable[U]) ro.Observable[V],
	operator22 func(ro.Observable[V]) ro.Observable[W],
) (ro.Observable[W], *otelCollector)

Pipe22 is a typesafe 🎉 implementation of Pipe, that takes a source and 22 operators.

func Pipe23 

func Pipe23[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any, O any, P any, Q any, R any, S any, T any, U any, V any, W any, X any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
	operator14 func(ro.Observable[N]) ro.Observable[O],
	operator15 func(ro.Observable[O]) ro.Observable[P],
	operator16 func(ro.Observable[P]) ro.Observable[Q],
	operator17 func(ro.Observable[Q]) ro.Observable[R],
	operator18 func(ro.Observable[R]) ro.Observable[S],
	operator19 func(ro.Observable[S]) ro.Observable[T],
	operator20 func(ro.Observable[T]) ro.Observable[U],
	operator21 func(ro.Observable[U]) ro.Observable[V],
	operator22 func(ro.Observable[V]) ro.Observable[W],
	operator23 func(ro.Observable[W]) ro.Observable[X],
) (ro.Observable[X], *otelCollector)

Pipe23 is a typesafe 🎉 implementation of Pipe, that takes a source and 23 operators.

func Pipe24 

func Pipe24[A any, B any, C any, D any, E any, F any, G any, H any, I any, J any, K any, L any, M any, N any, O any, P any, Q any, R any, S any, T any, U any, V any, W any, X any, Y any](
	collectorConfig CollectorConfig,
	source ro.Observable[A],
	operator1 func(ro.Observable[A]) ro.Observable[B],
	operator2 func(ro.Observable[B]) ro.Observable[C],
	operator3 func(ro.Observable[C]) ro.Observable[D],
	operator4 func(ro.Observable[D]) ro.Observable[E],
	operator5 func(ro.Observable[E]) ro.Observable[F],
	operator6 func(ro.Observable[F]) ro.Observable[G],
	operator7 func(ro.Observable[G]) ro.Observable[H],
	operator8 func(ro.Observable[H]) ro.Observable[I],
	operator9 func(ro.Observable[I]) ro.Observable[J],
	operator10 func(ro.Observable[J]) ro.Observable[K],
	operator11 func(ro.Observable[K]) ro.Observable[L],
	operator12 func(ro.Observable[L]) ro.Observable[M],
	operator13 func(ro.Observable[M]) ro.Observable[N],
	operator14 func(ro.Observable[N]) ro.Observable[O],
	operator15 func(ro.Observable[O]) ro.Observable[P],
	operator16 func(ro.Observable[P]) ro.Observable[Q],
	operator17 func(ro.Observable[Q]) ro.Observable[R],
	operator18 func(ro.Observable[R]) ro.Observable[S],
	operator19 func(ro.Observable[S]) ro.Observable[T],
	operator20 func(ro.Observable[T]) ro.Observable[U],
	operator21 func(ro.Observable[U]) ro.Observable[V],
	operator22 func(ro.Observable[V]) ro.Observable[W],
	operator23 func(ro.Observable[W]) ro.Observable[X],
	operator24 func(ro.Observable[X]) ro.Observable[Y],
) (ro.Observable[Y], *otelCollector)

Pipe24 is a typesafe 🎉 implementation of Pipe, that takes a source and 24 operators.

func StartTraceOnSubscription 

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

StartTraceOnSubscription is a pipe operator that create a new OTEL trace when a new subscription is created.

func TraceOnError 

func TraceOnError[T any](collector *otelCollector) func(ro.Observable[T]) ro.Observable[T]

TraceOnError is a pipe operator that records an error in the current OTEL trace.

Types

type CollectorConfig 

type CollectorConfig struct {
	// Enable or disable the collection of traces, metrics and logs
	EnableTracing bool
	EnableMetrics bool
	EnableLogging bool

	// Use the global providers if nil
	TracerProvider trace.TracerProvider
	MetricProvider metric.MeterProvider
	LoggerProvider log.LoggerProvider

	// Attributes to add to all traces, metrics and logs
	TraceAttributes   []attribute.KeyValue
	MetricAttributes  []attribute.KeyValue
	LoggingAttributes []attribute.KeyValue

	// Histogram buckets in seconds
	MetricHistogramObjectivesSeconds []float64

	// On subscription, completion or cancellation
	LogLevelSubscription log.Severity
	// On next
	LogLevelNext log.Severity
	// On error
	LogLevelError log.Severity
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.