Versions in this module Expand all Collapse all v0 v0.4.0 Jul 4, 2026 v0.3.0 Mar 2, 2026 Changes in this version + const ReplaySubjectUnlimitedBufferSize + const UnicastSubjectUnlimitedBufferSize + var ErrBufferWithCountWrongSize = errors.New("ro.BufferWithCount: size must be greater than 0") + var ErrBufferWithTimeOrCountWrongDuration = errors.New("ro.BufferWithTimeOrCount: duration must be greater than 0") + var ErrBufferWithTimeOrCountWrongSize = errors.New("ro.BufferWithTimeOrCount: size must be greater than 0") + var ErrBufferWithTimeWrongDuration = errors.New("ro.BufferWithTime: duration must be greater than 0") + var ErrClampLowerLessThanUpper = errors.New("ro.Clamp: lower must be less than or equal to upper") + var ErrConnectableObservableMissingConnectorFactory = errors.New("ro.ConnectableObservable: missing connector factory") + var ErrDetachOnWrongMode = errors.New("ro.detachOn: unexpected detach mode") + var ErrElementAtNotFound = errors.New("ro.ElementAt: nth element not found") + var ErrElementAtOrDefaultWrongNth = errors.New("ro.ElementAtOrDefault: nth must be greater or equal to 0") + var ErrElementAtWrongNth = errors.New("ro.ElementAt: nth must be greater or equal to 0") + var ErrFirstEmpty = errors.New("ro.First: empty") + var ErrHeadEmpty = errors.New("ro.First: empty") + var ErrLastEmpty = errors.New("ro.Last: empty") + var ErrObserveOnWrongBufferSize = errors.New("ro.ObserveOn: buffer size must be greater than 0") + var ErrPoolWrongSize = errors.New("ro.Pool: size must be greater than 0") + var ErrRangeWithStepAndIntervalWrongStep = errors.New("ro.RangeWithStepAndInterval: step must be greater than 0") + var ErrRangeWithStepWrongStep = errors.New("ro.RangeWithStep: step must be greater than 0") + var ErrRepeatWithIntervalWrongCount = errors.New("ro.RepeatWithInterval: count must be greater or equal to 0") + var ErrRepeatWithWrongCount = errors.New("ro.RepeatWith: count must be greater or equal to 0") + var ErrRepeatWrongCount = errors.New("ro.Repeat: count must be greater or equal to 0") + var ErrSkipLastWrongCount = errors.New("ro.SkipLast: count must be greater than 0") + var ErrSkipWrongCount = errors.New("ro.Skip: count must be greater or equal to 0") + var ErrSubscribeOnWrongBufferSize = errors.New("ro.SubscribeOn: buffer size must be greater than 0") + var ErrTailEmpty = errors.New("ro.Last: empty") + var ErrTakeLastWrongCount = errors.New("ro.TakeLast: count must be greater than 0") + var ErrTakeWrongCount = errors.New("ro.Take: count must be greater or equal to 0") + var ErrToChannelWrongSize = errors.New("ro.ErrToChannelWrongSize: size must be greater or equal to 0") + var ErrUnicastSubjectConcurrent = errors.New("ro.UnicastSubject: a single subscriber accepted") + var OnDroppedNotification = IgnoreOnDroppedNotification + var OnUnhandledError = IgnoreOnUnhandledError + func Abs() func(Observable[float64]) Observable[float64] + func AllIWithContext[T any](predicate func(ctx context.Context, item T, index int64) bool) func(Observable[T]) Observable[bool] + func AllI[T any](predicate func(item T, index int64) bool) func(Observable[T]) Observable[bool] + func AllWithContext[T any](predicate func(ctx context.Context, item T) bool) func(Observable[T]) Observable[bool] + func All[T any](predicate func(T) bool) func(Observable[T]) Observable[bool] + func Average[T constraints.Numeric]() func(Observable[T]) Observable[float64] + func BufferWhen[T, B any](boundary Observable[B]) func(Observable[T]) Observable[[]T] + func BufferWithCount[T any](size int) func(Observable[T]) Observable[[]T] + func BufferWithTimeOrCount[T any](size int, duration time.Duration) func(Observable[T]) Observable[[]T] + func BufferWithTime[T any](duration time.Duration) func(Observable[T]) Observable[[]T] + func Cast[T, U any]() func(Observable[T]) Observable[U] + func Catch[T any](finally func(err error) Observable[T]) func(Observable[T]) Observable[T] + func Ceil() func(Observable[float64]) Observable[float64] + func CeilWithPrecision(places int) func(Observable[float64]) Observable[float64] + func Clamp[T constraints.Numeric](lower, upper T) func(Observable[T]) Observable[T] + func CollectWithContext[T any](ctx context.Context, obs Observable[T]) ([]T, context.Context, error) + func Collect[T any](obs Observable[T]) ([]T, error) + func CombineLatestAllAny() func(Observable[Observable[any]]) Observable[[]any] + func CombineLatestAll[T any]() func(Observable[Observable[T]]) Observable[[]T] + func CombineLatestWith1[A, B any](obsB Observable[B]) func(Observable[A]) Observable[lo.Tuple2[A, B]] + func CombineLatestWith2[A, B, C any](obsB Observable[B], obsC Observable[C]) func(Observable[A]) Observable[lo.Tuple3[A, B, C]] + func CombineLatestWith3[A, B, C, D any](obsB Observable[B], obsC Observable[C], obsD Observable[D]) func(Observable[A]) Observable[lo.Tuple4[A, B, C, D]] + func CombineLatestWith4[A, B, C, D, E any](obsB Observable[B], obsC Observable[C], obsD Observable[D], obsE Observable[E]) func(Observable[A]) Observable[lo.Tuple5[A, B, C, D, E]] + func CombineLatestWith[A, B any](obsB Observable[B]) func(Observable[A]) Observable[lo.Tuple2[A, B]] + func ConcatAll[T any]() func(Observable[Observable[T]]) Observable[T] + func ConcatWith[T any](obs ...Observable[T]) func(Observable[T]) Observable[T] + func ContainsIWithContext[T any](predicate func(ctx context.Context, item T, index int64) bool) func(Observable[T]) Observable[bool] + func ContainsI[T any](predicate func(item T, index int64) bool) func(Observable[T]) Observable[bool] + func ContainsWithContext[T any](predicate func(ctx context.Context, item T) bool) func(Observable[T]) Observable[bool] + func Contains[T any](predicate func(item T) bool) func(Observable[T]) Observable[bool] + func ContextMapI[T any](project func(ctx context.Context, index int64) context.Context) func(Observable[T]) Observable[T] + func ContextMap[T any](project func(ctx context.Context) context.Context) func(Observable[T]) Observable[T] + func ContextReset[T any](newCtx context.Context) func(Observable[T]) Observable[T] + func ContextWithDeadline[T any](deadline time.Time) func(Observable[T]) Observable[T] + func ContextWithTimeout[T any](timeout time.Duration) func(Observable[T]) Observable[T] + func ContextWithValue[T any](k, v any) func(Observable[T]) Observable[T] + func Count[T any]() func(Observable[T]) Observable[int64] + func DefaultIfEmptyWithContext[T any](defaultCtx context.Context, defaultValue T) func(Observable[T]) Observable[T] + func DefaultIfEmpty[T any](defaultValue T) func(Observable[T]) Observable[T] + func DefaultOnDroppedNotification(ctx context.Context, notification fmt.Stringer) + func DefaultOnUnhandledError(ctx context.Context, err error) + func DelayEach[T any](duration time.Duration) func(Observable[T]) Observable[T] + func Delay[T any](duration time.Duration) func(Observable[T]) Observable[T] + func Dematerialize[T any]() func(Observable[Notification[T]]) Observable[T] + func DistinctByWithContext[T any, K comparable](keySelector func(ctx context.Context, item T) (context.Context, K)) func(Observable[T]) Observable[T] + func DistinctBy[T any, K comparable](keySelector func(item T) K) func(Observable[T]) Observable[T] + func Distinct[T comparable]() func(Observable[T]) Observable[T] + func DoOnCompleteWithContext[T any](onComplete func(ctx context.Context)) func(Observable[T]) Observable[T] + func DoOnComplete[T any](onComplete func()) func(Observable[T]) Observable[T] + func DoOnErrorWithContext[T any](onError func(ctx context.Context, err error)) func(Observable[T]) Observable[T] + func DoOnError[T any](onError func(err error)) func(Observable[T]) Observable[T] + func DoOnFinalize[T any](onFinalize func()) func(Observable[T]) Observable[T] + func DoOnNextWithContext[T any](onNext func(ctx context.Context, value T)) func(Observable[T]) Observable[T] + func DoOnNext[T any](onNext func(value T)) func(Observable[T]) Observable[T] + func DoOnSubscribeWithContext[T any](onSubscribe func(ctx context.Context)) func(Observable[T]) Observable[T] + func DoOnSubscribe[T any](onSubscribe func()) func(Observable[T]) Observable[T] + func DoWhileIWithContext[T any](condition func(ctx context.Context, index int64) (context.Context, bool)) func(Observable[T]) Observable[T] + func DoWhileI[T any](condition func(index int64) bool) func(Observable[T]) Observable[T] + func DoWhileWithContext[T any](condition func(ctx context.Context) (context.Context, bool)) func(Observable[T]) Observable[T] + func DoWhile[T any](condition func() bool) func(Observable[T]) Observable[T] + func DoWithContext[T any](onNext func(ctx context.Context, value T), ...) func(Observable[T]) Observable[T] + func Do[T any](onNext func(value T), onError func(err error), onComplete func()) func(Observable[T]) Observable[T] + func ElementAtOrDefault[T any](nth int64, fallback T) func(Observable[T]) Observable[T] + func ElementAt[T any](nth int) func(Observable[T]) Observable[T] + func EndWith[T any](suffixes ...T) func(Observable[T]) Observable[T] + func FilterIWithContext[T any](...) func(Observable[T]) Observable[T] + func FilterI[T any](predicate func(item T, index int64) bool) func(Observable[T]) Observable[T] + func FilterWithContext[T any](predicate func(ctx context.Context, item T) (context.Context, bool)) func(Observable[T]) Observable[T] + func Filter[T any](predicate func(item T) bool) func(Observable[T]) Observable[T] + func FindIWithContext[T any](predicate func(ctx context.Context, item T, index int64) bool) func(Observable[T]) Observable[T] + func FindI[T any](predicate func(item T, index int64) bool) func(Observable[T]) Observable[T] + func FindWithContext[T any](predicate func(ctx context.Context, item T) bool) func(Observable[T]) Observable[T] + func Find[T any](predicate func(item T) bool) func(Observable[T]) Observable[T] + func FirstIWithContext[T any](...) func(Observable[T]) Observable[T] + func FirstI[T any](predicate func(item T, index int64) bool) func(Observable[T]) Observable[T] + func FirstWithContext[T any](predicate func(ctx context.Context, item T) (context.Context, bool)) func(Observable[T]) Observable[T] + func First[T any](predicate func(item T) bool) func(Observable[T]) Observable[T] + func FlatMapIWithContext[T, R any](project func(ctx context.Context, item T, index int64) Observable[R]) func(Observable[T]) Observable[R] + func FlatMapI[T, R any](project func(item T, index int64) Observable[R]) func(Observable[T]) Observable[R] + func FlatMapWithContext[T, R any](project func(ctx context.Context, item T) Observable[R]) func(Observable[T]) Observable[R] + func FlatMap[T, R any](project func(item T) Observable[R]) func(Observable[T]) Observable[R] + func Flatten[T any]() func(Observable[[]T]) Observable[T] + func Floor() func(Observable[float64]) Observable[float64] + func FloorWithPrecision(places int) func(Observable[float64]) Observable[float64] + func GroupByIWithContext[T any, K comparable](iteratee func(ctx context.Context, item T, index int64) (context.Context, K)) func(Observable[T]) Observable[Observable[T]] + func GroupByI[T any, K comparable](iteratee func(item T, index int64) K) func(Observable[T]) Observable[Observable[T]] + func GroupByWithContext[T any, K comparable](iteratee func(ctx context.Context, item T) (context.Context, K)) func(Observable[T]) Observable[Observable[T]] + func GroupBy[T any, K comparable](iteratee func(item T) K) func(Observable[T]) Observable[Observable[T]] + func Head[T any]() func(Observable[T]) Observable[T] + func IgnoreElements[T any]() func(Observable[T]) Observable[T] + func IgnoreOnDroppedNotification(ctx context.Context, notification fmt.Stringer) + func IgnoreOnUnhandledError(ctx context.Context, err error) + func Iif[T any](predicate func() bool, source1, source2 Observable[T]) func() Observable[T] + func LastIWithContext[T any](...) func(Observable[T]) Observable[T] + func LastI[T any](predicate func(item T, index int64) bool) func(Observable[T]) Observable[T] + func LastWithContext[T any](predicate func(ctx context.Context, item T) (context.Context, bool)) func(Observable[T]) Observable[T] + func Last[T any](predicate func(item T) bool) func(Observable[T]) Observable[T] + func MapErrIWithContext[T, R any](...) func(Observable[T]) Observable[R] + func MapErrI[T, R any](project func(item T, index int64) (R, error)) func(Observable[T]) Observable[R] + func MapErrWithContext[T, R any](project func(ctx context.Context, item T) (R, context.Context, error)) func(Observable[T]) Observable[R] + func MapErr[T, R any](project func(item T) (R, error)) func(Observable[T]) Observable[R] + func MapIWithContext[T, R any](project func(ctx context.Context, item T, index int64) (context.Context, R)) func(Observable[T]) Observable[R] + func MapI[T, R any](project func(item T, index int64) R) func(Observable[T]) Observable[R] + func MapTo[T, R any](output R) func(Observable[T]) Observable[R] + func MapWithContext[T, R any](project func(ctx context.Context, item T) (context.Context, R)) func(Observable[T]) Observable[R] + func Map[T, R any](project func(item T) R) func(Observable[T]) Observable[R] + func Materialize[T any]() func(Observable[T]) Observable[Notification[T]] + func Max[T constraints.Numeric]() func(Observable[T]) Observable[T] + func MergeAll[T any]() func(Observable[Observable[T]]) Observable[T] + func MergeMapIWithContext[T, R any](...) func(Observable[T]) Observable[R] + func MergeMapI[T, R any](projection func(item T, index int64) Observable[R]) func(Observable[T]) Observable[R] + func MergeMapWithContext[T, R any](projection func(ctx context.Context, item T) Observable[R]) func(Observable[T]) Observable[R] + func MergeMap[T, R any](projection func(item T) Observable[R]) func(Observable[T]) Observable[R] + func MergeWith1[T any](obsB Observable[T]) func(Observable[T]) Observable[T] + func MergeWith2[T any](obsB, obsC Observable[T]) func(Observable[T]) Observable[T] + func MergeWith3[T any](obsB, obsC, obsD Observable[T]) func(Observable[T]) Observable[T] + func MergeWith4[T any](obsB, obsC, obsD, obsE Observable[T]) func(Observable[T]) Observable[T] + func MergeWith5[T any](obsB, obsC, obsD, obsE, obsF Observable[T]) func(Observable[T]) Observable[T] + func MergeWith[T any](observables ...Observable[T]) func(Observable[T]) Observable[T] + func Min[T constraints.Numeric]() func(Observable[T]) Observable[T] + func NewScheduler() + func ObserveOn[T any](bufferSize int) func(Observable[T]) Observable[T] + func OnErrorResumeNextWith[T any](finally ...Observable[T]) func(Observable[T]) Observable[T] + func OnErrorReturn[T any](finally T) func(Observable[T]) Observable[T] + func Pairwise[T any]() func(Observable[T]) Observable[[]T] + func PipeOp10[A, B, C, D, E, F, G, H, I, J, K any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[K] + func PipeOp11[A, B, C, D, E, F, G, H, I, J, K, L any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[L] + func PipeOp12[A, B, C, D, E, F, G, H, I, J, K, L, M any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[M] + func PipeOp13[A, B, C, D, E, F, G, H, I, J, K, L, M, N any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[N] + func PipeOp14[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[O] + func PipeOp15[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[P] + func PipeOp16[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[Q] + func PipeOp17[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[R] + func PipeOp18[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[S] + func PipeOp19[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[T] + func PipeOp1[A, B any](operator1 func(Observable[A]) Observable[B]) func(Observable[A]) Observable[B] + func PipeOp20[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[U] + func PipeOp21[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[V] + func PipeOp22[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[W] + func PipeOp23[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W, X any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[X] + func PipeOp24[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W, X, Y any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[Y] + func PipeOp25[...](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[Z] + func PipeOp2[A, B, C any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[C] + func PipeOp3[A, B, C, D any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[D] + func PipeOp4[A, B, C, D, E any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[E] + func PipeOp5[A, B, C, D, E, F any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[F] + func PipeOp6[A, B, C, D, E, F, G any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[G] + func PipeOp7[A, B, C, D, E, F, G, H any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[H] + func PipeOp8[A, B, C, D, E, F, G, H, I any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[I] + func PipeOp9[A, B, C, D, E, F, G, H, I, J any](operator1 func(Observable[A]) Observable[B], ...) func(Observable[A]) Observable[J] + func PipeOp[First, Last any](operators ...any) func(Observable[First]) Observable[Last] + func RaceWith[T any](sources ...Observable[T]) func(Observable[T]) Observable[T] + func ReduceIWithContext[T, R any](...) func(Observable[T]) Observable[R] + func ReduceI[T, R any](accumulator func(agg R, item T, index int64) R, seed R) func(Observable[T]) Observable[R] + func ReduceWithContext[T, R any](accumulator func(ctx context.Context, agg R, item T) (context.Context, R), ...) func(Observable[T]) Observable[R] + func Reduce[T, R any](accumulator func(agg R, item T) R, seed R) func(Observable[T]) Observable[R] + func RepeatWith[T any](count int64) func(Observable[T]) Observable[T] + func RetryWithConfig[T any](opts RetryConfig) func(Observable[T]) Observable[T] + func Retry[T any]() func(Observable[T]) Observable[T] + func Round() func(Observable[float64]) Observable[float64] + func SampleTime[T any](interval time.Duration) func(Observable[T]) Observable[T] + func SampleWhen[T, t any](tick Observable[t]) func(Observable[T]) Observable[T] + func ScanIWithContext[T, R any](...) func(Observable[T]) Observable[R] + func ScanI[T, R any](reduce func(accumulator R, item T, index int64) R, seed R) func(Observable[T]) Observable[R] + func ScanWithContext[T, R any](reduce func(ctx context.Context, accumulator R, item T) (context.Context, R), ...) func(Observable[T]) Observable[R] + func Scan[T, R any](reduce func(accumulator R, item T) R, seed R) func(Observable[T]) Observable[R] + func SequenceEqual[T comparable](obsB Observable[T]) func(Observable[T]) Observable[bool] + func Serialize[T any]() func(Observable[T]) Observable[T] + func ShareReplayWithConfig[T any](bufferSize int, config ShareReplayConfig) func(Observable[T]) Observable[T] + func ShareReplay[T any](bufferSize int) func(Observable[T]) Observable[T] + func ShareWithConfig[T any](config ShareConfig[T]) func(Observable[T]) Observable[T] + func Share[T any]() func(Observable[T]) Observable[T] + func SkipLast[T any](count int) func(Observable[T]) Observable[T] + func SkipUntil[T, S any](signal Observable[S]) func(Observable[T]) Observable[T] + func SkipWhileIWithContext[T any](...) func(Observable[T]) Observable[T] + func SkipWhileI[T any](predicate func(item T, index int64) bool) func(Observable[T]) Observable[T] + func SkipWhileWithContext[T any](predicate func(ctx context.Context, item T) (context.Context, bool)) func(Observable[T]) Observable[T] + func SkipWhile[T any](predicate func(item T) bool) func(Observable[T]) Observable[T] + func Skip[T any](count int64) func(Observable[T]) Observable[T] + func StartWith[T any](prefixes ...T) func(Observable[T]) Observable[T] + func SubscribeOn[T any](bufferSize int) func(Observable[T]) Observable[T] + func Sum[T constraints.Numeric]() func(Observable[T]) Observable[T] + func Tail[T any]() func(Observable[T]) Observable[T] + func TakeLast[T any](count int) func(Observable[T]) Observable[T] + func TakeUntil[T, S any](signal Observable[S]) func(Observable[T]) Observable[T] + func TakeWhileIWithContext[T any](...) func(Observable[T]) Observable[T] + func TakeWhileI[T any](predicate func(item T, index int64) bool) func(Observable[T]) Observable[T] + func TakeWhileWithContext[T any](predicate func(ctx context.Context, item T) (context.Context, bool)) func(Observable[T]) Observable[T] + func TakeWhile[T any](predicate func(item T) bool) func(Observable[T]) Observable[T] + func Take[T any](count int64) func(Observable[T]) Observable[T] + func TapOnCompleteWithContext[T any](onComplete func(ctx context.Context)) func(Observable[T]) Observable[T] + func TapOnComplete[T any](onComplete func()) func(Observable[T]) Observable[T] + func TapOnErrorWithContext[T any](onError func(ctx context.Context, err error)) func(Observable[T]) Observable[T] + func TapOnError[T any](onError func(err error)) func(Observable[T]) Observable[T] + func TapOnFinalize[T any](onFinalize func()) func(Observable[T]) Observable[T] + func TapOnNextWithContext[T any](onNext func(ctx context.Context, value T)) func(Observable[T]) Observable[T] + func TapOnNext[T any](onNext func(value T)) func(Observable[T]) Observable[T] + func TapOnSubscribeWithContext[T any](onSubscribe func(ctx context.Context)) func(Observable[T]) Observable[T] + func TapOnSubscribe[T any](onSubscribe func()) func(Observable[T]) Observable[T] + func TapWithContext[T any](onNext func(ctx context.Context, value T), ...) func(Observable[T]) Observable[T] + func Tap[T any](onNext func(value T), onError func(err error), onComplete func()) func(Observable[T]) Observable[T] + func ThrottleTime[T any](interval time.Duration) func(Observable[T]) Observable[T] + func ThrottleWhen[T, t any](tick Observable[t]) func(Observable[T]) Observable[T] + func ThrowIfEmpty[T any](throw func() error) func(Observable[T]) Observable[T] + func ThrowOnContextCancel[T any]() func(Observable[T]) Observable[T] + func TimeInterval[T any]() func(Observable[T]) Observable[IntervalValue[T]] + func Timeout[T any](duration time.Duration) func(Observable[T]) Observable[T] + func Timestamp[T any]() func(Observable[T]) Observable[TimestampValue[T]] + func ToChannel[T any](size int) func(Observable[T]) Observable[<-chan Notification[T]] + func ToMapIWithContext[T any, K comparable, V any](mapper func(ctx context.Context, item T, index int64) (K, V)) func(Observable[T]) Observable[map[K]V] + func ToMapI[T any, K comparable, V any](mapper func(item T, index int64) (K, V)) func(Observable[T]) Observable[map[K]V] + func ToMapWithContext[T any, K comparable, V any](project func(ctx context.Context, item T) (K, V)) func(Observable[T]) Observable[map[K]V] + func ToMap[T any, K comparable, V any](project func(item T) (K, V)) func(Observable[T]) Observable[map[K]V] + func ToSlice[T any]() func(Observable[T]) Observable[[]T] + func Trunc() func(Observable[float64]) Observable[float64] + func WhileIWithContext[T any](condition func(ctx context.Context, index int64) (context.Context, bool)) func(Observable[T]) Observable[T] + func WhileI[T any](condition func(index int64) bool) func(Observable[T]) Observable[T] + func WhileWithContext[T any](condition func(ctx context.Context) (context.Context, bool)) func(Observable[T]) Observable[T] + func While[T any](condition func() bool) func(Observable[T]) Observable[T] + func WindowWhen[T, B any](boundary Observable[B]) func(Observable[T]) Observable[Observable[T]] + func ZipAll[T any]() func(Observable[Observable[T]]) Observable[[]T] + func ZipWith1[A, B any](obsB Observable[B]) func(Observable[A]) Observable[lo.Tuple2[A, B]] + func ZipWith2[A, B, C any](obsB Observable[B], obsC Observable[C]) func(Observable[A]) Observable[lo.Tuple3[A, B, C]] + func ZipWith3[A, B, C, D any](obsB Observable[B], obsC Observable[C], obsD Observable[D]) func(Observable[A]) Observable[lo.Tuple4[A, B, C, D]] + func ZipWith4[A, B, C, D, E any](obsB Observable[B], obsC Observable[C], obsD Observable[D], obsE Observable[E]) func(Observable[A]) Observable[lo.Tuple5[A, B, C, D, E]] + func ZipWith5[A, B, C, D, E, F any](obsB Observable[B], obsC Observable[C], obsD Observable[D], obsE Observable[E], ...) func(Observable[A]) Observable[lo.Tuple6[A, B, C, D, E, F]] + func ZipWith[A, B any](obsB Observable[B]) func(Observable[A]) Observable[lo.Tuple2[A, B]] + type Backpressure int8 + const BackpressureBlock + const BackpressureDrop + type ConcurrencyMode int8 + const ConcurrencyModeEventuallySafe + const ConcurrencyModeSafe + const ConcurrencyModeUnsafe + type ConnectableConfig struct + Connector func() Subject[T] + ResetOnDisconnect bool + type ConnectableObservable interface + Connect func() Subscription + ConnectWithContext func(ctx context.Context) Subscription + func ConnectableWithConfig[T any](source Observable[T], config ConnectableConfig[T]) ConnectableObservable[T] + func Connectable[T any](source Observable[T]) ConnectableObservable[T] + func NewConnectableObservableWithConfigAndContext[T any](subscribe func(ctx context.Context, destination Observer[T]) Teardown, ...) ConnectableObservable[T] + func NewConnectableObservableWithConfig[T any](subscribe func(destination Observer[T]) Teardown, config ConnectableConfig[T]) ConnectableObservable[T] + func NewConnectableObservableWithContext[T any](subscribe func(ctx context.Context, destination Observer[T]) Teardown) ConnectableObservable[T] + func NewConnectableObservable[T any](subscribe func(destination Observer[T]) Teardown) ConnectableObservable[T] + type IntervalValue struct + Interval time.Duration + Value T + type Kind uint8 + const KindComplete + const KindError + const KindNext + func (k Kind) String() string + type Notification struct + Err error + Kind Kind + Value T + func NewNotificationComplete[T any]() Notification[T] + func NewNotificationError[T any](err error) Notification[T] + func NewNotificationNext[T any](value T) Notification[T] + func (n Notification[T]) String() string + type Observable interface + Subscribe func(destination Observer[T]) Subscription + SubscribeWithContext func(ctx context.Context, destination Observer[T]) Subscription + func Amb[T any](sources ...Observable[T]) Observable[T] + func CombineLatest2[A, B any](obsA Observable[A], obsB Observable[B]) Observable[lo.Tuple2[A, B]] + func CombineLatest3[A, B, C any](obsA Observable[A], obsB Observable[B], obsC Observable[C]) Observable[lo.Tuple3[A, B, C]] + func CombineLatest4[A, B, C, D any](obsA Observable[A], obsB Observable[B], obsC Observable[C], obsD Observable[D]) Observable[lo.Tuple4[A, B, C, D]] + func CombineLatest5[A, B, C, D, E any](obsA Observable[A], obsB Observable[B], obsC Observable[C], obsD Observable[D], ...) Observable[lo.Tuple5[A, B, C, D, E]] + func CombineLatestAny(sources ...Observable[any]) Observable[[]any] + func Concat[T any](obs ...Observable[T]) Observable[T] + func Defer[T any](factory func() Observable[T]) Observable[T] + func Empty[T any]() Observable[T] + func FromChannel[T any](in <-chan T) Observable[T] + func FromSlice[T any](collections ...[]T) Observable[T] + func Future[T any](factory func() (T, error)) Observable[T] + func Interval(interval time.Duration) Observable[int64] + func IntervalWithInitial(initial, interval time.Duration) Observable[int64] + func Just[T any](values ...T) Observable[T] + func Merge[T any](sources ...Observable[T]) Observable[T] + func Never() Observable[struct{}] + func NewEventuallySafeObservableWithContext[T any](subscribe func(ctx context.Context, destination Observer[T]) Teardown) Observable[T] + func NewEventuallySafeObservable[T any](subscribe func(destination Observer[T]) Teardown) Observable[T] + func NewObservableWithConcurrencyMode[T any](subscribe func(ctx context.Context, destination Observer[T]) Teardown, ...) Observable[T] + func NewObservableWithContext[T any](subscribe func(ctx context.Context, destination Observer[T]) Teardown) Observable[T] + func NewObservable[T any](subscribe func(destination Observer[T]) Teardown) Observable[T] + func NewSafeObservableWithContext[T any](subscribe func(ctx context.Context, destination Observer[T]) Teardown) Observable[T] + func NewSafeObservable[T any](subscribe func(destination Observer[T]) Teardown) Observable[T] + func NewUnsafeObservableWithContext[T any](subscribe func(ctx context.Context, destination Observer[T]) Teardown) Observable[T] + func NewUnsafeObservable[T any](subscribe func(destination Observer[T]) Teardown) Observable[T] + func Of[T any](values ...T) Observable[T] + func Pipe10[A, B, C, D, E, F, G, H, I, J, K any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[K] + func Pipe11[A, B, C, D, E, F, G, H, I, J, K, L any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[L] + func Pipe12[A, B, C, D, E, F, G, H, I, J, K, L, M any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[M] + func Pipe13[A, B, C, D, E, F, G, H, I, J, K, L, M, N any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[N] + func Pipe14[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[O] + func Pipe15[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[P] + func Pipe16[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[Q] + func Pipe17[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[R] + func Pipe18[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[S] + func Pipe19[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[T] + func Pipe1[A, B any](source Observable[A], operator1 func(Observable[A]) Observable[B]) Observable[B] + func Pipe20[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[U] + func Pipe21[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[V] + func Pipe22[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[W] + func Pipe23[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W, X any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[X] + func Pipe24[A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W, X, Y any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[Y] + func Pipe25[...](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[Z] + func Pipe2[A, B, C any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[C] + func Pipe3[A, B, C, D any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[D] + func Pipe4[A, B, C, D, E any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[E] + func Pipe5[A, B, C, D, E, F any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[F] + func Pipe6[A, B, C, D, E, F, G any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[G] + func Pipe7[A, B, C, D, E, F, G, H any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[H] + func Pipe8[A, B, C, D, E, F, G, H, I any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[I] + func Pipe9[A, B, C, D, E, F, G, H, I, J any](source Observable[A], operator1 func(Observable[A]) Observable[B], ...) Observable[J] + func Pipe[First, Last any](source Observable[First], operators ...any) Observable[Last] + func Race[T any](sources ...Observable[T]) Observable[T] + func RandFloat64(count int) Observable[float64] + func RandIntN(n, count int) Observable[int] + func Range(start, end int64) Observable[int64] + func RangeWithInterval(start, end int64, interval time.Duration) Observable[int64] + func RangeWithStep(start, end, step float64) Observable[float64] + func RangeWithStepAndInterval(start, end, step float64, interval time.Duration) Observable[float64] + func RepeatWithInterval[T any](item T, count int64, interval time.Duration) Observable[T] + func Repeat[T any](item T, count int64) Observable[T] + func Start[T any](cb func() T) Observable[T] + func Throw[T any](err error) Observable[T] + func Timer(duration time.Duration) Observable[time.Duration] + func Zip2[A, B any](obsA Observable[A], obsB Observable[B]) Observable[lo.Tuple2[A, B]] + func Zip3[A, B, C any](obsA Observable[A], obsB Observable[B], obsC Observable[C]) Observable[lo.Tuple3[A, B, C]] + func Zip4[A, B, C, D any](obsA Observable[A], obsB Observable[B], obsC Observable[C], obsD Observable[D]) Observable[lo.Tuple4[A, B, C, D]] + func Zip5[A, B, C, D, E any](obsA Observable[A], obsB Observable[B], obsC Observable[C], obsD Observable[D], ...) Observable[lo.Tuple5[A, B, C, D, E]] + func Zip6[A, B, C, D, E, F any](obsA Observable[A], obsB Observable[B], obsC Observable[C], obsD Observable[D], ...) Observable[lo.Tuple6[A, B, C, D, E, F]] + func Zip[T any](sources ...Observable[T]) Observable[[]T] + type Observer interface + Complete func() + CompleteWithContext func(ctx context.Context) + Error func(err error) + ErrorWithContext func(ctx context.Context, err error) + HasThrown func() bool + IsClosed func() bool + IsCompleted func() bool + Next func(value T) + NextWithContext func(ctx context.Context, value T) + func NewObserverWithContext[T any](onNext func(ctx context.Context, value T), ...) Observer[T] + func NewObserver[T any](onNext func(value T), onError func(err error), onComplete func()) Observer[T] + func NoopObserver[T any]() Observer[T] + func OnCompleteWithContext[T any](onComplete func(ctx context.Context)) Observer[T] + func OnComplete[T any](onComplete func()) Observer[T] + func OnErrorWithContext[T any](onError func(ctx context.Context, err error)) Observer[T] + func OnError[T any](onError func(err error)) Observer[T] + func OnNextWithContext[T any](onNext func(ctx context.Context, value T)) Observer[T] + func OnNext[T any](onNext func(value T)) Observer[T] + func PrintObserver[T any]() Observer[T] + type RetryConfig struct + Delay time.Duration + MaxRetries uint64 + ResetOnSuccess bool + type ShareConfig struct + Connector func() Subject[T] + ResetOnComplete bool + ResetOnError bool + ResetOnRefCountZero bool + type ShareReplayConfig struct + ResetOnRefCountZero bool + type Subject interface + AsObservable func() Observable[T] + AsObserver func() Observer[T] + CountObservers func() int + HasObserver func() bool + HasThrown func() bool + IsClosed func() bool + IsCompleted func() bool + func NewAsyncSubject[T any]() Subject[T] + func NewBehaviorSubject[T any](initial T) Subject[T] + func NewPublishSubject[T any]() Subject[T] + func NewReplaySubject[T any](bufferSize int) Subject[T] + func NewSubject[T any]() Subject[T] + func NewUnicastSubject[T any](bufferSize int) Subject[T] + type Subscriber interface + func NewEventuallySafeSubscriber[T any](destination Observer[T]) Subscriber[T] + func NewSafeSubscriber[T any](destination Observer[T]) Subscriber[T] + func NewSubscriberWithConcurrencyMode[T any](destination Observer[T], mode ConcurrencyMode) Subscriber[T] + func NewSubscriber[T any](destination Observer[T]) Subscriber[T] + func NewUnsafeSubscriber[T any](destination Observer[T]) Subscriber[T] + type Subscription interface + Add func(teardown Teardown) + AddUnsubscribable func(unsubscribable Unsubscribable) + IsClosed func() bool + Wait func() + func NewSubscription(teardown Teardown) Subscription + type Teardown func() + type TimestampValue struct + Timestamp time.Duration + Value T + type Unsubscribable interface + Unsubscribe func() v0.2.0 Oct 25, 2025 v0.1.0 Oct 14, 2025
Go to main page