DOCSEARCHbyAlgolia
Submit Docs

RxGo Documentation

repository·master·Indexed 26 days ago

https://github.com/reactivex/rxgo

A Reactive Extensions implementation for the Go language providing an API for programming with asynchronous Observable streams. RxGo leverages Go's native channels and goroutines to build processing pipelines, supporting Hot and Cold Observables, Connectable Observables, and various transformation, filtering, and combination operators. It includes features for backpressure strategies, worker pool parallelization, and a dedicated Assert API for unit testing reactive streams.

Tokens
24K
Snippets
101
Records
216
Agent score
88%

What's inside RxGo

  1. Use the Take operator to limit emitted items

    master
    The Take operator limits the number of items emitted by an Observable. It will only emit the first n items and then complete. If the Observable completes before n items are emitted, it simply emits whatever was available.

  2. Use the TakeUntil operator to limit emissions

    master

    The TakeUntil operator discards any items emitted by an Observable once a second Observable emits an item or terminates. You can provide a predicate function to determine when the termination condition is met.

    observable := rxgo.Just(1, 2, 3, 4, 5)().TakeUntil(func(i interface{}) bool {
	return i == 3
})

  3. Use the Run operator to execute an Observable without consuming items

    master

    The Run operator allows you to create an Observer that executes the Observable's lifecycle but does not consume the emitted items. It returns a receive-only channel of empty structs (<-chan struct{}) which closes once the Observable terminates. This is useful when you want to trigger side effects or ensure an Observable completes without needing to process its data.

    <-rxgo.Just(1, 2, errors.New("foo"))().Run()

  4. Use the Distinct operator to suppress duplicate items

    master

    The Distinct operator filters an Observable to suppress duplicate items. You provide a function that determines the uniqueness of an item. If the function returns the same value for two items, the second item is suppressed.

    observable := rxgo.Just(1, 2, 2, 3, 4, 4, 5)().
	Distinct(func(_ context.Context, i interface{}) (interface{}, error) {
		return i, nil
	})

  5. Use the BackOffRetry operator to handle errors with exponential backoff

    master

    The BackOffRetry operator implements a retry mechanism that resubscribes to a source Observable if it emits an error. It uses the github.com/cenkalti/backoff/v4 library to manage the backoff timing and strategy. This is useful for transient errors where waiting before retrying might allow the source to succeed.

    // Backoff retry configuration
backOffCfg := backoff.NewExponentialBackOff()
backOffCfg.InitialInterval = 10 * time.Millisecond

observable := rxgo.Defer([]rxgo.Producer{func(ctx context.Context, next chan<- rxgo.Item, done func()) {
	next <- rxgo.Of(1)
	next <- rxgo.Of(2)
	next <- rxgo.Error(errors.New("foo"))
	done()
}}).BackOffRetry(backoff.WithMaxRetries(backOffCfg, 2))

  6. Use the IgnoreElements operator

    master
    The IgnoreElements operator prevents any items from being emitted by an Observable, but it still mirrors the termination notification (such as OnCompleted or OnError). This is useful when you only care about whether an Observable finished successfully or failed with an error, rather than the data it produced.

  7. Use the TakeWhile operator to filter items by condition

    master

    The TakeWhile operator mirrors items emitted by an Observable until a specified predicate function returns false. Once the condition is no longer met, the Observable stops emitting further items.

    observable := rxgo.Just(1, 2, 3, 4, 5)().TakeWhile(func(i interface{}) bool {
	return i != 3
})

  8. Use the Filter operator to emit items passing a predicate

    master

    The Filter operator allows you to emit only those items from an Observable that satisfy a specific predicate function. The predicate function receives an interface{} and must return true to allow the item to pass through or false to discard it.

    observable := rxgo.Just(1, 2, 3)().
	Filter(func(i interface{}) bool {
		return i != 2
	})

  9. Use Connectable Observables

    master

    A Connectable Observable does not start emitting items when subscribed to; it only begins emitting when the Connect() method is called. This allows multiple observers to subscribe before data starts flowing. Connectable Observables also publish items, meaning all observers receive a copy of the same items.

    To create one, use rxgo.WithPublishStrategy() with an operator like FromChannel.

    Connect() returns a disposed channel and a cancel function to manage the subscription lifecycle.

    ch := make(chan rxgo.Item)
go func() {
	ch <- rxgo.Of(1)
	ch <- rxgo.Of(2)
	ch <- rxgo.Of(3)
	close(ch)
}()

// Create a Connectable Observable
observable := rxgo.FromChannel(ch, rxgo.WithPublishStrategy())

observable.DoOnNext(func(i interface{}) {
	fmt.Printf("First observer: %d\n", i)
})

observable.DoOnNext(func(i interface{}) {
	fmt.Printf("Second observer: %d\n", i)
})

disposed, cancel := observable.Connect()
go func() {
	time.Sleep(time.Second)
	cancel()
}()

<-disposed

  10. Use the StartWith operator to prepend items to an Observable

    master
    The StartWith operator allows you to emit a specified set of items (from another Observable) before the source Observable begins emitting its own items. This is useful for initializing a stream with default values or header information.

  11. Use the FlatMap operator to flatten Observables

    master

    The FlatMap operator transforms each item emitted by an Observable into a new Observable, then flattens the emissions from all those resulting Observables into a single Observable. This is useful for handling asynchronous operations where each input item triggers a new stream of data.

    observable := rxgo.Just(1, 2, 3)().FlatMap(func(i rxgo.Item) rxgo.Observable {
	return rxgo.Just(i.V.(int) * 10, i.V.(int) * 100)()
})

  12. Use the Window operator to subdivide Observables

    master

    The Window operator periodically subdivides items from an Observable into smaller Observable windows. Instead of emitting individual items, it emits these windows as Observables.

    Available instances include:

    • WindowWithCount: Subdivides based on a specific number of items.
    • WindowWithTime: Subdivides based on a time interval.
    • WindowWithTimeOrCount: Subdivides based on either a time interval or a specific item count, whichever comes first.