1.x: deprecate create(), add alternatives

Author: akarnokd

src/main/java/rx/Completable.java

@@ -2249,7 +2249,7 @@ public final <R> R to(Func1<? super Completable, R> converter) {
      * @return the new Observable created
      */
     public final <T> Observable<T> toObservable() {
-        return Observable.create(new Observable.OnSubscribe<T>() {
+        return Observable.unsafeCreate(new Observable.OnSubscribe<T>() {
             @Override
             public void call(Subscriber<? super T> s) {
                 unsafeSubscribe(s);

src/main/java/rx/Observable.java

@@ -202,7 +202,7 @@ public interface Transformer<T, R> extends Func1<Single<T>, Single<R>> {
      */
     private static <T> Observable<T> asObservable(Single<T> t) {
         // is this sufficient, or do I need to keep the outer Single and subscribe to it?
-        return Observable.create(new SingleToObservable<T>(t.onSubscribe));
+        return Observable.unsafeCreate(new SingleToObservable<T>(t.onSubscribe));
     }
 
     /* *********************************************************************************************************

src/main/java/rx/Single.java

@@ -28,7 +28,7 @@ public enum EmptyObservableHolder implements OnSubscribe<Object> {
     ;
 
     /** The singleton instance. */
-    static final Observable<Object> EMPTY = Observable.create(INSTANCE);
+    static final Observable<Object> EMPTY = Observable.unsafeCreate(INSTANCE);
 
 
     /**

src/main/java/rx/internal/operators/EmptyObservableHolder.java

@@ -28,7 +28,7 @@ public enum NeverObservableHolder implements OnSubscribe<Object> {
     ;
 
     /** The singleton instance. */
-    static final Observable<Object> NEVER = Observable.create(INSTANCE);
+    static final Observable<Object> NEVER = Observable.unsafeCreate(INSTANCE);
 
     /**
      * Returns a type-corrected singleton instance of the never Observable.

src/main/java/rx/internal/operators/NeverObservableHolder.java

@@ -29,13 +29,13 @@
 import rx.plugins.RxJavaHooks;
 import rx.subscriptions.SerialSubscription;
 
-public final class OnSubscribeFromEmitter<T> implements OnSubscribe<T> {
+public final class OnSubscribeCreate<T> implements OnSubscribe<T> {
 
     final Action1<Emitter<T>> Emitter;
 
     final Emitter.BackpressureMode backpressure;
 
-    public OnSubscribeFromEmitter(Action1<Emitter<T>> Emitter, Emitter.BackpressureMode backpressure) {
+    public OnSubscribeCreate(Action1<Emitter<T>> Emitter, Emitter.BackpressureMode backpressure) {
         this.Emitter = Emitter;
         this.backpressure = backpressure;
     }
@@ -268,7 +268,7 @@ public void onError(Throwable e) {
 
         @Override
         void onOverflow() {
-            onError(new MissingBackpressureException("fromEmitter: could not emit value due to lack of requests"));
+            onError(new MissingBackpressureException("create: could not emit value due to lack of requests"));
         }
 
     }

…al/operators/OnSubscribeFromEmitter.java …nternal/operators/OnSubscribeCreate.javasrc/main/java/rx/internal/operators/OnSubscribeFromEmitter.java renamed to src/main/java/rx/internal/operators/OnSubscribeCreate.java src/main/java/rx/internal/operators/OnSubscribeFromEmitter.java renamed to src/main/java/rx/internal/operators/OnSubscribeCreate.java

@@ -70,9 +70,9 @@ public static <T, R> Observable<R> createFrom(Observable<? extends T> source,
             Func1<? super T, ? extends Iterable<? extends R>> mapper, int prefetch) {
         if (source instanceof ScalarSynchronousObservable) {
             T scalar = ((ScalarSynchronousObservable<? extends T>) source).get();
-            return Observable.create(new OnSubscribeScalarFlattenIterable<T, R>(scalar, mapper));
+            return Observable.unsafeCreate(new OnSubscribeScalarFlattenIterable<T, R>(scalar, mapper));
         }
-        return Observable.create(new OnSubscribeFlattenIterable<T, R>(source, mapper, prefetch));
+        return Observable.unsafeCreate(new OnSubscribeFlattenIterable<T, R>(source, mapper, prefetch));
     }
 
     static final class FlattenIterableSubscriber<T, R> extends Subscriber<T> {

src/main/java/rx/internal/operators/OnSubscribeFlattenIterable.java

@@ -170,7 +170,7 @@ public void onNext(T1 args) {
                         leftMap().put(id, subjSerial);
                     }
 
-                    Observable<T2> window = Observable.create(new WindowObservableFunc<T2>(subj, cancel));
+                    Observable<T2> window = Observable.unsafeCreate(new WindowObservableFunc<T2>(subj, cancel));
 
                     Observable<D1> duration = leftDuration.call(args);
 

src/main/java/rx/internal/operators/OnSubscribeGroupJoin.java

@@ -31,7 +31,7 @@
  * limitations under the License.
  */
 
-import static rx.Observable.create; // NOPMD
+import static rx.Observable.unsafeCreate; // NOPMD
 
 import java.util.concurrent.atomic.*;
 
@@ -133,11 +133,11 @@ public static <T> Observable<T> retry(Observable<T> source, final long count) {
     }
 
     public static <T> Observable<T> retry(Observable<T> source, Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> notificationHandler) {
-        return create(new OnSubscribeRedo<T>(source, notificationHandler, true, false, Schedulers.trampoline()));
+        return unsafeCreate(new OnSubscribeRedo<T>(source, notificationHandler, true, false, Schedulers.trampoline()));
     }
 
     public static <T> Observable<T> retry(Observable<T> source, Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> notificationHandler, Scheduler scheduler) {
-        return create(new OnSubscribeRedo<T>(source, notificationHandler, true, false, scheduler));
+        return unsafeCreate(new OnSubscribeRedo<T>(source, notificationHandler, true, false, scheduler));
     }
 
     public static <T> Observable<T> repeat(Observable<T> source) {
@@ -163,15 +163,15 @@ public static <T> Observable<T> repeat(Observable<T> source, final long count, S
     }
 
     public static <T> Observable<T> repeat(Observable<T> source, Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> notificationHandler) {
-        return create(new OnSubscribeRedo<T>(source, notificationHandler, false, true, Schedulers.trampoline()));
+        return unsafeCreate(new OnSubscribeRedo<T>(source, notificationHandler, false, true, Schedulers.trampoline()));
     }
 
     public static <T> Observable<T> repeat(Observable<T> source, Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> notificationHandler, Scheduler scheduler) {
-        return create(new OnSubscribeRedo<T>(source, notificationHandler, false, true, scheduler));
+        return unsafeCreate(new OnSubscribeRedo<T>(source, notificationHandler, false, true, scheduler));
     }
 
     public static <T> Observable<T> redo(Observable<T> source, Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> notificationHandler, Scheduler scheduler) {
-        return create(new OnSubscribeRedo<T>(source, notificationHandler, false, false, scheduler));
+        return unsafeCreate(new OnSubscribeRedo<T>(source, notificationHandler, false, false, scheduler));
     }
 
     private OnSubscribeRedo(Observable<T> source, Func1<? super Observable<? extends Notification<?>>, ? extends Observable<?>> f, boolean stopOnComplete, boolean stopOnError,

src/main/java/rx/internal/operators/OnSubscribeRedo.java

@@ -123,7 +123,7 @@ public static <T, R> Observable<R> create(final Observable<? extends T> source,
 
     public static <T, R> Observable<R> create(final Observable<? extends T> source,
             final Func1<? super Observable<T>, ? extends Observable<R>> selector, final boolean delayError) {
-        return create(new OnSubscribe<R>() {
+        return unsafeCreate(new OnSubscribe<R>() {
             @Override
             public void call(final Subscriber<? super R> child) {
                 final OnSubscribePublishMulticast<T> op = new OnSubscribePublishMulticast<T>(RxRingBuffer.SIZE, delayError);
@@ -155,7 +155,7 @@ public void setProducer(Producer p) {
                 child.add(op);
                 child.add(subscriber);
 
-                selector.call(Observable.create(op)).unsafeSubscribe(subscriber);
+                selector.call(Observable.unsafeCreate(op)).unsafeSubscribe(subscriber);
 
                 source.unsafeSubscribe(op.subscriber());
             }