rx

package

Go to main page

Versions in this module

v1

v1.3.7

Jul 15, 2021

v1.3.6

Jun 28, 2021

v1.3.5

Jun 24, 2021

v1.3.4

Jun 21, 2021

v1.3.3

Jun 18, 2021

v1.3.2

Jun 11, 2021

v1.3.1

Jun 7, 2021

v1.3.0

Jun 4, 2021

v1.2.520

May 20, 2021

Changes in this version

type RxStream

+ func FromReaderWithDecoder(readers chan io.Reader) RxStream

v1.1.0

Apr 30, 2021

Changes in this version

+ type Handler func(interface{}) error

type RxStream

+ SlidingWindowWithCount func(windowSize int, slideSize int, handler Handler, opts ...rxgo.Option) RxStream

+ SlidingWindowWithTime func(windowTimeInMS uint32, slideTimeInMS uint32, handler Handler, ...) RxStream

+ ZipMultiObservers func(observers []yy3.KeyObserveFunc, ...) RxStream

type RxStreamImpl

+ func (s *RxStreamImpl) SlidingWindowWithCount(windowSize int, slideSize int, handler Handler, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) SlidingWindowWithTime(windowTimeInMS uint32, slideTimeInMS uint32, handler Handler, ...) RxStream

+ func (s *RxStreamImpl) ZipMultiObservers(observers []yy3.KeyObserveFunc, ...) RxStream

v1.0.0

Mar 1, 2021

v0

v0.7.7

Feb 23, 2021

Changes in this version

type RxStream

+ Encode func(key byte, opts ...rxgo.Option) RxStream

type RxStreamImpl

+ func (s *RxStreamImpl) Encode(key byte, opts ...rxgo.Option) RxStream

v0.7.6

Feb 2, 2021

v0.7.5

Jan 26, 2021

v0.7.4

Jan 26, 2021

v0.7.3

Jan 21, 2021

Changes in this version

type RxStream

+ func FromReaderWithY3(reader io.Reader) RxStream

v0.7.2

Jan 20, 2021

Changes in this version

type RxStream

+ func ConvertObservableWithY3(observable y3.Observable) RxStream

+ OnObserve func(function func(v []byte) (interface{}, error)) RxStream

+ Subscribe func(key byte) RxStream

type RxStreamImpl

+ func (s *RxStreamImpl) OnObserve(function func(v []byte) (interface{}, error)) RxStream

+ func (s *RxStreamImpl) Subscribe(key byte) RxStream

v0.7.1

Jan 11, 2021

v0.7.0

Jan 6, 2021

v0.6.0

Dec 30, 2020

Changes in this version

type RxStream

+ func FromChannel(channel chan []byte) RxStream

+ func FromReaderWithFunc(f func() io.Reader) RxStream

+ MergeReadWriterWithFunc func(rwf func() (io.ReadWriter, func()), opts ...rxgo.Option) RxStream

type RxStreamImpl

+ func (s *RxStreamImpl) MergeReadWriterWithFunc(rwf func() (io.ReadWriter, func()), opts ...rxgo.Option) RxStream

v0.5.2

Dec 29, 2020

v0.5.1

Dec 25, 2020

v0.5.0

Dec 23, 2020

v0.4.1

Dec 19, 2020

v0.4.0

Dec 18, 2020

Changes in this version

+ func Of(i interface{}) rxgo.Item

+ type RxStream interface

+ All func(predicate rxgo.Predicate, opts ...rxgo.Option) RxStream

+ AuditTime func(timespan time.Duration, opts ...rxgo.Option) RxStream

+ AverageFloat32 func(opts ...rxgo.Option) RxStream

+ AverageFloat64 func(opts ...rxgo.Option) RxStream

+ AverageInt func(opts ...rxgo.Option) RxStream

+ AverageInt16 func(opts ...rxgo.Option) RxStream

