rxpd

Documentation

Index

• func WithLogReadWrite[T any](id string) func(*propertyOption[T])
• func WithRead[T any](read func() rx.ResultChan[T]) func(*propertyOption[T])
• func WithReadCallsNext[T any]() func(*propertyOption[T])
• func WithReadTimeout[T any](timeout time.Duration, onTimeout func() rx.Result[T]) func(*propertyOption[T])
• func WithSetInterval[T any](setInterval func(time.Duration)) func(*propertyOption[T])
• func WithTapRead[T any](tapRead func(func() rx.ResultChan[T]) rx.ResultChan[T]) func(*propertyOption[T])
• func WithTapWrite[T any](tapWrite func(T, func(T) <-chan error) <-chan error) func(*propertyOption[T])
• func WithWrite[T any](write func(T) <-chan error) func(*propertyOption[T])
• type Property
  • func CombineLatest[T any](combine func(next ...any) T, sources ...Subscribable[any]) Property[T]
  • func CombineLatest2[S1 any, S2 any, T any](combine func(S1, S2) T, s1 Subscribable[S1], s2 Subscribable[S2]) Property[T]
  • func CombineLatest3[S1 any, S2 any, S3 any, T any](combine func(S1, S2, S3) T, s1 Subscribable[S1], s2 Subscribable[S2], ...) Property[T]
  • func CombineLatest4[S1 any, S2 any, S3 any, S4 any, T any](combine func(S1, S2, S3, S4) T, s1 Subscribable[S1], s2 Subscribable[S2], ...) Property[T]
  • func CombineLatest5[S1 any, S2 any, S3 any, S4 any, S5 any, T any](combine func(S1, S2, S3, S4, S5) T, s1 Subscribable[S1], s2 Subscribable[S2], ...) Property[T]
  • func CombineLatest6[S1 any, S2 any, S3 any, S4 any, S5 any, S6 any, T any](combine func(S1, S2, S3, S4, S5, S6) T, s1 Subscribable[S1], ...) Property[T]
  • func CombineLatest7[S1 any, S2 any, S3 any, S4 any, S5 any, S6 any, S7 any, T any](combine func(S1, S2, S3, S4, S5, S6, S7) T, s1 Subscribable[S1], ...) Property[T]
  • func CombineLatest8[S1 any, S2 any, S3 any, S4 any, S5 any, S6 any, S7 any, S8 any, T any](combine func(S1, S2, S3, S4, S5, S6, S7, S8) T, s1 Subscribable[S1], ...) Property[T]
  • func CombineLatest9[S1 any, S2 any, S3 any, S4 any, S5 any, S6 any, S7 any, S8 any, S9 any, T any](combine func(S1, S2, S3, S4, S5, S6, S7, S8, S9) T, s1 Subscribable[S1], ...) Property[T]
  • func CombineLatestToSlice[T any](sst []Subscribable[T]) Property[[]T]
  • func Concat[T any](sources ...Subscribable[T]) Property[T]
  • func FromRead[T any](ctx context.Context, read func() rx.ResultChan[T], ...) Property[T]
  • func Map[T any, U any](s Subscribable[T], mapToU func(T) U, mapToT func(U) (T, error)) Property[U]
  • func ToAny[T any](s Subscribable[T]) Property[any]
  • func ToProperty[T any](s rx.Subscribable[T], options ...PropertyOption[T]) Property[T]
• type PropertyOption
• type PropertySource
  • func NewPropertySource[ID comparable, T any](createProperty func(ID) Property[T]) *PropertySource[ID, T]
  • func (ps *PropertySource[ID, T]) GetProperty(id ID) Property[T]
• type Subscribable

Functions

func WithLogReadWrite

func WithLogReadWrite[T any](id string) func(*propertyOption[T])

func WithRead

func WithRead[T any](read func() rx.ResultChan[T]) func(*propertyOption[T])

func WithReadCallsNext

func WithReadCallsNext[T any]() func(*propertyOption[T])

WithReadCallsNext configures that the result from the Read rx.ResultChan is passed to the Next function. When used together with WithReadTimeout and the timeout elapsed, first the timeout value is passed to the Next function. When Read finally returns, this value is also passed to the Next function.

func WithReadTimeout

func WithReadTimeout[T any](timeout time.Duration, onTimeout func() rx.Result[T]) func(*propertyOption[T])

WithReadTimeout configures a timeout for the Read function. If the timeout is reached, the onTimeout function is called and its result is returned.

func WithSetInterval

func WithSetInterval[T any](setInterval func(time.Duration)) func(*propertyOption[T])

func WithTapRead

func WithTapRead[T any](tapRead func(func() rx.ResultChan[T]) rx.ResultChan[T]) func(*propertyOption[T])

func WithTapWrite

func WithTapWrite[T any](tapWrite func(T, func(T) <-chan error) <-chan error) func(*propertyOption[T])

func WithWrite

func WithWrite[T any](write func(T) <-chan error) func(*propertyOption[T])

Types

type Property

type Property[T any] interface {
	Subscribable[T]
	rx.ObservableExtension[Property[T], Subscribable[T], T]

	ToAny() Property[any]
	With(options ...PropertyOption[T]) Property[T]
}

