Skip to content
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.

Sign up for GitHub

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

2.x: Add Observable switchMapX and concatMapX operators #5875

Merged
merged 2 commits into from
Mar 3, 2018
Merged

2.x: Add Observable switchMapX and concatMapX operators #5875

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b115886
2.x: Add Observable switchMapX and concatMapX operators
akarnokd Mar 2, 2018
e6168be
Rename local variables to match their datatype better.
akarnokd Mar 2, 2018
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
550 changes: 547 additions & 3 deletions src/main/java/io/reactivex/Observable.java
View file
Open in desktop

Large diffs are not rendered by default.

6 changes: 3 additions & 3 deletions src/main/java/io/reactivex/internal/operators/mixed/FlowableConcatMapSingle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -255,10 +255,10 @@ void drain() {
consumed = c;
}

SingleSource<? extends R> ms;
SingleSource<? extends R> ss;

try {
ms = ObjectHelper.requireNonNull(mapper.apply(v), "The mapper returned a null SingleSource");
ss = ObjectHelper.requireNonNull(mapper.apply(v), "The mapper returned a null SingleSource");
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
upstream.cancel();
Expand All @@ -270,7 +270,7 @@ void drain() {
}

state = STATE_ACTIVE;
ms.subscribe(inner);
ss.subscribe(inner);
break;
} else if (s == STATE_RESULT_VALUE) {
long e = emitted;
Expand Down
6 changes: 3 additions & 3 deletions src/main/java/io/reactivex/internal/operators/mixed/FlowableSwitchMapSingle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,10 +115,10 @@ public void onNext(T t) {
current.dispose();
}

SingleSource<? extends R> ms;
SingleSource<? extends R> ss;

try {
ms = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null SingleSource");
ss = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null SingleSource");
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
upstream.cancel();
Expand All @@ -135,7 +135,7 @@ public void onNext(T t) {
break;
}
if (inner.compareAndSet(current, observer)) {
ms.subscribe(observer);
ss.subscribe(observer);
break;
}
}
Expand Down
277 changes: 277 additions & 0 deletions src/main/java/io/reactivex/internal/operators/mixed/ObservableConcatMapCompletable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,277 @@
/**
* Copyright (c) 2016-present, RxJava Contributors.
*
* 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.mixed;

import java.util.concurrent.atomic.*;

import io.reactivex.*;
import io.reactivex.annotations.Experimental;
import io.reactivex.disposables.Disposable;
import io.reactivex.exceptions.*;
import io.reactivex.functions.Function;
import io.reactivex.internal.disposables.DisposableHelper;
import io.reactivex.internal.functions.ObjectHelper;
import io.reactivex.internal.fuseable.SimplePlainQueue;
import io.reactivex.internal.queue.SpscLinkedArrayQueue;
import io.reactivex.internal.util.*;
import io.reactivex.plugins.RxJavaPlugins;

/**
* Maps the upstream intems into {@link CompletableSource}s and subscribes to them one after the
* other completes or terminates (in error-delaying mode).
* @param <T> the upstream value type
* @since 2.1.11 - experimental
*/
@Experimental
public final class ObservableConcatMapCompletable<T> extends Completable {

final Observable<T> source;

final Function<? super T, ? extends CompletableSource> mapper;

final ErrorMode errorMode;

final int prefetch;

public ObservableConcatMapCompletable(Observable<T> source,
Function<? super T, ? extends CompletableSource> mapper,
ErrorMode errorMode,
int prefetch) {
this.source = source;
this.mapper = mapper;
this.errorMode = errorMode;
this.prefetch = prefetch;
}

@Override
protected void subscribeActual(CompletableObserver s) {
source.subscribe(new ConcatMapCompletableObserver<T>(s, mapper, errorMode, prefetch));
}

static final class ConcatMapCompletableObserver<T>
extends AtomicInteger
implements Observer<T>, Disposable {

private static final long serialVersionUID = 3610901111000061034L;

final CompletableObserver downstream;

final Function<? super T, ? extends CompletableSource> mapper;

final ErrorMode errorMode;

final AtomicThrowable errors;

final ConcatMapInnerObserver inner;

final int prefetch;

final SimplePlainQueue<T> queue;

Disposable upstream;

volatile boolean active;

volatile boolean done;

volatile boolean disposed;

ConcatMapCompletableObserver(CompletableObserver downstream,
Function<? super T, ? extends CompletableSource> mapper,
ErrorMode errorMode, int prefetch) {
this.downstream = downstream;
this.mapper = mapper;
this.errorMode = errorMode;
this.prefetch = prefetch;
this.errors = new AtomicThrowable();
this.inner = new ConcatMapInnerObserver(this);
this.queue = new SpscLinkedArrayQueue<T>(prefetch);
}

@Override
public void onSubscribe(Disposable s) {
if (DisposableHelper.validate(upstream, s)) {
this.upstream = s;
downstream.onSubscribe(this);
}
}

@Override
public void onNext(T t) {
queue.offer(t);
drain();
}

@Override
public void onError(Throwable t) {
if (errors.addThrowable(t)) {
if (errorMode == ErrorMode.IMMEDIATE) {
disposed = true;
inner.dispose();
t = errors.terminate();
if (t != ExceptionHelper.TERMINATED) {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this path does not seem to be covered. same for 126. Is travis just flaky again?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's a probabilistic race that the CI tends to not hit, no matter the race-loop count. It gets covered 80% of the time and unfortunately, I'd have to hack the state of the operator to get it execute that inner part a second time.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah thought so. Then never mind.

downstream.onError(t);
}
if (getAndIncrement() == 0) {
queue.clear();
}
} else {
done = true;
drain();
}
} else {
RxJavaPlugins.onError(t);
}
}

@Override
public void onComplete() {
done = true;
drain();
}

@Override
public void dispose() {
disposed = true;
upstream.dispose();
inner.dispose();
if (getAndIncrement() == 0) {
queue.clear();
}
}

@Override
public boolean isDisposed() {
return disposed;
}

void innerError(Throwable ex) {
if (errors.addThrowable(ex)) {
if (errorMode == ErrorMode.IMMEDIATE) {
disposed = true;
upstream.dispose();
ex = errors.terminate();
if (ex != ExceptionHelper.TERMINATED) {
downstream.onError(ex);
}
if (getAndIncrement() == 0) {
queue.clear();
}
} else {
active = false;
drain();
}
} else {
RxJavaPlugins.onError(ex);
}
}

void innerComplete() {
active = false;
drain();
}

void drain() {
if (getAndIncrement() != 0) {
return;
}

do {
if (disposed) {
queue.clear();
return;
}

if (!active) {

if (errorMode == ErrorMode.BOUNDARY) {
if (errors.get() != null) {
disposed = true;
queue.clear();
Throwable ex = errors.terminate();
downstream.onError(ex);
return;
}
}

boolean d = done;
T v = queue.poll();
boolean empty = v == null;

if (d && empty) {
disposed = true;
Throwable ex = errors.terminate();
if (ex != null) {
downstream.onError(ex);
} else {
downstream.onComplete();
}
return;
}

if (!empty) {

CompletableSource cs;

try {
cs = ObjectHelper.requireNonNull(mapper.apply(v), "The mapper returned a null CompletableSource");
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
disposed = true;
queue.clear();
upstream.dispose();
errors.addThrowable(ex);
ex = errors.terminate();
downstream.onError(ex);
return;
}
active = true;
cs.subscribe(inner);
}
}
} while (decrementAndGet() != 0);
}

static final class ConcatMapInnerObserver extends AtomicReference<Disposable>
implements CompletableObserver {

private static final long serialVersionUID = 5638352172918776687L;

final ConcatMapCompletableObserver<?> parent;

ConcatMapInnerObserver(ConcatMapCompletableObserver<?> parent) {
this.parent = parent;
}

@Override
public void onSubscribe(Disposable d) {
DisposableHelper.replace(this, d);
}

@Override
public void onError(Throwable e) {
parent.innerError(e);
}

@Override
public void onComplete() {
parent.innerComplete();
}

void dispose() {
DisposableHelper.dispose(this);
}
}
}
}
Loading