+ AverageInt32 func(opts ...rxgo.Option) RxStream

+ AverageInt64 func(opts ...rxgo.Option) RxStream

+ AverageInt8 func(opts ...rxgo.Option) RxStream

+ BackOffRetry func(backOffCfg backoff.BackOff, opts ...rxgo.Option) RxStream

+ BufferWithCount func(count int, opts ...rxgo.Option) RxStream

+ BufferWithTime func(timespan rxgo.Duration, opts ...rxgo.Option) RxStream

+ BufferWithTimeOrCount func(timespan rxgo.Duration, count int, opts ...rxgo.Option) RxStream

+ Connect func(ctx context.Context) (context.Context, rxgo.Disposable)

+ Contains func(equal rxgo.Predicate, opts ...rxgo.Option) RxStream

+ Count func(opts ...rxgo.Option) RxStream

+ Debounce func(timespan rxgo.Duration, opts ...rxgo.Option) RxStream

+ DefaultIfEmpty func(defaultValue interface{}, opts ...rxgo.Option) RxStream

+ DefaultIfEmptyWithTime func(timespan time.Duration, defaultValue interface{}, opts ...rxgo.Option) RxStream

+ Distinct func(apply rxgo.Func, opts ...rxgo.Option) RxStream

+ DistinctUntilChanged func(apply rxgo.Func, opts ...rxgo.Option) RxStream

+ DoOnCompleted func(completedFunc rxgo.CompletedFunc, opts ...rxgo.Option) rxgo.Disposed

+ DoOnError func(errFunc rxgo.ErrFunc, opts ...rxgo.Option) rxgo.Disposed

+ DoOnNext func(nextFunc rxgo.NextFunc, opts ...rxgo.Option) rxgo.Disposed

+ ElementAt func(index uint, opts ...rxgo.Option) RxStream

+ Error func(opts ...rxgo.Option) error

+ Errors func(opts ...rxgo.Option) []error

+ Filter func(apply rxgo.Predicate, opts ...rxgo.Option) RxStream

+ Find func(find rxgo.Predicate, opts ...rxgo.Option) RxStream

+ First func(opts ...rxgo.Option) RxStream

+ FirstOrDefault func(defaultValue interface{}, opts ...rxgo.Option) RxStream

+ FlatMap func(apply rxgo.ItemToObservable, opts ...rxgo.Option) RxStream

+ ForEach func(nextFunc rxgo.NextFunc, errFunc rxgo.ErrFunc, completedFunc rxgo.CompletedFunc, ...) rxgo.Disposed

+ GroupBy func(length int, distribution func(rxgo.Item) int, opts ...rxgo.Option) RxStream

+ GroupByDynamic func(distribution func(rxgo.Item) string, opts ...rxgo.Option) RxStream

+ IgnoreElements func(opts ...rxgo.Option) RxStream

+ Join func(joiner rxgo.Func2, right rxgo.Observable, ...) RxStream

+ Last func(opts ...rxgo.Option) RxStream

+ LastOrDefault func(defaultValue interface{}, opts ...rxgo.Option) RxStream

+ Map func(apply rxgo.Func, opts ...rxgo.Option) RxStream

+ Marshal func(marshaller rxgo.Marshaller, opts ...rxgo.Option) RxStream

+ Max func(comparator rxgo.Comparator, opts ...rxgo.Option) RxStream

+ Min func(comparator rxgo.Comparator, opts ...rxgo.Option) RxStream

+ OnErrorResumeNext func(resumeSequence rxgo.ErrorToObservable, opts ...rxgo.Option) RxStream

+ OnErrorReturn func(resumeFunc rxgo.ErrorFunc, opts ...rxgo.Option) RxStream

+ OnErrorReturnItem func(resume interface{}, opts ...rxgo.Option) RxStream

+ Reduce func(apply rxgo.Func2, opts ...rxgo.Option) RxStream

