2.x: count, elementAt, ingoreElements, last, single, reduce, reduceWith (#4576)

to return non-Flowable

Author: akarnokd

src/main/java/io/reactivex/Flowable.java

@@ -3159,7 +3159,7 @@ public final Maybe<T> retry() {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Maybe<T> retry(BiPredicate<? super Integer, ? super Throwable> predicate) {
-        return toFlowable().retry(predicate).toMaybe();
+        return toFlowable().retry(predicate).singleElement();
     }
 
     /**
@@ -3199,7 +3199,7 @@ public final Maybe<T> retry(long count) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Maybe<T> retry(long times, Predicate<? super Throwable> predicate) {
-        return toFlowable().retry(times, predicate).toMaybe();
+        return toFlowable().retry(times, predicate).singleElement();
     }
 
     /**
@@ -3283,7 +3283,7 @@ public final Maybe<T> retryUntil(final BooleanSupplier stop) {
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Maybe<T> retryWhen(
             final Function<? super Flowable<? extends Throwable>, ? extends Publisher<?>> handler) {
-        return toFlowable().retryWhen(handler).toMaybe();
+        return toFlowable().retryWhen(handler).singleElement();
     }
 
     /**

src/main/java/io/reactivex/Maybe.java

@@ -453,7 +453,7 @@ public static <T> Single<T> fromCallable(final Callable<? extends T> callable) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public static <T> Single<T> fromFuture(Future<? extends T> future) {
-        return Flowable.<T>fromFuture(future).toSingle();
+        return toSingle(Flowable.<T>fromFuture(future));
     }
 
     /**
@@ -485,7 +485,7 @@ public static <T> Single<T> fromFuture(Future<? extends T> future) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public static <T> Single<T> fromFuture(Future<? extends T> future, long timeout, TimeUnit unit) {
-        return Flowable.<T>fromFuture(future, timeout, unit).toSingle();
+        return toSingle(Flowable.<T>fromFuture(future, timeout, unit));
     }
 
     /**
@@ -519,7 +519,7 @@ public static <T> Single<T> fromFuture(Future<? extends T> future, long timeout,
      */
     @SchedulerSupport(SchedulerSupport.CUSTOM)
     public static <T> Single<T> fromFuture(Future<? extends T> future, long timeout, TimeUnit unit, Scheduler scheduler) {
-        return Flowable.<T>fromFuture(future, timeout, unit, scheduler).toSingle();
+        return toSingle(Flowable.<T>fromFuture(future, timeout, unit, scheduler));
     }
 
     /**
@@ -548,7 +548,7 @@ public static <T> Single<T> fromFuture(Future<? extends T> future, long timeout,
      */
     @SchedulerSupport(SchedulerSupport.CUSTOM)
     public static <T> Single<T> fromFuture(Future<? extends T> future, Scheduler scheduler) {
-        return Flowable.<T>fromFuture(future, scheduler).toSingle();
+        return toSingle(Flowable.<T>fromFuture(future, scheduler));
     }
 
     /**
@@ -978,7 +978,7 @@ public static <T> Single<T> wrap(SingleSource<T> source) {
     @SchedulerSupport(SchedulerSupport.NONE)
     public static <T, R> Single<R> zip(final Iterable<? extends SingleSource<? extends T>> sources, Function<? super Object[], ? extends R> zipper) {
         ObjectHelper.requireNonNull(sources, "sources is null");
-        return Flowable.zipIterable(SingleInternalHelper.iterableToFlowable(sources), zipper, false, 1).toSingle();
+        return toSingle(Flowable.zipIterable(SingleInternalHelper.iterableToFlowable(sources), zipper, false, 1));
     }
 
     /**
@@ -1418,7 +1418,7 @@ public static <T, R> Single<R> zipArray(Function<? super Object[], ? extends R>
             sourcePublishers[i] = RxJavaPlugins.onAssembly(new SingleToFlowable<T>(s));
             i++;
         }
-        return Flowable.zipArray(zipper, false, 1, sourcePublishers).toSingle();
+        return toSingle(Flowable.zipArray(zipper, false, 1, sourcePublishers));
     }
 
     /**
@@ -2252,7 +2252,7 @@ public final Flowable<T> repeatUntil(BooleanSupplier stop) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Single<T> retry() {
-        return toFlowable().retry().toSingle();
+        return toSingle(toFlowable().retry());
     }
 
     /**
@@ -2268,7 +2268,7 @@ public final Single<T> retry() {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Single<T> retry(long times) {
-        return toFlowable().retry(times).toSingle();
+        return toSingle(toFlowable().retry(times));
     }
 
     /**
@@ -2285,7 +2285,7 @@ public final Single<T> retry(long times) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Single<T> retry(BiPredicate<? super Integer, ? super Throwable> predicate) {
-        return toFlowable().retry(predicate).toSingle();
+        return toSingle(toFlowable().retry(predicate));
     }
 
     /**
@@ -2302,7 +2302,7 @@ public final Single<T> retry(BiPredicate<? super Integer, ? super Throwable> pre
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Single<T> retry(Predicate<? super Throwable> predicate) {
-        return toFlowable().retry(predicate).toSingle();
+        return toSingle(toFlowable().retry(predicate));
     }
 
     /**
@@ -2323,7 +2323,7 @@ public final Single<T> retry(Predicate<? super Throwable> predicate) {
      */
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Single<T> retryWhen(Function<? super Flowable<? extends Throwable>, ? extends Publisher<Object>> handler) {
-        return toFlowable().retryWhen(handler).toSingle();
+        return toSingle(toFlowable().retryWhen(handler));
     }
 
     /**
@@ -2844,4 +2844,8 @@ public final TestObserver<T> test(boolean cancelled) {
         subscribe(ts);
         return ts;
     }
+
+    private static <T> Single<T> toSingle(Flowable<T> source) {
+        return RxJavaPlugins.onAssembly(new FlowableSingleSingle<T>(source, null));
+    }
 }

src/main/java/io/reactivex/Single.java

@@ -0,0 +1,91 @@
+/**
+ * Copyright 2016 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 io.reactivex.internal.operators.flowable;
+
+import org.reactivestreams.*;
+
+import io.reactivex.*;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.fuseable.FuseToFlowable;
+import io.reactivex.internal.subscriptions.SubscriptionHelper;
+import io.reactivex.plugins.RxJavaPlugins;
+
+public final class FlowableCountSingle<T> extends Single<Long> implements FuseToFlowable<Long> {
+
+    final Publisher<T> source;
+
+    public FlowableCountSingle(Publisher<T> source) {
+        this.source = source;
+    }
+
+    @Override
+    protected void subscribeActual(SingleObserver<? super Long> s) {
+        source.subscribe(new CountSubscriber(s));
+    }
+
+    @Override
+    public Flowable<Long> fuseToFlowable() {
+        return RxJavaPlugins.onAssembly(new FlowableCount<T>(source));
+    }
+
+    static final class CountSubscriber implements Subscriber<Object>, Disposable {
+
+        final SingleObserver<? super Long> actual;
+
+        Subscription s;
+
+        long count;
+
+        CountSubscriber(SingleObserver<? super Long> actual) {
+            this.actual = actual;
+        }
+
+        @Override
+        public void onSubscribe(Subscription s) {
+            if (SubscriptionHelper.validate(this.s, s)) {
+                this.s = s;
+                actual.onSubscribe(this);
+                s.request(Long.MAX_VALUE);
+            }
+        }
+
+        @Override
+        public void onNext(Object t) {
+            count++;
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            s = SubscriptionHelper.CANCELLED;
+            actual.onError(t);
+        }
+
+        @Override
+        public void onComplete() {
+            s = SubscriptionHelper.CANCELLED;
+            actual.onSuccess(count);
+        }
+
+        @Override
+        public void dispose() {
+            s.cancel();
+            s = SubscriptionHelper.CANCELLED;
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return s == SubscriptionHelper.CANCELLED;
+        }
+    }
+}

src/main/java/io/reactivex/internal/operators/flowable/FlowableCountSingle.java

@@ -16,6 +16,7 @@
 import org.reactivestreams.*;
 
 import io.reactivex.internal.subscriptions.*;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class FlowableElementAt<T> extends AbstractFlowableWithUpstream<T, T> {
     final long index;
@@ -77,6 +78,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable t) {
             if (done) {
+                RxJavaPlugins.onError(t);
                 return;
             }
             done = true;
@@ -89,7 +91,7 @@ public void onComplete() {
                 done = true;
                 T v = defaultValue;
                 if (v == null) {
-                    actual.onError(new IndexOutOfBoundsException());
+                    actual.onComplete();
                 } else {
                     complete(v);
                 }

src/main/java/io/reactivex/internal/operators/flowable/FlowableElementAt.java

@@ -0,0 +1,118 @@
+/**
+ * Copyright 2016 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 io.reactivex.internal.operators.flowable;
+
+import org.reactivestreams.*;
+
+import io.reactivex.*;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.fuseable.FuseToFlowable;
+import io.reactivex.internal.subscriptions.SubscriptionHelper;
+import io.reactivex.plugins.RxJavaPlugins;
+
+public final class FlowableElementAtMaybe<T> extends Maybe<T> implements FuseToFlowable<T> {
+    final Publisher<T> source;
+
+    final long index;
+
+    public FlowableElementAtMaybe(Publisher<T> source, long index) {
+        this.source = source;
+        this.index = index;
+    }
+
+    @Override
+    protected void subscribeActual(MaybeObserver<? super T> s) {
+        source.subscribe(new ElementAtSubscriber<T>(s, index));
+    }
+
+    @Override
+    public Flowable<T> fuseToFlowable() {
+        return RxJavaPlugins.onAssembly(new FlowableElementAt<T>(source, index, null));
+    }
+
+    static final class ElementAtSubscriber<T> implements Subscriber<T>, Disposable {
+
+        final MaybeObserver<? super T> actual;
+
+        final long index;
+
+        Subscription s;
+
+        long count;
+
+        boolean done;
+
+        ElementAtSubscriber(MaybeObserver<? super T> actual, long index) {
+            this.actual = actual;
+            this.index = index;
+        }
+
+        @Override
+        public void onSubscribe(Subscription s) {
+            if (SubscriptionHelper.validate(this.s, s)) {
+                this.s = s;
+                actual.onSubscribe(this);
+                s.request(Long.MAX_VALUE);
+            }
+        }
+
+        @Override
+        public void onNext(T t) {
+            if (done) {
+                return;
+            }
+            long c = count;
+            if (c == index) {
+                done = true;
+                s.cancel();
+                s = SubscriptionHelper.CANCELLED;
+                actual.onSuccess(t);
+                return;
+            }
+            count = c + 1;
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            if (done) {
+                RxJavaPlugins.onError(t);
+                return;
+            }
+            done = true;
+            s = SubscriptionHelper.CANCELLED;
+            actual.onError(t);
+        }
+
+        @Override
+        public void onComplete() {
+            s = SubscriptionHelper.CANCELLED;
+            if (index <= count && !done) {
+                done = true;
+                actual.onComplete();
+            }
+        }
+
+        @Override
+        public void dispose() {
+            s.cancel();
+            s = SubscriptionHelper.CANCELLED;
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return s == SubscriptionHelper.CANCELLED;
+        }
+
+    }
+}