Skip to content

1.x: map() and filter() should unsubscribe on crash eagerly #3890

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

Merged
merged 1 commit into from
Apr 29, 2016
Merged

1.x: map() and filter() should unsubscribe on crash eagerly #3890

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
86 changes: 61 additions & 25 deletions src/main/java/rx/internal/operators/OperatorFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,15 +15,17 @@
*/
package rx.internal.operators;

import rx.*;
import rx.Observable.Operator;
import rx.Subscriber;
import rx.exceptions.*;
import rx.functions.Func1;
import rx.internal.util.RxJavaPluginUtils;

/**
* Filters an Observable by discarding any items it emits that do not meet some test.
* <p>
* <img width="640" src="https://github.com/ReactiveX/RxJava/wiki/images/rx-operators/filter.png" alt="">
* @param <T> the value type
*/
public final class OperatorFilter<T> implements Operator<T, T> {

Expand All @@ -35,33 +37,67 @@ public OperatorFilter(Func1<? super T, Boolean> predicate) {

@Override
public Subscriber<? super T> call(final Subscriber<? super T> child) {
return new Subscriber<T>(child) {
FilterSubscriber<T> parent = new FilterSubscriber<T>(child, predicate);
child.add(parent);
return parent;
}

@Override
public void onCompleted() {
child.onCompleted();
}
static final class FilterSubscriber<T> extends Subscriber<T> {

final Subscriber<? super T> actual;

final Func1<? super T, Boolean> predicate;

@Override
public void onError(Throwable e) {
child.onError(e);
boolean done;

public FilterSubscriber(Subscriber<? super T> actual, Func1<? super T, Boolean> predicate) {
this.actual = actual;
this.predicate = predicate;
}

@Override
public void onNext(T t) {
boolean result;

try {
result = predicate.call(t);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
unsubscribe();
onError(OnErrorThrowable.addValueAsLastCause(ex, t));
return;
}

@Override
public void onNext(T t) {
try {
if (predicate.call(t)) {
child.onNext(t);
} else {
// TODO consider a more complicated version that batches these
request(1);
}
} catch (Throwable e) {
Exceptions.throwOrReport(e, child, t);
}

if (result) {
actual.onNext(t);
} else {
request(1);
}

};
}

@Override
public void onError(Throwable e) {
if (done) {
RxJavaPluginUtils.handleException(e);
return;
}
done = true;

actual.onError(e);
}


@Override
public void onCompleted() {
if (done) {
return;
}
actual.onCompleted();
}
@Override
public void setProducer(Producer p) {
super.setProducer(p);
actual.setProducer(p);
}
}

}
80 changes: 60 additions & 20 deletions src/main/java/rx/internal/operators/OperatorMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,16 +15,20 @@
*/
package rx.internal.operators;

import rx.*;
import rx.Observable.Operator;
import rx.Subscriber;
import rx.exceptions.Exceptions;
import rx.exceptions.*;
import rx.functions.Func1;
import rx.internal.util.RxJavaPluginUtils;

/**
* Applies a function of your choosing to every item emitted by an {@code Observable}, and emits the results of
* this transformation as a new {@code Observable}.
* <p>
* <img width="640" height="305" src="https://raw.githubusercontent.com/wiki/ReactiveX/RxJava/images/rx-operators/map.png" alt="">
*
* @param <T> the input value type
* @param <R> the return value type
*/
public final class OperatorMap<T, R> implements Operator<R, T> {

Expand All @@ -36,28 +40,64 @@ public OperatorMap(Func1<? super T, ? extends R> transformer) {

@Override
public Subscriber<? super T> call(final Subscriber<? super R> o) {
return new Subscriber<T>(o) {
MapSubscriber<T, R> parent = new MapSubscriber<T, R>(o, transformer);
o.add(parent);
return parent;
}

static final class MapSubscriber<T, R> extends Subscriber<T> {

final Subscriber<? super R> actual;

final Func1<? super T, ? extends R> mapper;

@Override
public void onCompleted() {
o.onCompleted();
boolean done;

public MapSubscriber(Subscriber<? super R> actual, Func1<? super T, ? extends R> mapper) {
this.actual = actual;
this.mapper = mapper;
}

@Override
public void onNext(T t) {
R result;

try {
result = mapper.call(t);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
unsubscribe();
onError(OnErrorThrowable.addValueAsLastCause(ex, t));
return;
}

@Override
public void onError(Throwable e) {
o.onError(e);

actual.onNext(result);
}

@Override
public void onError(Throwable e) {
if (done) {
RxJavaPluginUtils.handleException(e);
return;
}

@Override
public void onNext(T t) {
try {
o.onNext(transformer.call(t));
} catch (Throwable e) {
Exceptions.throwOrReport(e, this, t);
}
done = true;

actual.onError(e);
}


@Override
public void onCompleted() {
if (done) {
return;
}

};
actual.onCompleted();
}

@Override
public void setProducer(Producer p) {
actual.setProducer(p);
Copy link
Contributor

Choose a reason for hiding this comment

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

Missing a call to super.setProducer?

Copy link
Member Author

Choose a reason for hiding this comment

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

Map doesn't request on its own, unlike filter, so it can just pass along the producer to downstream.

Copy link
Contributor

Choose a reason for hiding this comment

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

Ah, yeah. Makes sense.

}
}

}
Expand Down
27 changes: 27 additions & 0 deletions src/test/java/rx/internal/operators/OperatorFilterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import rx.functions.*;
import rx.internal.util.RxRingBuffer;
import rx.observers.TestSubscriber;
import rx.subjects.PublishSubject;

public class OperatorFilterTest {

Expand All @@ -54,6 +55,7 @@ public Boolean call(String t1) {

/**
* Make sure we are adjusting subscriber.request() for filtered items
* @throws InterruptedException on interrupt
*/
@Test(timeout = 500)
public void testWithBackpressure() throws InterruptedException {
Expand Down Expand Up @@ -100,6 +102,7 @@ public void onNext(String t) {

/**
* Make sure we are adjusting subscriber.request() for filtered items
* @throws InterruptedException on interrupt
*/
@Test(timeout = 500000)
public void testWithBackpressure2() throws InterruptedException {
Expand Down Expand Up @@ -167,4 +170,28 @@ public void call(Integer t) {
}
}
}

@Test
public void functionCrashUnsubscribes() {

PublishSubject<Integer> ps = PublishSubject.create();

TestSubscriber<Integer> ts = new TestSubscriber<Integer>();

ps.filter(new Func1<Integer, Boolean>() {
@Override
public Boolean call(Integer v) {
throw new TestException();
}
}).unsafeSubscribe(ts);

Assert.assertTrue("Not subscribed?", ps.hasObservers());

ps.onNext(1);

Assert.assertFalse("Subscribed?", ps.hasObservers());

ts.assertError(TestException.class);
}

}
46 changes: 31 additions & 15 deletions src/test/java/rx/internal/operators/OperatorMapTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,28 +16,21 @@
package rx.internal.operators;

import static org.mockito.Matchers.any;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.*;

import java.util.HashMap;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.*;

import org.junit.Before;
import org.junit.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.junit.*;
import org.mockito.*;

import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.exceptions.OnErrorNotImplementedException;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.internal.operators.OperatorMap;
import rx.exceptions.*;
import rx.functions.*;
import rx.observers.TestSubscriber;
import rx.schedulers.Schedulers;
import rx.subjects.PublishSubject;

public class OperatorMapTest {

Expand Down Expand Up @@ -339,4 +332,27 @@ public void call(String s) {
}
});
}

@Test
public void functionCrashUnsubscribes() {

PublishSubject<Integer> ps = PublishSubject.create();

TestSubscriber<Integer> ts = new TestSubscriber<Integer>();

ps.map(new Func1<Integer, Integer>() {
@Override
public Integer call(Integer v) {
throw new TestException();
}
}).unsafeSubscribe(ts);

Assert.assertTrue("Not subscribed?", ps.hasObservers());

ps.onNext(1);

Assert.assertFalse("Subscribed?", ps.hasObservers());

ts.assertError(TestException.class);
}
}