Skip to content

Commit 0954ee8

Browse files
davidmotendavidmoten
committed
add OperatorTakeLastOne
1 parent 5b75d32 commit 0954ee8

File tree

4 files changed

+323
-1
lines changed

4 files changed

+323
-1
lines changed

‎src/main/java/rx/Observable.java

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7771,7 +7771,12 @@ public final Observable<T> takeFirst(Func1<? super T, Boolean> predicate) {
77717771
* @see <a href="http://reactivex.io/documentation/operators/takelast.html">ReactiveX operators documentation: TakeLast</a>
77727772
*/
77737773
public final Observable<T> takeLast(final int count) {
7774-
return lift(new OperatorTakeLast<T>(count));
7774+
if (count == 0)
7775+
return ignoreElements();
7776+
else if (count == 1 )
7777+
return lift(OperatorTakeLastOne.<T>instance());
7778+
else
7779+
return lift(new OperatorTakeLast<T>(count));
77757780
}
77767781

77777782
/**

‎src/main/java/rx/internal/operators/OperatorTakeLastOne.java

Lines changed: 168 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,168 @@
1+
package rx.internal.operators;
2+
3+
import java.util.concurrent.atomic.AtomicInteger;
4+
5+
import rx.Observable.Operator;
6+
import rx.Producer;
7+
import rx.Subscriber;
8+
9+
public class OperatorTakeLastOne<T> implements Operator<T, T> {
10+
11+
private static class Holder {
12+
static final OperatorTakeLastOne<Object> INSTANCE = new OperatorTakeLastOne<Object>();
13+
}
14+
15+
@SuppressWarnings("unchecked")
16+
public static <T> OperatorTakeLastOne<T> instance() {
17+
return (OperatorTakeLastOne<T>) Holder.INSTANCE;
18+
}
19+
20+
private OperatorTakeLastOne() {
21+
22+
}
23+
24+
@Override
25+
public Subscriber<? super T> call(Subscriber<? super T> child) {
26+
final ParentSubscriber<T> parent = new ParentSubscriber<T>(child);
27+
child.setProducer(new Producer() {
28+
29+
@Override
30+
public void request(long n) {
31+
parent.requestMore(n);
32+
}
33+
});
34+
child.add(parent);
35+
return parent;
36+
}
37+
38+
private static class ParentSubscriber<T> extends Subscriber<T> {
39+
40+
private final static int NOT_REQUESTED_NOT_COMPLETED = 0;
41+
private final static int NOT_REQUESTED_COMPLETED = 1;
42+
private final static int REQUESTED_NOT_COMPLETED = 2;
43+
private final static int REQUESTED_COMPLETED = 3;
44+
45+
/*
46+
* These are the expected state transitions:
47+
*
48+
* NOT_REQUESTED_NOT_COMPLETED --> REQUESTED_NOT_COMPLETED
49+
* | |
50+
* V V
51+
* NOT_REQUESTED_COMPLETED --> REQUESTED_COMPLETED
52+
*
53+
* Once at REQUESTED_COMPLETED we emit the last value if one exists
54+
*/
55+
56+
// Used as the initial value of last
57+
private static final Object ABSENT = new Object();
58+
59+
// the downstream subscriber
60+
private final Subscriber<? super T> child;
61+
62+
@SuppressWarnings("unchecked")
63+
// we can get away with this cast at runtime because of type erasure
64+
private T last = (T) ABSENT;
65+
66+
// holds the current state of the stream so that we can make atomic
67+
// updates to it
68+
private final AtomicInteger state = new AtomicInteger(NOT_REQUESTED_NOT_COMPLETED);
69+
70+
ParentSubscriber(Subscriber<? super T> child) {
71+
this.child = child;
72+
}
73+
74+
void requestMore(long n) {
75+
if (n > 0) {
76+
// CAS loop to atomically change state given that onCompleted()
77+
// or another requestMore() may be acting concurrently
78+
while (true) {
79+
// read the value of state and then try state transitions
80+
// only if the value of state does not change in the
81+
// meantime (in another requestMore() or onCompleted()). If
82+
// the value has changed and we expect to do a transition
83+
// still then we loop and try again.
84+
final int s = state.get();
85+
if (s == NOT_REQUESTED_NOT_COMPLETED) {
86+
if (state.compareAndSet(NOT_REQUESTED_NOT_COMPLETED,
87+
REQUESTED_NOT_COMPLETED)) {
88+
return;
89+
}
90+
} else if (s == NOT_REQUESTED_COMPLETED) {
91+
if (state.compareAndSet(NOT_REQUESTED_COMPLETED, REQUESTED_COMPLETED)) {
92+
emit();
93+
return;
94+
}
95+
} else
96+
// already requested so we exit
97+
return;
98+
}
99+
}
100+
}
101+
102+
@Override
103+
public void onCompleted() {
104+
// CAS loop to atomically change state given that requestMore()
105+
// may be acting concurrently
106+
while (true) {
107+
// read the value of state and then try state transitions
108+
// only if the value of state does not change in the meantime
109+
// (in another requestMore()). If the value has changed and
110+
// we expect to do a transition still then we loop and try
111+
// again.
112+
final int s = state.get();
113+
if (s == NOT_REQUESTED_NOT_COMPLETED) {
114+
if (state.compareAndSet(NOT_REQUESTED_NOT_COMPLETED, NOT_REQUESTED_COMPLETED)) {
115+
return;
116+
}
117+
} else if (s == REQUESTED_NOT_COMPLETED) {
118+
if (state.compareAndSet(REQUESTED_NOT_COMPLETED, REQUESTED_COMPLETED)) {
119+
emit();
120+
return;
121+
}
122+
} else
123+
// already completed so we exit
124+
return;
125+
}
126+
}
127+
128+
/**
129+
* If not unsubscribed then emits last value and completed to the child
130+
* subscriber.
131+
*/
132+
private void emit() {
133+
if (isUnsubscribed()) {
134+
// release for gc
135+
last = null;
136+
return;
137+
}
138+
// Note that last is safely published despite not being volatile
139+
// because a CAS update must have happened in the current thread just before
140+
// emit() was called
141+
T t = last;
142+
// release for gc
143+
last = null;
144+
if (t != ABSENT) {
145+
try {
146+
child.onNext(t);
147+
} catch (Throwable e) {
148+
child.onError(e);
149+
return;
150+
}
151+
}
152+
if (!isUnsubscribed())
153+
child.onCompleted();
154+
}
155+
156+
@Override
157+
public void onError(Throwable e) {
158+
child.onError(e);
159+
}
160+
161+
@Override
162+
public void onNext(T t) {
163+
last = t;
164+
}
165+
166+
}
167+
168+
}

