Creating Observables

This section explains methods that create Observables.

• from( ) — convert an Iterable, a Future, or an Array into an Observable
• fromFuture( ) — convert a Future into an Observable, but do not attempt to get the Future's value until a Subscriber subscribes
• forIterable( ) — apply a function to the elements of an Iterable to create Observables which are then concatenated
• just( ) — convert an object into an Observable that emits that object
• repeat( ) — create an Observable that emits a particular item or sequence of items repeatedly
• create( ) — create an Observable from scratch by means of a function
• defer( ) — do not create the Observable until a Subscriber subscribes; create a fresh Observable on each subscription
• range( ) — create an Observable that emits a range of sequential integers
• interval( ) — create an Observable that emits a sequence of integers spaced by a given time interval
• timer( ) — create an Observable that emits a single item after a given delay
• generate( ) and generateAbsoluteTime( ) — create an Observable that emits a sequence of items as generated by a function of your choosing
• empty( ) — create an Observable that emits nothing and then completes
• error( ) — create an Observable that emits nothing and then signals an error
• never( ) — create an Observable that emits nothing at all

---

from( )

convert an Iterable, a Future, or an Array into an Observable

You can convert an object that supports Iterable into an Observable that emits each iterable item in the object, or an object that supports Future into an Observable that emits the result of the get call, simply by passing the object into the from( ) methods, for example:

myObservable = Observable.from(myIterable);

You can also do this with arrays, for example:

myArray = [1, 2, 3, 4, 5];
myArrayObservable = Observable.from(myArray);

This converts the sequence of values in the iterable object or array into a sequence of items emitted, one at a time, by an Observable.

An empty iterable (or array) can be converted to an Observable in this way. The resulting Observable will invoke onCompleted() without first invoking onNext().

Note that when the from( ) method transforms a Future into an Observable, such an Observable will be effectively blocking, as its underlying Future blocks.

see also:

• javadoc: from(future)
• javadoc: from(future, timeout, unit)
• javadoc: from(future, scheduler)
• javadoc: from(iterable)
• javadoc: from(array)
• RxJS: fromArray
• RxJS: fromPromise
• Linq: ToObservable

---

fromFuture( )

convert a Future into an Observable, but do not attempt to get the Future's value until a Subscriber subscribes

The fromFuture( ) method also converts a Future into an Observable, but it obtains this Future indirectly, by means of a function you provide. It creates the Observable immediately, but waits to call the function and to obtain the Future until a Subscriber subscribes to it.

---

forIterable( )

apply a function to the elements of an Iterable to create Observables which are then concatenated

forIterable( ) is similar to from(Iterable ) but instead of the resulting Observable emitting the elements of the Iterable as its own emitted items, it applies a specified function to each of these elements to generate one Observable per element, and then concatenates the emissions of these Observables to be its own sequence of emitted items.

---

just( )

convert an object into an Observable that emits that object

To convert any object into an Observable that emits that object, pass that object into the just( ) method.

// Observable emits "some string" as a single item
def observableThatEmitsAString = Observable.just("some string"); 
// Observable emits the list [1, 2, 3, 4, 5] as a single item
def observableThatEmitsAList = Observable.just([1, 2, 3, 4, 5]); 

This has some similarities to the from( ) method, but note that if you pass an iterable to from( ), it will convert an iterable object into an Observable that emits each of the items in the iterable, one at a time, while the just( ) method would convert the iterable into an Observable that emits the entire iterable as a single item.

Note that if you pass null to just( ), the resulting Observable will not merely call onCompleted( ) without calling onNext( ). It will instead call onNext( null ) before calling onCompleted( ).

see also:

• javadoc: just(value)

---

repeat( )

create an Observable that emits the sequence of items emitted by the source Observable repeatedly

There are also versions of repeat( ) that operate on a scheduler that you specify, and that repeat only a certain number of times before terminating.

see also:

• javadoc: repeat( ) and repeat(scheduler)
• Linq: Repeat
• RxJS: repeat

---

create( )

create an Observable from scratch by means of a function

You can create an Observable from scratch by using the create( ) method. You pass this method a function that accepts as its parameter the Subscriber that is passed to an Observable’s subscribe( ) method (or that is derived from the Observerer that is passed to that method). Write the function you pass to create( ) so that it behaves as an Observable — calling the passed-in Subscriber’s onNext( ), onError( ), and onCompleted( ) methods appropriately. For example:

def myObservable = Observable.create({ aSubscriber ->
  try {
    for (int i = 1; i < 1000000; i++) {
      if (true == aSubscriber.isUnsubscribed()) {
        return;
      }
      aSubscriber.onNext(i);
    }
    if (false == aSubscriber.isUnsubscribed()) {
      aSubscriber.onCompleted();
    }
  } catch(Throwable t) {
    if (false == aSubscriber.isUnsubscribed()) {
      aSubscriber.onError(t);
    }
  }
})

