Merged master into buffer-operation branch

Author: michaeldejong

CHANGES.md

@@ -1,5 +1,16 @@
 # RxJava Releases #
 
+### Version 0.9.1 ([Maven Central](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22com.netflix.rxjava%22%20AND%20v%3A%220.9.1%22)) ###
+
+* [Pull 303](https://github.com/Netflix/RxJava/pull/303) CombineLatest
+* [Pull 290](https://github.com/Netflix/RxJava/pull/290) Zip overload with FuncN
+* [Pull 302](https://github.com/Netflix/RxJava/pull/302) NPE fix when no package on class
+* [Pull 284](https://github.com/Netflix/RxJava/pull/284) GroupBy fixes (items still [oustanding](https://github.com/Netflix/RxJava/issues/282))
+* [Pull 288](https://github.com/Netflix/RxJava/pull/288) PublishSubject concurrent modification fixes
+* [Issue 198](https://github.com/Netflix/RxJava/issues/198) Throw if no onError handler specified
+* [Issue 278](https://github.com/Netflix/RxJava/issues/278) Subscribe argument validation
+* Javadoc improvements and many new marble diagrams
+
 ### Version 0.9.0 ([Maven Central](http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22com.netflix.rxjava%22%20AND%20v%3A%220.9.0%22)) ###
 
 This release includes breaking changes that move all blocking operators (such as `single`, `last`, `forEach`) to `BlockingObservable`.

gradle.properties

@@ -1 +1 @@
-version=0.9.1-SNAPSHOT
+version=0.9.2-SNAPSHOT

language-adaptors/rxjava-groovy/src/test/groovy/rx/lang/groovy/ObservableTests.groovy

@@ -19,17 +19,21 @@ import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
 
 import org.junit.Before;
 import org.junit.Test;
 import static org.junit.Assert.*;
+
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 
 import rx.Notification;
 import rx.Observable;
 import rx.Observer;
 import rx.Subscription;
+import rx.observables.GroupedObservable;
 import rx.subscriptions.Subscriptions;
 import rx.util.functions.Func1;
 
@@ -280,6 +284,29 @@ def class ObservableTests {
         Observable.from(1, 2, 3).all({ x -> x > 0 }).subscribe({ result -> a.received(result) });
         verify(a, times(1)).received(true);
     }
+    
+    @Test
+    public void testGroupBy() {
+        int count=0;
+        
+        Observable.from("one", "two", "three", "four", "five", "six")
+        .groupBy({String s -> s.length()})
+        .mapMany({
+            groupObservable ->
+            
+            return groupObservable.map({
+                s ->
+                return "Value: " + s + " Group: " + groupObservable.getKey(); 
+            });
+          }).toBlockingObservable().forEach({
+            s ->
+            println(s);
+            count++;
+          })
+          
+          assertEquals(6, count);
+    }
+    
 
     def class AsyncObservable implements Func1<Observer<Integer>, Subscription> {
 

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

@@ -20,12 +20,31 @@
 import rx.Subscription;
 import rx.util.functions.Func1;
 
+/**
+ * A ConnectableObservable resembles an ordinary {@link Observable}, except that it does not begin
+ * emitting items when it is subscribed to, but only when its {@link #connect} method is called. In
+ * this way you can wait for all intended {@link Observer}s to {@link Observable#subscribe} to the
+ * Observable before the Observable begins emitting items.
+ * <p>
+ * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/publishConnect.png">
+ * <p>
+ * For more information see
+ * <a href="https://github.com/Netflix/RxJava/wiki/Connectable-Observable-Operators">Connectable
+ * Observable Operators</a> at the RxJava Wiki
+ *
+ * @param <T>
+ */
+
 public abstract class ConnectableObservable<T> extends Observable<T> {
 
     protected ConnectableObservable(Func1<Observer<T>, Subscription> onSubscribe) {
         super(onSubscribe);
     }
 
+    /**
+     * Call a ConnectableObservable's connect() method to instruct it to begin emitting the
+     * items from its underlying {@link Observable} to its {@link Observer}s. 
+     */
     public abstract Subscription connect();
 
 }

rxjava-core/src/main/java/rx/observables/BlockingObservable.java

@@ -21,7 +21,8 @@
 import rx.util.functions.Func1;
 
 /**
- * An {@link Observable} that has been grouped by a key whose value can be obtained using {@link #getKey()} <p>
+ * An {@link Observable} that has been grouped by a key whose value can be obtained using
+ * {@link #getKey()} <p>
  * 
  * @see Observable#groupBy(Observable, Func1)
  * 

rxjava-core/src/main/java/rx/observables/ConnectableObservable.java

@@ -6,6 +6,7 @@
 import rx.Observer;
 import rx.plugins.RxJavaPlugins;
 import rx.util.CompositeException;
+import rx.util.OnErrorNotImplementedException;
 
 /**
  * Wrapper around Observer to ensure compliance with Rx contract.
@@ -70,13 +71,28 @@ public void onError(Exception e) {
             try {
                 actual.onError(e);
             } catch (Exception e2) {
-                // if the onError itself fails then pass to the plugin
-                // see https://github.com/Netflix/RxJava/issues/216 for further discussion
-                RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
-                RxJavaPlugins.getInstance().getErrorHandler().handleError(e2);
-                // and throw exception despite that not being proper for Rx
-                // https://github.com/Netflix/RxJava/issues/198
-                throw new RuntimeException("Error occurred when trying to propagate error to Observer.onError", new CompositeException(Arrays.asList(e, e2)));
+                if (e2 instanceof OnErrorNotImplementedException) {
+                    /**
+                     * onError isn't implemented so throw
+                     * 
+                     * https://github.com/Netflix/RxJava/issues/198
+                     * 
+                     * Rx Design Guidelines 5.2
+                     * 
+                     * "when calling the Subscribe method that only has an onNext argument, the OnError behavior will be
+                     * to rethrow the exception on the thread that the message comes out from the observable sequence.
+                     * The OnCompleted behavior in this case is to do nothing."
+                     */
+                    throw (OnErrorNotImplementedException) e2;
+                } else {
+                    // if the onError itself fails then pass to the plugin
+                    // see https://github.com/Netflix/RxJava/issues/216 for further discussion
+                    RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
+                    RxJavaPlugins.getInstance().getErrorHandler().handleError(e2);
+                    // and throw exception despite that not being proper for Rx
+                    // https://github.com/Netflix/RxJava/issues/198
+                    throw new RuntimeException("Error occurred when trying to propagate error to Observer.onError", new CompositeException(Arrays.asList(e, e2)));
+                }
             }
             // auto-unsubscribe
             subscription.unsubscribe();

rxjava-core/src/main/java/rx/observables/GroupedObservable.java

@@ -12,6 +12,12 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 
+/**
+ * Returns an Observable that emits a Boolean that indicates whether all items emitted by an
+ * Observable satisfy a condition.
+ * <p>
+ * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/all.png">
+ */
 public class OperationAll {
 
     public static <T> Func1<Observer<Boolean>, Subscription> all(Observable<T> sequence, Func1<T, Boolean> predicate) {

rxjava-core/src/main/java/rx/operators/AtomicObserver.java

@@ -398,9 +398,9 @@ public void onNext(T args) {
      * @param <T> The type of object all internal {@link Buffer} objects record.
      */
     private interface BufferCreator<T> {
-    	/**
-    	 * Signifies a onNext event.
-    	 */
+        /**
+         * Signifies a onNext event.
+         */
         void onValuePushed();
 
         /**

rxjava-core/src/main/java/rx/operators/OperationAll.java

@@ -33,15 +33,17 @@
 import rx.util.functions.Func1;
 
 /**
- * Similar to {@link Observable#replay()} except that this auto-subscribes to the source sequence.
+ * This method has similar behavior to {@link Observable#replay()} except that this auto-subscribes
+ * to the source Observable rather than returning a connectable Observable.
  * <p>
- * This is useful when returning an Observable that you wish to cache responses but can't control the
+ * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/cache.png">
+ * <p>
+ * This is useful with an Observable that you want to cache responses when you can't control the
  * subscribe/unsubscribe behavior of all the Observers.
  * <p>
- * NOTE: You sacrifice the ability to unsubscribe from the origin with this operator so be careful to not
- * use this on infinite or very large sequences that will use up memory. This is similar to
- * the {@link Observable#toList()} operator in this caution.
- * 
+ * NOTE: You sacrifice the ability to unsubscribe from the origin when you use this operator, so be
+ * careful not to use this operator on Observables that emit infinite or very large numbers of
+ * items, as this will use up memory.
  */
 public class OperationCache {