‎src/perf/java/rx/operators/OperatorTakeLastOnePerf.java

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
package rx.operators;
2+
3+
import org.openjdk.jmh.annotations.Benchmark;
4+
import org.openjdk.jmh.annotations.Param;
5+
import org.openjdk.jmh.annotations.Scope;
6+
import org.openjdk.jmh.annotations.State;
7+
8+
import rx.internal.operators.OperatorTakeLast;
9+
import rx.internal.operators.OperatorTakeLastOne;
10+
import rx.jmh.InputWithIncrementingInteger;
11+
12+
public class OperatorTakeLastOnePerf {
13+
14+
private static final OperatorTakeLast<Integer> TAKE_LAST = new OperatorTakeLast<Integer>(1);
15+
16+
@State(Scope.Thread)
17+
public static class Input extends InputWithIncrementingInteger {
18+
19+
@Param({ "5", "100", "1000000" })
20+
public int size;
21+
22+
@Override
23+
public int getSize() {
24+
return size;
25+
}
26+
27+
}
28+
29+
@Benchmark
30+
public void takeLastOneUsingTakeLast(Input input) {
31+
input.observable.lift(TAKE_LAST).subscribe(input.observer);
32+
}
33+
34+
@Benchmark
35+
public void takeLastOneUsingTakeLastOne(Input input) {
36+
input.observable.lift(OperatorTakeLastOne.<Integer>instance()).subscribe(input.observer);
37+
}
38+
39+
}

