Upgrading SyncOnSubscribe and AsyncOnSubscribe from experimental to beta

Author: Aaron Tull

src/main/java/rx/Observable.java

@@ -126,7 +126,7 @@ public static <T> Observable<T> create(OnSubscribe<T> f) {
      * @see <a href="http://reactivex.io/documentation/operators/create.html">ReactiveX operators documentation: Create</a>
      * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
      */
-    @Experimental
+    @Beta
     public static <S, T> Observable<T> create(SyncOnSubscribe<S, T> syncOnSubscribe) {
         return new Observable<T>(hook.onCreate(syncOnSubscribe));
     }
@@ -161,7 +161,7 @@ public static <S, T> Observable<T> create(SyncOnSubscribe<S, T> syncOnSubscribe)
      * @see <a href="http://reactivex.io/documentation/operators/create.html">ReactiveX operators documentation: Create</a>
      * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
      */
-    @Experimental
+    @Beta
     public static <S, T> Observable<T> create(AsyncOnSubscribe<S, T> asyncOnSubscribe) {
         return new Observable<T>(hook.onCreate(asyncOnSubscribe));
     }

src/main/java/rx/observables/AsyncOnSubscribe.java

@@ -18,17 +18,17 @@
 
 import java.util.*;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-
+ 
 import rx.*;
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.Observer;
-import rx.annotations.Experimental;
 import rx.functions.*;
 import rx.internal.operators.*;
 import rx.observers.*;
 import rx.plugins.RxJavaPlugins;
 import rx.subscriptions.CompositeSubscription;
+import rx.annotations.Beta;
 
 /**
  * A utility class to create {@code OnSubscribe<T>} functions that respond correctly to back
@@ -44,7 +44,7 @@
  * @param <T>
  *            the type of {@code Subscribers} that will be compatible with {@code this}.
  */
-@Experimental
+@Beta
 public abstract class AsyncOnSubscribe<S, T> implements OnSubscribe<T> {
 
     /**
@@ -108,7 +108,7 @@ protected void onUnsubscribe(S state) {
      *            {@link #next(Object, long, Observer) next(S, long, Observer)})
      * @return an AsyncOnSubscribe that emits data in a protocol compatible with back-pressure.
      */
-    @Experimental
+    @Beta
     public static <S, T> AsyncOnSubscribe<S, T> createSingleState(Func0<? extends S> generator, 
             final Action3<? super S, Long, ? super Observer<Observable<? extends T>>> next) {
         Func3<S, Long, ? super Observer<Observable<? extends T>>, S> nextFunc = 
@@ -137,7 +137,7 @@ public S call(S state, Long requested, Observer<Observable<? extends T>> subscri
      * @return an AsyncOnSubscribe that emits data downstream in a protocol compatible with
      *         back-pressure.
      */
-    @Experimental
+    @Beta
     public static <S, T> AsyncOnSubscribe<S, T> createSingleState(Func0<? extends S> generator, 
             final Action3<? super S, Long, ? super Observer<Observable<? extends T>>> next, 
             final Action1<? super S> onUnsubscribe) {
@@ -165,7 +165,7 @@ public S call(S state, Long requested, Observer<Observable<? extends T>> subscri
      * @return an AsyncOnSubscribe that emits data downstream in a protocol compatible with
      *         back-pressure.
      */
-    @Experimental
+    @Beta
     public static <S, T> AsyncOnSubscribe<S, T> createStateful(Func0<? extends S> generator, 
             Func3<? super S, Long, ? super Observer<Observable<? extends T>>, ? extends S> next, 
             Action1<? super S> onUnsubscribe) {
@@ -184,7 +184,7 @@ public static <S, T> AsyncOnSubscribe<S, T> createStateful(Func0<? extends S> ge
      * @return an AsyncOnSubscribe that emits data downstream in a protocol compatible with
      *         back-pressure.
      */
-    @Experimental
+    @Beta
     public static <S, T> AsyncOnSubscribe<S, T> createStateful(Func0<? extends S> generator, 
             Func3<? super S, Long, ? super Observer<Observable<? extends T>>, ? extends S> next) {
         return new AsyncOnSubscribeImpl<S, T>(generator, next);
@@ -203,7 +203,7 @@ public static <S, T> AsyncOnSubscribe<S, T> createStateful(Func0<? extends S> ge
      * @return an AsyncOnSubscribe that emits data downstream in a protocol compatible with
      *         back-pressure.
      */
-    @Experimental
+    @Beta
     public static <T> AsyncOnSubscribe<Void, T> createStateless(final Action2<Long, ? super Observer<Observable<? extends T>>> next) {
         Func3<Void, Long, Observer<Observable<? extends T>>, Void> nextFunc = 
                 new Func3<Void, Long, Observer<Observable<? extends T>>, Void>() {
@@ -230,7 +230,7 @@ public Void call(Void state, Long requested, Observer<Observable<? extends T>> s
      * @return an AsyncOnSubscribe that emits data downstream in a protocol compatible with
      *         back-pressure.
      */
-    @Experimental
+    @Beta
     public static <T> AsyncOnSubscribe<Void, T> createStateless(final Action2<Long, ? super Observer<Observable<? extends T>>> next, 
             final Action0 onUnsubscribe) {
         Func3<Void, Long, Observer<Observable<? extends T>>, Void> nextFunc = 

src/main/java/rx/observables/SyncOnSubscribe.java

@@ -23,6 +23,7 @@
 import rx.Producer;
 import rx.Subscriber;
 import rx.Subscription;
+import rx.annotations.Beta;
 import rx.annotations.Experimental;
 import rx.exceptions.Exceptions;
 import rx.functions.Action0;
@@ -46,7 +47,7 @@
  * @param <T>
  *            the type of {@code Subscribers} that will be compatible with {@code this}.
  */
-@Experimental
+@Beta
 public abstract class SyncOnSubscribe<S, T> implements OnSubscribe<T> {
 
     /* (non-Javadoc)
@@ -126,7 +127,7 @@ protected void onUnsubscribe(S state) {
      *            next(S, Subscriber)})
      * @return a SyncOnSubscribe that emits data in a protocol compatible with back-pressure.
      */
-    @Experimental
+    @Beta
     public static <S, T> SyncOnSubscribe<S, T> createSingleState(Func0<? extends S> generator, 
             final Action2<? super S, ? super Observer<? super T>> next) {
         Func2<S, ? super Observer<? super T>, S> nextFunc = new Func2<S, Observer<? super T>, S>() {
@@ -155,7 +156,7 @@ public S call(S state, Observer<? super T> subscriber) {
      * @return a SyncOnSubscribe that emits data downstream in a protocol compatible with
      *         back-pressure.
      */
-    @Experimental
+    @Beta
     public static <S, T> SyncOnSubscribe<S, T> createSingleState(Func0<? extends S> generator, 
             final Action2<? super S, ? super Observer<? super T>> next, 
             final Action1<? super S> onUnsubscribe) {
@@ -183,7 +184,7 @@ public S call(S state, Observer<? super T> subscriber) {
      * @return a SyncOnSubscribe that emits data downstream in a protocol compatible with
      *         back-pressure.
      */
-    @Experimental
+    @Beta
     public static <S, T> SyncOnSubscribe<S, T> createStateful(Func0<? extends S> generator, 
             Func2<? super S, ? super Observer<? super T>, ? extends S> next, 
             Action1<? super S> onUnsubscribe) {
@@ -202,7 +203,7 @@ public static <S, T> SyncOnSubscribe<S, T> createStateful(Func0<? extends S> gen
      * @return a SyncOnSubscribe that emits data downstream in a protocol compatible with
      *         back-pressure.
      */
-    @Experimental
+    @Beta
     public static <S, T> SyncOnSubscribe<S, T> createStateful(Func0<? extends S> generator, 
             Func2<? super S, ? super Observer<? super T>, ? extends S> next) {
         return new SyncOnSubscribeImpl<S, T>(generator, next);
@@ -221,7 +222,7 @@ public static <S, T> SyncOnSubscribe<S, T> createStateful(Func0<? extends S> gen
      * @return a SyncOnSubscribe that emits data downstream in a protocol compatible with
      *         back-pressure.
      */
-    @Experimental
+    @Beta
     public static <T> SyncOnSubscribe<Void, T> createStateless(final Action1<? super Observer<? super T>> next) {
         Func2<Void, Observer<? super T>, Void> nextFunc = new Func2<Void, Observer<? super T>, Void>() {
             @Override
@@ -248,7 +249,7 @@ public Void call(Void state, Observer<? super T> subscriber) {
      * @return a SyncOnSubscribe that emits data downstream in a protocol compatible with
      *         back-pressure.
      */
-    @Experimental
+    @Beta
     public static <T> SyncOnSubscribe<Void, T> createStateless(final Action1<? super Observer<? super T>> next, 
             final Action0 onUnsubscribe) {
         Func2<Void, Observer<? super T>, Void> nextFunc = new Func2<Void, Observer<? super T>, Void>() {