This book contains the documentation of about 80 operators. The RxPY implementation contains about 130 operators. One of the most difficult parts of ReactiveX is using the most suitable operator for a given task. The following list will help you find which operator to use at that time:

To create an observable:

• That emits a single item: just
• That emits a single item returned from a function: Start
• That emits a sequence from an iterable or a generator: from_
• That emits a single item after a specified delay: timer
• That emits a single item from future: from_future
• That emits a sequence of items repeatedly: repeat

• From custom logic: create
• That emits a sequence of consecutive integers: range or from_ (in combination with the Python range function)
• That emits a sequence of items at particular intervals of time: interval
• That completes without emitting items: empty
• That does nothing and never completes: never
• That drops all items of an observable and mimics its completion: ignore_elements

To create an observable from other observables:

• That emits all of the items from the other observables in whatever order they are received: merge
• That emits all of the items from the other observables, sequentially, one observable at a time: concat
• That combines the items from several observables sequentially, emitting new items each time all of the observables emit an item: zip
• That emits a composed item whenever any of the observables emits an item: combine_latest
• That emits an item when one observable emits an item within a window defined by an item emitted on another observable: join
• That emits the items from only the most recently emitted of those observables: switch
• That emits items from the first observable that emits an item: amb

To emit items from an observable after a transformation:

• One item at a time with a function: map
• By emitting all of the items emitted by observables created from each received item: flat_map
• Based on all of the items that preceded them: scan and reduce
• By attaching a timestamp to them: timestamp
• By emitting the amount of time that elapsed before the emission of the item: time_interval
• By delaying the items emitted by an observable before re-emitting them: delay

To serialize and deserialize an observable:

• Wrapping them in notification objects: materialize
• Unwrapping them from notification objects: dematerialize

To prefix items:

• To an observable: start_with
• Only if its sequence is empty: default_if_empty

To group items emitted by an observable:

• By collecting items from an observable and re-emitting them as buffers of items: buffer
• By splitting one observable into multiple observables: window
• So that similar items are emitted on the same observable: group_by

To re-emit only some items from an observable:

• By dropping those that do not match some predicate: filter
• Only the first item: first
• Only the first items: take
• Only the last item: last
• Only one item at the specified index: element_at
• Only the items after the first items:
  • After the first n items: skip
  • Until one of those items matches a predicate: skip_while
  • After a second observable emits an item: skip_until
• Some items except the last item:
  • Except the last n items: skip_last
  • Until one of those items matches a predicate: take_while
  • Except items emitted after a second observable emits an item: take_until
• By sampling the observable periodically: sample
• By only emitting items that are not followed by other items within some duration: debounce
• By suppressing items that are duplicates of already-emitted items: distinct:
  • Only when two consecutive items are duplicates: distinct_until_changed

To evaluate the entire sequence of items emitted by an observable:

• And emit a single Boolean indicating whether all of the items pass some test: all
• And emit a single Boolean indicating whether the observable emitted any item that matches a predicate: contains
• And emit a single Boolean indicating whether the observable emitted no items: is_empty
• And emit a single Boolean indicating whether the sequence is identical to one emitted by a second observable: sequence_equal
• And emit the average of all of their values: average
• And emit the sum of all of their values: sum
• And emit a number indicating how many items were in the sequence: count
• And emit the item with the maximum value: max
• And emit the item with the minimum value: min
• By applying an aggregation function to each item in turn and emitting the result: scan and reduce

To convert the entire sequence of items emitted by an observable into a single item, being a list: to_list
To make an operator chain:

• Operate on a particular scheduler: subscribe_on
• And change it inside an operator chain: observe_on

To invoke a particular action when certain events occur: do_action
To notify observers:

• Of an error: throw
• If a specified period of time elapses without it emitting an item: timeout

To recover gracefully from errors:

• From a timeout by switching to a backup observable: timeout
• From an upstream error notification: catch
• By attempting to resubscribe to the upstream observable: retry

To create a resource that has the same lifespan as the observable: using

To transform a cold observable to a hot observable: share