NOTE: A well-formed finite Observable must attempt to call either the Subscriber’s onCompleted( ) method exactly once or its onError( ) method exactly once, and must not thereafter attempt to call any of the Subscriber’s other methods. It is good practice to check the Subscriber’s isUnsubscribed( ) state so that your Observable can stop emitting items or doing expensive calculations when there is no longer an interested Subscriber.

see also:

• javadoc: create(OnSubscribe)
• RxJS: create
• Linq: Create

---

defer( )

do not create the Observable until a Subscriber subscribes; create a fresh Observable on each subscription

Pass defer( ) an Observable factory function (a function that generates Observables), and defer( ) will return an Observable that will call this function to generate its Observable sequence afresh each time a new Subscriber subscribes.

see also:

• javadoc: defer(observableFactory)
• RxJS: defer
• Linq: Defer

---

range( )

create an Observable that emits a range of sequential integers

To create an Observable that emits a range of sequential integers, pass the starting integer and the number of integers to emit to the range( ) method.

// myObservable emits the integers 5, 6, and 7 before completing:
def myObservable = Observable.range(5, 3);

In calls to range(n,m), a value of zero for m will result in no numbers being emitted (values less than zero will cause an exception). n may be any integer that can be represented as a BigDecimal — posititve, negative, or zero.

see also:

• javadoc: range(start, count)
• RxJS: range
• Linq: Range
• Introduction to Rx: Range

---

interval( )

create an Observable that emits a sequence of integers spaced by a given time interval

To create an Observable that emits items spaced by a particular interval of time, pass the time interval and the units of time that interval is measured in (and, optionally, a scheduler) to the interval( ) method.

see also:

• javadoc: interval(interval,unit)
• javadoc: interval(interval,unit,scheduler)
• RxJS: interval
• Linq: Interval
• Introduction to Rx: Interval

---

timer( )

create an Observable that emits a single item after a given delay

The timer( ) method returns an Observable that, when subscribed to, waits for a span of time that you have defined, then emits a single zero and completes.

There is also a version of timer( ) that emits a single zero after a specified delay, and then emits incrementally increasing numbers periodically thereafter on a specified periodicity:

For both of these versions of timer( ) you can optionally specify a Scheduler on which the timing will take place.

see also:

• javadoc: timer( )
• RxJS: timer
• Linq: Timer
• Introduction to Rx: Timer

---

generate( ) and generateAbsoluteTime( )

create an Observable that emits a sequence of items as generated by a function of your choosing

The basic form of generate( ) takes four parameters. These are initialState and three functions: iterate( ), condition( ), and resultSelector( ). generate( ) uses these four parameters to generate an Observable sequence, which is its return value. It does so in the following way.

generate( ) creates each emission from the sequence by applying the resultSelector( ) function to the current state and emitting the resulting item. The first state, which determines the first emitted item, is initialState. generate( ) determines each subsequent state by applying iterate( ) to the current state. Before emitting an item, generate( ) tests the result of condition( ) applied to the current state. If the result of this test is false, instead of calling resultSelector( ) and emitting the resulting value, generate( ) terminates the sequence and stops iterating the state.

There are also versions of generate( ) that allow you to do the work of generating the sequence on a particular Scheduler and that allow you to set the time interval between emissions by applying a function to the current state. The generateAbsoluteTime( ) allows you to control the time at which an item is emitted by applying a function to the state to get an absolute system clock time (rather than an interval from the previous emission).

see also:

• Introduction to Rx: Generate
• Linq: Generate
• RxJS: generate, generateWithAbsoluteTime, and generateWithRelativeTime

---

empty( ), error( ), and never( )

Observables that can be useful for testing purposes

• empty( ) creates an Observable that does not emit any items but instead immediately calls the Subscriber’s onCompleted( ) method.
• error( ) creates an Observable that does not emit any items but instead immediately calls the Subscriber’s onError( ) method.
• never( ) creates an Observable that does not emit any items, nor does it call either the Subscriber’s onCompleted( ) or onError( ) methods.

import rx.Observable;
import rx.Observer;
import rx.Subscription;
import rx.subscriptions.Subscriptions;
import rx.util.functions.Func1;

println("*** empty() ***");
Observable.empty().subscribe(
  { println("empty: " + it); },             // onNext
  { println("empty: error - " + it.getMessage()); }, // onError
  { println("empty: Sequence complete"); }  // onCompleted
);

println("*** error() ***");
Observable.error(new Throwable("badness")).subscribe(
  { println("error: " + it); },             // onNext
  { println("error: error - " + it.getMessage()); }, // onError
  { println("error: Sequence complete"); }  // onCompleted
);

println("*** never() ***");
Observable.never().subscribe(
  { println("never: " + it); },             // onNext
  { println("never: error - " + it.getMessage()); }, // onError
  { println("never: Sequence complete"); }  // onCompleted
);
println("*** END ***");

*** empty() ***
empty: Sequence complete
*** error() ***
error: error - badness
*** never() ***
*** END ***

see also:

• javadoc: empty()
• javadoc: error(exception)
• javadoc: never()
• RxJS: empty and never
• Linq: Empty and Never
• Introduction to Rx: Simple factory methods