‎src/test/java/rx/internal/operators/OperatorTakeLastOneTest.java

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
package rx.internal.operators;
2+
3+
import static org.junit.Assert.assertEquals;
4+
import static org.junit.Assert.assertTrue;
5+
6+
import java.util.ArrayList;
7+
import java.util.Arrays;
8+
import java.util.Collections;
9+
import java.util.List;
10+
import java.util.concurrent.atomic.AtomicBoolean;
11+
12+
import org.junit.Test;
13+
14+
import rx.Observable;
15+
import rx.Subscriber;
16+
import rx.functions.Action0;
17+
import rx.observers.TestSubscriber;
18+
19+
public class OperatorTakeLastOneTest {
20+
21+
@Test
22+
public void testLastOfManyReturnsLast() {
23+
TestSubscriber<Integer> s = new TestSubscriber<Integer>();
24+
Observable.range(1, 10).takeLast(1).subscribe(s);
25+
s.assertReceivedOnNext(Arrays.asList(10));
26+
s.assertNoErrors();
27+
s.assertTerminalEvent();
28+
s.assertUnsubscribed();
29+
}
30+
31+
@Test
32+
public void testLastOfEmptyReturnsEmpty() {
33+
TestSubscriber<Object> s = new TestSubscriber<Object>();
34+
Observable.empty().takeLast(1).subscribe(s);
35+
s.assertReceivedOnNext(Collections.emptyList());
36+
s.assertNoErrors();
37+
s.assertTerminalEvent();
38+
s.assertUnsubscribed();
39+
}
40+
41+
@Test
42+
public void testLastOfOneReturnsLast() {
43+
TestSubscriber<Integer> s = new TestSubscriber<Integer>();
44+
Observable.just(1).takeLast(1).subscribe(s);
45+
s.assertReceivedOnNext(Arrays.asList(1));
46+
s.assertNoErrors();
47+
s.assertTerminalEvent();
48+
s.assertUnsubscribed();
49+
}
50+
51+
@Test
52+
public void testUnsubscribesFromUpstream() {
53+
final AtomicBoolean unsubscribed = new AtomicBoolean(false);
54+
Action0 unsubscribeAction = new Action0() {
55+
@Override
56+
public void call() {
57+
unsubscribed.set(true);
58+
}
59+
};
60+
Observable.just(1).doOnUnsubscribe(unsubscribeAction)
61+
.takeLast(1).subscribe();
62+
assertTrue(unsubscribed.get());
63+
}
64+
65+
@Test
66+
public void testLastWithBackpressure() {
67+
MySubscriber<Integer> s = new MySubscriber<Integer>(0);
68+
Observable.just(1).takeLast(1).subscribe(s);
69+
assertEquals(0, s.list.size());
70+
s.requestMore(1);
71+
assertEquals(1, s.list.size());
72+
}
73+
74+
private static class MySubscriber<T> extends Subscriber<T> {
75+
76+
private long initialRequest;
77+
78+
MySubscriber(long initialRequest) {
79+
this.initialRequest = initialRequest;
80+
}
81+
82+
final List<T> list = new ArrayList<T>();
83+
84+
public void requestMore(long n) {
85+
request(n);
86+
}
87+
88+
@Override
89+
public void onStart() {
90+
request(initialRequest);
91+
}
92+
93+
@Override
94+
public void onCompleted() {
95+
96+
}
97+
98+
@Override
99+
public void onError(Throwable e) {
100+
101+
}
102+
103+
@Override
104+
public void onNext(T t) {
105+
list.add(t);
106+
}
107+
108+
}
109+
110+
}

0 commit comments

Comments
 (0)