+ Repeat func(count int64, frequency rxgo.Duration, opts ...rxgo.Option) RxStream

+ Retry func(count int, shouldRetry func(error) bool, opts ...rxgo.Option) RxStream

+ Run func(opts ...rxgo.Option) rxgo.Disposed

+ Sample func(iterable rxgo.Iterable, opts ...rxgo.Option) RxStream

+ Scan func(apply rxgo.Func2, opts ...rxgo.Option) RxStream

+ Send func(output chan<- rxgo.Item, opts ...rxgo.Option)

+ SequenceEqual func(iterable rxgo.Iterable, opts ...rxgo.Option) RxStream

+ Serialize func(from int, identifier func(interface{}) int, opts ...rxgo.Option) RxStream

+ Skip func(nth uint, opts ...rxgo.Option) RxStream

+ SkipLast func(nth uint, opts ...rxgo.Option) RxStream

+ SkipWhile func(apply rxgo.Predicate, opts ...rxgo.Option) RxStream

+ StartWith func(iterable rxgo.Iterable, opts ...rxgo.Option) RxStream

+ StdOut func(opts ...rxgo.Option) RxStream

+ SumFloat32 func(opts ...rxgo.Option) RxStream

+ SumFloat64 func(opts ...rxgo.Option) RxStream

+ SumInt64 func(opts ...rxgo.Option) RxStream

+ Take func(nth uint, opts ...rxgo.Option) RxStream

+ TakeLast func(nth uint, opts ...rxgo.Option) RxStream

+ TakeUntil func(apply rxgo.Predicate, opts ...rxgo.Option) RxStream

+ TakeWhile func(apply rxgo.Predicate, opts ...rxgo.Option) RxStream

+ TimeInterval func(opts ...rxgo.Option) RxStream

+ Timestamp func(opts ...rxgo.Option) RxStream

+ ToMap func(keySelector rxgo.Func, opts ...rxgo.Option) RxStream

+ ToMapWithValueSelector func(keySelector, valueSelector rxgo.Func, opts ...rxgo.Option) RxStream

+ ToSlice func(initialCapacity int, opts ...rxgo.Option) ([]interface{}, error)

+ Unmarshal func(unmarshaller rxgo.Unmarshaller, factory func() interface{}, ...) RxStream

+ WindowWithCount func(count int, opts ...rxgo.Option) RxStream

+ WindowWithTime func(timespan rxgo.Duration, opts ...rxgo.Option) RxStream

+ WindowWithTimeOrCount func(timespan rxgo.Duration, count int, opts ...rxgo.Option) RxStream

+ Y3Decoder func(key string, mold interface{}, opts ...rxgo.Option) RxStream

+ ZipFromIterable func(iterable rxgo.Iterable, zipper rxgo.Func2, opts ...rxgo.Option) RxStream

+ func ConvertObservable(observable rxgo.Observable) RxStream

+ func CreateObservable(f func(ctx context.Context, next chan rxgo.Item), opts ...rxgo.Option) RxStream

+ func FromReader(reader io.Reader) RxStream

+ type RxStreamImpl struct

