Factored a pure interface out of Observable: IObservable.

The IObservable interface supports the core contract of observability: handling of each element, detection of completion, and detection of errors. The concrete Observable class is now focused on its rightful role: providing a set of highly usable fluent utilities for working with IObservables. For another example of this pattern, see e.g. Guava’s FluentIterable vs. the JDK’s Iterable. (If not for backward compatibility, Observable might have been called “FluentObservable” and IObservable simply “Observable”.)

Author: rickbw

rxjava-core/src/main/java/rx/IObservable.java

@@ -0,0 +1,67 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx;
+
+
+/**
+ * The Observable interface that implements the Reactive Pattern.
+ * <p>
+ * The documentation for this interface and its implementations makes use of
+ * marble diagrams. The following legend explains these diagrams:
+ * <p>
+ * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/legend.png">
+ * <p>
+ * For more information see the
+ * <a href="https://github.com/Netflix/RxJava/wiki/Observable">RxJava Wiki</a>
+ *
+ * @param <T> the type of the item emitted by the Observable.
+ */
+public interface IObservable<T> {
+
+    /**
+     * An {@link Observer} must call an Observable's {@code subscribe} method in
+     * order to receive items and notifications from the Observable.
+     * <p>
+     * A typical implementation of {@code subscribe} does the following:
+     * <ol>
+     * <li>It stores a reference to the Observer in a collection object, such as
+     *     a {@code List<T>} object.</li>
+     * <li>It returns a reference to the {@link Subscription} interface. This
+     *     enables Observers to unsubscribe, that is, to stop receiving items
+     *     and notifications before the Observable stops sending them, which
+     *     also invokes the Observer's {@link Observer#onCompleted onCompleted}
+     *     method.</li>
+     * </ol><p>
+     * An <code>IObservable&lt;T&gt;</code> instance is responsible for accepting
+     * all subscriptions and notifying all Observers. Unless the documentation
+     * for a particular <code>IObservable&lt;T&gt;</code> implementation
+     * indicates otherwise, Observers should make no assumptions about the order
+     * in which multiple Observers will receive their notifications.
+     * <p>
+     * For more information see the
+     * <a href="https://github.com/Netflix/RxJava/wiki/Observable">RxJava Wiki</a>
+     *
+     * @param observer the Observer
+     * @return a {@link Subscription} reference with which the {@link Observer}
+     *         can stop receiving items before the Observable has finished
+     *         sending them
+     * @throws IllegalArgumentException if the {@link Observer} provided as the
+     *                                  argument to {@code subscribe()} is
+     *                                  {@code null}.
+     */
+    public abstract Subscription subscribe(Observer<? super T> observer);
+
+}

rxjava-core/src/main/java/rx/Observable.java

@@ -130,10 +130,14 @@
 import rx.util.functions.Function;
 
 /**
- * The Observable interface that implements the Reactive Pattern.
+ * An implementation of the {@link IObservable} interface that provides
+ * overloaded methods for subscribing as well as delegate methods to the
+ * various operators in a fluent style.
  * <p>
- * This interface provides overloaded methods for subscribing as well as
- * delegate methods to the various operators.
+ * It is expected that most applications will lean heavily upon this
+ * particular IObservable implementation. To simplify the documentation,
+ * instances of the IObservable interface generally, whether or not of this
+ * class in particular, will be referred to simply as "Observables".
  * <p>
  * The documentation for this interface makes use of marble diagrams. The
  * following legend explains these diagrams:
@@ -145,7 +149,7 @@
  * 
  * @param <T> the type of the item emitted by the Observable
  */
-public class Observable<T> {
+public class Observable<T> implements IObservable<T> {
 
     private final static ConcurrentHashMap<Class, Boolean> internalClassMap = new ConcurrentHashMap<Class, Boolean>();
 
@@ -213,6 +217,7 @@ protected Observable(OnSubscribeFunc<T> onSubscribe) {
      *                                  argument to {@code subscribe()} is
      *                                  {@code null}
      */
+    @Override
     public Subscription subscribe(Observer<? super T> observer) {
         // allow the hook to intercept and/or decorate
         OnSubscribeFunc<T> onSubscribeFunction = hook.onSubscribeStart(this, onSubscribe);
@@ -2201,6 +2206,30 @@ public Observable<Timestamped<T>> timestamp() {
         return create(OperationTimestamp.timestamp(this));
     }
 
+    /**
+     * If the given {@link IObservable} is an {@link Observable} already,
+     * simply return it, cast to its concrete type. If not, wrap it in a new
+     * Observable that will delegate its own {@link #subscribe(Observer)} to
+     * the given Observable.
+     *
+     * @return  the given Observable if it is of the correct type, or a new
+     *          one that delegates to it if not.
+     */
+    public static <T> Observable<T> from(final IObservable<T> observable) {
+        if (null == observable) {
+            throw new NullPointerException("IObservable argument");
+        } else if (observable instanceof Observable<?>) {
+            return (Observable<T>) observable;
+        } else {
+            return create(new OnSubscribeFunc<T>() {
+                @Override
+                public Subscription onSubscribe(Observer<? super T> observer) {
+                    return observable.subscribe(observer);
+                }
+            });
+        }
+    }
+
     /**
      * Converts a {@link Future} into an Observable.
      * <p>