Skip to content

2.x: factor out inner classes from the base reactive types #4360

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Aug 17, 2016
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 1 addition & 7 deletions src/main/java/io/reactivex/Completable.java
Original file line number Diff line number Diff line change
Expand Up @@ -323,13 +323,7 @@ public static Completable fromCallable(final Callable<?> callable) {
@SchedulerSupport(SchedulerSupport.NONE)
public static Completable fromFuture(final Future<?> future) {
Objects.requireNonNull(future, "future is null");
return fromCallable(new Callable<Object>() {
@Override
public Object call() throws Exception {
future.get();
return null;
}
});
return fromAction(Functions.futureAction(future));
}

/**
Expand Down
505 changes: 84 additions & 421 deletions src/main/java/io/reactivex/Flowable.java

Large diffs are not rendered by default.

440 changes: 63 additions & 377 deletions src/main/java/io/reactivex/Observable.java

Large diffs are not rendered by default.

74 changes: 8 additions & 66 deletions src/main/java/io/reactivex/Single.java
Original file line number Diff line number Diff line change
Expand Up @@ -13,16 +13,14 @@

package io.reactivex;

import java.util.*;
import java.util.concurrent.*;

import org.reactivestreams.*;

import io.reactivex.disposables.Disposable;
import io.reactivex.exceptions.Exceptions;
import io.reactivex.functions.*;
import io.reactivex.internal.functions.Functions;
import io.reactivex.internal.functions.Objects;
import io.reactivex.internal.functions.*;
import io.reactivex.internal.operators.completable.CompletableFromSingle;
import io.reactivex.internal.operators.flowable.*;
import io.reactivex.internal.operators.flowable.FlowableConcatMap.ErrorMode;
Expand Down Expand Up @@ -87,12 +85,7 @@ public static <T> Single<T> amb(final Iterable<? extends SingleSource<? extends
@SuppressWarnings("unchecked")
public static <T> Single<T> amb(final SingleSource<? extends T>... sources) {
if (sources.length == 0) {
return error(new Callable<Throwable>() {
@Override
public Throwable call() {
return new NoSuchElementException();
}
});
return error(SingleInternalHelper.<T>emptyThrower());
}
if (sources.length == 1) {
return wrap((SingleSource<T>)sources[0]);
Expand Down Expand Up @@ -130,13 +123,7 @@ public static <T> Flowable<T> concat(Iterable<? extends SingleSource<? extends T
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
public static <T> Flowable<T> concat(Publisher<? extends SingleSource<? extends T>> sources) {
return new FlowableConcatMap(sources,
new Function<SingleSource<? extends T>, Publisher<? extends T>>() {
@Override
public Publisher<? extends T> apply(SingleSource<? extends T> v){
return new SingleToFlowable<T>(v);
}
}, 2, ErrorMode.IMMEDIATE);
return new FlowableConcatMap(sources, SingleInternalHelper.toFlowable(), 2, ErrorMode.IMMEDIATE);
}

/**
Expand Down Expand Up @@ -520,12 +507,7 @@ public static <T> Single<T> error(final Callable<? extends Throwable> errorSuppl
*/
public static <T> Single<T> error(final Throwable exception) {
Objects.requireNonNull(exception, "error is null");
return error(new Callable<Throwable>() {
@Override
public Throwable call() {
return exception;
}
});
return error(Functions.justCallable(exception));
}

/**
Expand Down Expand Up @@ -740,13 +722,7 @@ public static <T> Flowable<T> merge(Iterable<? extends SingleSource<? extends T>
*/
@SuppressWarnings({ "unchecked", "rawtypes" })
public static <T> Flowable<T> merge(Publisher<? extends SingleSource<? extends T>> sources) {
return new FlowableFlatMap(sources,
new Function<SingleSource<? extends T>, Publisher<? extends T>>() {
@Override
public Publisher<? extends T> apply(SingleSource<? extends T> v){
return new SingleToFlowable<T>(v);
}
}, false, Integer.MAX_VALUE, Flowable.bufferSize());
return new FlowableFlatMap(sources, SingleInternalHelper.toFlowable(), false, Integer.MAX_VALUE, Flowable.bufferSize());
}

/**
Expand Down Expand Up @@ -1289,31 +1265,7 @@ static <T> Single<T> wrap(SingleSource<T> source) {
*/
public static <T, R> Single<R> zip(final Iterable<? extends SingleSource<? extends T>> sources, Function<? super Object[], ? extends R> zipper) {
Objects.requireNonNull(sources, "sources is null");

Iterable<? extends Flowable<T>> it = new Iterable<Flowable<T>>() {
@Override
public Iterator<Flowable<T>> iterator() {
final Iterator<? extends SingleSource<? extends T>> sit = sources.iterator();
return new Iterator<Flowable<T>>() {

@Override
public boolean hasNext() {
return sit.hasNext();
}

@Override
public Flowable<T> next() {
return new SingleToFlowable<T>(sit.next());
}

@Override
public void remove() {
throw new UnsupportedOperationException();
}
};
}
};
return Flowable.zipIterable(it, zipper, false, 1).toSingle();
return Flowable.zipIterable(SingleInternalHelper.iterableToFlowable(sources), zipper, false, 1).toSingle();
}

/**
Expand Down Expand Up @@ -1826,12 +1778,7 @@ public final Single<T> cache() {
*/
public final <U> Single<U> cast(final Class<? extends U> clazz) {
Objects.requireNonNull(clazz, "clazz is null");
return map(new Function<T, U>() {
@Override
public U apply(T v) {
return clazz.cast(v);
}
});
return map(Functions.castFunction(clazz));
}

/**
Expand Down Expand Up @@ -2345,12 +2292,7 @@ public final Single<T> onErrorReturnValue(final T value) {
*/
public final Single<T> onErrorResumeNext(final Single<? extends T> resumeSingleInCaseOfError) {
Objects.requireNonNull(resumeSingleInCaseOfError, "resumeSingleInCaseOfError is null");
return onErrorResumeNext(new Function<Throwable, Single<? extends T>>() {
@Override
public Single<? extends T> apply(Throwable t) throws Exception {
return resumeSingleInCaseOfError;
}
});
return onErrorResumeNext(Functions.justFunction(resumeSingleInCaseOfError));
}

/**
Expand Down
Loading