func CombineLatest

func CombineLatest[T any](combine func(next ...any) T, sources ...Subscribable[any]) Property[T]

func CombineLatest2

func CombineLatest2[S1 any, S2 any, T any](combine func(S1, S2) T,
	s1 Subscribable[S1], s2 Subscribable[S2]) Property[T]

func CombineLatest3

func CombineLatest3[S1 any, S2 any, S3 any, T any](combine func(S1, S2, S3) T,
	s1 Subscribable[S1], s2 Subscribable[S2], s3 Subscribable[S3]) Property[T]

func CombineLatest4

func CombineLatest4[S1 any, S2 any, S3 any, S4 any, T any](combine func(S1, S2, S3, S4) T,
	s1 Subscribable[S1], s2 Subscribable[S2], s3 Subscribable[S3], s4 Subscribable[S4]) Property[T]

func CombineLatest5

func CombineLatest5[S1 any, S2 any, S3 any, S4 any, S5 any, T any](combine func(S1, S2, S3, S4, S5) T,
	s1 Subscribable[S1], s2 Subscribable[S2], s3 Subscribable[S3], s4 Subscribable[S4],
	s5 Subscribable[S5]) Property[T]

func CombineLatest6

func CombineLatest6[S1 any, S2 any, S3 any, S4 any, S5 any, S6 any, T any](
	combine func(S1, S2, S3, S4, S5, S6) T,
	s1 Subscribable[S1], s2 Subscribable[S2], s3 Subscribable[S3], s4 Subscribable[S4],
	s5 Subscribable[S5], s6 Subscribable[S6]) Property[T]

func CombineLatest7

func CombineLatest7[S1 any, S2 any, S3 any, S4 any, S5 any, S6 any, S7 any, T any](
	combine func(S1, S2, S3, S4, S5, S6, S7) T,
	s1 Subscribable[S1], s2 Subscribable[S2], s3 Subscribable[S3], s4 Subscribable[S4],
	s5 Subscribable[S5], s6 Subscribable[S6], s7 Subscribable[S7]) Property[T]

func CombineLatest8

func CombineLatest8[S1 any, S2 any, S3 any, S4 any, S5 any, S6 any, S7 any, S8 any, T any](
	combine func(S1, S2, S3, S4, S5, S6, S7, S8) T,
	s1 Subscribable[S1], s2 Subscribable[S2], s3 Subscribable[S3], s4 Subscribable[S4],
	s5 Subscribable[S5], s6 Subscribable[S6], s7 Subscribable[S7], s8 Subscribable[S8]) Property[T]

func CombineLatest9

func CombineLatest9[S1 any, S2 any, S3 any, S4 any, S5 any, S6 any, S7 any, S8 any, S9 any, T any](
	combine func(S1, S2, S3, S4, S5, S6, S7, S8, S9) T,
	s1 Subscribable[S1], s2 Subscribable[S2], s3 Subscribable[S3], s4 Subscribable[S4],
	s5 Subscribable[S5], s6 Subscribable[S6], s7 Subscribable[S7], s8 Subscribable[S8],
	s9 Subscribable[S9]) Property[T]

func CombineLatestToSlice

func CombineLatestToSlice[T any](sst []Subscribable[T]) Property[[]T]

func Concat

func Concat[T any](sources ...Subscribable[T]) Property[T]

Concat creates an output Property which sequentially emits all values from the first given Subscribable and then moves on to the next.

func FromRead

func FromRead[T any](ctx context.Context, read func() rx.ResultChan[T], options ...PropertyOption[T]) Property[T]

FromRead creates a Property from a WithRead option. The reactive stream is created by a goroutine which calls the read function in the interval set by SetInterval.

func Map

func Map[T any, U any](s Subscribable[T], mapToU func(T) U, mapToT func(U) (T, error)) Property[U]

func ToAny

func ToAny[T any](s Subscribable[T]) Property[any]

func ToProperty

func ToProperty[T any](s rx.Subscribable[T], options ...PropertyOption[T]) Property[T]

ToProperty extends a Subscribable to a Property

type PropertyOption

type PropertyOption[T any] func(*propertyOption[T])

type PropertySource

type PropertySource[ID comparable, T any] struct {
	// contains filtered or unexported fields
}

PropertySource creates and stores Property objects by IDs that can be reused.

func NewPropertySource

func NewPropertySource[ID comparable, T any](createProperty func(ID) Property[T]) *PropertySource[ID, T]

NewPropertySource creates a PropertySource which creates Property objects for IDs with the createProperty function.

func (*PropertySource[ID, T]) GetProperty

func (ps *PropertySource[ID, T]) GetProperty(id ID) Property[T]

GetProperty returns the Property object for an ID. The Property has ShareReplay behavior with MaxBufferSize 1. This means each new subscriber will get the last value of the Property, if there is any.

type Subscribable

type Subscribable[T any] interface {
	rx.Subscribable[T]

	SetInterval(time.Duration)
	Interval() time.Duration

	Read() rx.ResultChan[T]
	Write(T) <-chan error
}