+ func (s *RxStreamImpl) All(predicate rxgo.Predicate, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) AuditTime(timespan time.Duration, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) AverageFloat32(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) AverageFloat64(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) AverageInt(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) AverageInt16(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) AverageInt32(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) AverageInt64(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) AverageInt8(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) BackOffRetry(backOffCfg backoff.BackOff, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) BufferWithCount(count int, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) BufferWithTime(timespan rxgo.Duration, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) BufferWithTimeOrCount(timespan rxgo.Duration, count int, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Connect(ctx context.Context) (context.Context, rxgo.Disposable)

+ func (s *RxStreamImpl) Contains(equal rxgo.Predicate, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Count(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Debounce(timespan rxgo.Duration, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) DefaultIfEmpty(defaultValue interface{}, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) DefaultIfEmptyWithTime(timespan time.Duration, defaultValue interface{}, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Distinct(apply rxgo.Func, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) DistinctUntilChanged(apply rxgo.Func, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) DoOnCompleted(completedFunc rxgo.CompletedFunc, opts ...rxgo.Option) rxgo.Disposed

+ func (s *RxStreamImpl) DoOnError(errFunc rxgo.ErrFunc, opts ...rxgo.Option) rxgo.Disposed

+ func (s *RxStreamImpl) DoOnNext(nextFunc rxgo.NextFunc, opts ...rxgo.Option) rxgo.Disposed

+ func (s *RxStreamImpl) ElementAt(index uint, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Error(opts ...rxgo.Option) error

+ func (s *RxStreamImpl) Errors(opts ...rxgo.Option) []error

+ func (s *RxStreamImpl) Filter(apply rxgo.Predicate, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Find(find rxgo.Predicate, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) First(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) FirstOrDefault(defaultValue interface{}, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) FlatMap(apply rxgo.ItemToObservable, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) ForEach(nextFunc rxgo.NextFunc, errFunc rxgo.ErrFunc, completedFunc rxgo.CompletedFunc, ...) rxgo.Disposed

+ func (s *RxStreamImpl) GroupBy(length int, distribution func(rxgo.Item) int, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) GroupByDynamic(distribution func(rxgo.Item) string, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) IgnoreElements(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Join(joiner rxgo.Func2, right rxgo.Observable, ...) RxStream

+ func (s *RxStreamImpl) Last(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) LastOrDefault(defaultValue interface{}, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Map(apply rxgo.Func, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Marshal(marshaller rxgo.Marshaller, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Max(comparator rxgo.Comparator, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Min(comparator rxgo.Comparator, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Observe(opts ...rxgo.Option) <-chan rxgo.Item

+ func (s *RxStreamImpl) OnErrorResumeNext(resumeSequence rxgo.ErrorToObservable, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) OnErrorReturn(resumeFunc rxgo.ErrorFunc, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) OnErrorReturnItem(resume interface{}, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Reduce(apply rxgo.Func2, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Repeat(count int64, frequency rxgo.Duration, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Retry(count int, shouldRetry func(error) bool, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Run(opts ...rxgo.Option) rxgo.Disposed

+ func (s *RxStreamImpl) Sample(iterable rxgo.Iterable, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Scan(apply rxgo.Func2, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Send(output chan<- rxgo.Item, opts ...rxgo.Option)

+ func (s *RxStreamImpl) SequenceEqual(iterable rxgo.Iterable, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Serialize(from int, identifier func(interface{}) int, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Skip(nth uint, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) SkipLast(nth uint, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) SkipWhile(apply rxgo.Predicate, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) StartWith(iterable rxgo.Iterable, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) StdOut(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) SumFloat32(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) SumFloat64(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) SumInt64(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Take(nth uint, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) TakeLast(nth uint, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) TakeUntil(apply rxgo.Predicate, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) TakeWhile(apply rxgo.Predicate, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) TimeInterval(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Timestamp(opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) ToMap(keySelector rxgo.Func, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) ToMapWithValueSelector(keySelector, valueSelector rxgo.Func, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) ToSlice(initialCapacity int, opts ...rxgo.Option) ([]interface{}, error)

+ func (s *RxStreamImpl) Unmarshal(unmarshaller rxgo.Unmarshaller, factory func() interface{}, ...) RxStream

+ func (s *RxStreamImpl) WindowWithCount(count int, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) WindowWithTime(timespan rxgo.Duration, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) WindowWithTimeOrCount(timespan rxgo.Duration, count int, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) Y3Decoder(key string, mold interface{}, opts ...rxgo.Option) RxStream

+ func (s *RxStreamImpl) ZipFromIterable(iterable rxgo.Iterable, zipper rxgo.Func2, opts ...rxgo.Option) RxStream

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL