Subscribers for common factory methods

Similar to Observers.

Author: benjchristensen

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

@@ -47,27 +47,6 @@ protected Subscriber(Subscriber<?> op) {
         this(op.cs);
     }
 
-    public static <T> Subscriber<T> from(final Observer<? super T> o) {
-        return new Subscriber<T>() {
-
-            @Override
-            public void onCompleted() {
-                o.onCompleted();
-            }
-
-            @Override
-            public void onError(Throwable e) {
-                o.onError(e);
-            }
-
-            @Override
-            public void onNext(T t) {
-                o.onNext(t);
-            }
-
-        };
-    }
-
     /**
      * Used to register an unsubscribe callback.
      */

rxjava-core/src/main/java/rx/observers/Subscribers.java

@@ -0,0 +1,151 @@
+package rx.observers;
+
+import rx.Observer;
+import rx.Subscriber;
+import rx.util.OnErrorNotImplementedException;
+import rx.util.functions.Action0;
+import rx.util.functions.Action1;
+
+public class Subscribers {
+
+    public static <T> Subscriber<T> from(final Observer<? super T> o) {
+        return new Subscriber<T>() {
+
+            @Override
+            public void onCompleted() {
+                o.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                o.onError(e);
+            }
+
+            @Override
+            public void onNext(T t) {
+                o.onNext(t);
+            }
+
+        };
+    }
+
+    /**
+     * Create an empty Subscriber that ignores all events.
+     */
+    public static final <T> Subscriber<T> create() {
+        return new Subscriber<T>() {
+
+            @Override
+            public final void onCompleted() {
+                // do nothing
+            }
+
+            @Override
+            public final void onError(Throwable e) {
+                throw new OnErrorNotImplementedException(e);
+            }
+
+            @Override
+            public final void onNext(T args) {
+                // do nothing
+            }
+
+        };
+    }
+
+    /**
+     * Create an Subscriber that receives `onNext` and ignores `onError` and `onCompleted`.
+     */
+    public static final <T> Subscriber<T> create(final Action1<? super T> onNext) {
+        if (onNext == null) {
+            throw new IllegalArgumentException("onNext can not be null");
+        }
+
+        return new Subscriber<T>() {
+
+            @Override
+            public final void onCompleted() {
+                // do nothing
+            }
+
+            @Override
+            public final void onError(Throwable e) {
+                throw new OnErrorNotImplementedException(e);
+            }
+
+            @Override
+            public final void onNext(T args) {
+                onNext.call(args);
+            }
+
+        };
+    }
+
+    /**
+     * Create an Subscriber that receives `onNext` and `onError` and ignores `onCompleted`.
+     * 
+     */
+    public static final <T> Subscriber<T> create(final Action1<? super T> onNext, final Action1<Throwable> onError) {
+        if (onNext == null) {
+            throw new IllegalArgumentException("onNext can not be null");
+        }
+        if (onError == null) {
+            throw new IllegalArgumentException("onError can not be null");
+        }
+
+        return new Subscriber<T>() {
+
+            @Override
+            public final void onCompleted() {
+                // do nothing
+            }
+
+            @Override
+            public final void onError(Throwable e) {
+                onError.call(e);
+            }
+
+            @Override
+            public final void onNext(T args) {
+                onNext.call(args);
+            }
+
+        };
+    }
+
+    /**
+     * Create an Subscriber that receives `onNext`, `onError` and `onCompleted`.
+     * 
+     */
+    public static final <T> Subscriber<T> create(final Action1<? super T> onNext, final Action1<Throwable> onError, final Action0 onComplete) {
+        if (onNext == null) {
+            throw new IllegalArgumentException("onNext can not be null");
+        }
+        if (onError == null) {
+            throw new IllegalArgumentException("onError can not be null");
+        }
+        if (onComplete == null) {
+            throw new IllegalArgumentException("onComplete can not be null");
+        }
+
+        return new Subscriber<T>() {
+
+            @Override
+            public final void onCompleted() {
+                onComplete.call();
+            }
+
+            @Override
+            public final void onError(Throwable e) {
+                onError.call(e);
+            }
+
+            @Override
+            public final void onNext(T args) {
+                onNext.call(args);
+            }
+
+        };
+    }
+
+}

rxjava-core/src/main/java/rx/observers/TestSubscriber.java

@@ -26,7 +26,7 @@
  */
 public class TestSubscriber<T> extends Subscriber<T> {
 
-    private final Subscriber<Object> EMPTY = Subscriber.from(new EmptyObserver<Object>());
+    private final Subscriber<Object> EMPTY = Subscribers.create();
 
     private final TestObserver<T> testObserver;