Skip to content

Commit 3bac5b9

Browse files
benjchristensenbenjchristensen
committed
Merge pull request #797 from benjchristensen/scheduler-inner-outer
Scheduler Outer/Inner [Preview]
2 parents b78983f + 4bdf08d commit 3bac5b9

File tree

63 files changed

+1260
-1824
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

63 files changed

+1260
-1824
lines changed

‎rxjava-contrib/rxjava-android/src/main/java/rx/android/schedulers/HandlerThreadScheduler.java

Lines changed: 70 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,12 @@
11
/**
22
* Copyright 2013 Netflix, Inc.
3-
*
3+
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
66
* You may obtain a copy of the License at
7-
*
8-
* http://www.apache.org/licenses/LICENSE-2.0
9-
*
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
1010
* Unless required by applicable law or agreed to in writing, software
1111
* distributed under the License is distributed on an "AS IS" BASIS,
1212
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -19,8 +19,8 @@
1919

2020
import rx.Scheduler;
2121
import rx.Subscription;
22-
import rx.operators.SafeObservableSubscription;
23-
import rx.util.functions.Func2;
22+
import rx.subscriptions.BooleanSubscription;
23+
import rx.util.functions.Action1;
2424
import android.os.Handler;
2525

2626
/**
@@ -32,25 +32,29 @@ public class HandlerThreadScheduler extends Scheduler {
3232

3333
/**
3434
* Constructs a {@link HandlerThreadScheduler} using the given {@link Handler}
35-
* @param handler {@link Handler} to use when scheduling actions
35+
*
36+
* @param handler
37+
* {@link Handler} to use when scheduling actions
3638
*/
3739
public HandlerThreadScheduler(Handler handler) {
3840
this.handler = handler;
3941
}
4042

4143
/**
42-
* Calls {@link HandlerThreadScheduler#schedule(Object, rx.util.functions.Func2, long, java.util.concurrent.TimeUnit)}
43-
* with a delay of zero milliseconds.
44-
*
44+
* Calls {@link HandlerThreadScheduler#schedule(Object, rx.util.functions.Func2, long, java.util.concurrent.TimeUnit)} with a delay of zero milliseconds.
45+
*
4546
* See {@link #schedule(Object, rx.util.functions.Func2, long, java.util.concurrent.TimeUnit)}
4647
*/
4748
@Override
48-
public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action) {
49-
return schedule(state, action, 0L, TimeUnit.MILLISECONDS);
49+
public Subscription schedule(Action1<Inner> action) {
50+
InnerHandlerThreadScheduler inner = new InnerHandlerThreadScheduler(handler);
51+
inner.schedule(action);
52+
return inner;
5053
}
5154

5255
/**
5356
* Calls {@link Handler#postDelayed(Runnable, long)} with a runnable that executes the given action.
57+
*
5458
* @param state
5559
* State to pass into the action.
5660
* @param action
@@ -62,17 +66,60 @@ public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ?
6266
* @return A Subscription from which one can unsubscribe from.
6367
*/
6468
@Override
65-
public <T> Subscription schedule(final T state, final Func2<? super Scheduler, ? super T, ? extends Subscription> action, long delayTime, TimeUnit unit) {
66-
final SafeObservableSubscription subscription = new SafeObservableSubscription();
67-
final Scheduler _scheduler = this;
68-
handler.postDelayed(new Runnable() {
69-
@Override
70-
public void run() {
71-
subscription.wrap(action.call(_scheduler, state));
72-
}
73-
}, unit.toMillis(delayTime));
74-
return subscription;
69+
public Subscription schedule(Action1<Inner> action, long delayTime, TimeUnit unit) {
70+
InnerHandlerThreadScheduler inner = new InnerHandlerThreadScheduler(handler);
71+
inner.schedule(action, delayTime, unit);
72+
return inner;
7573
}
76-
}
7774

75+
private static class InnerHandlerThreadScheduler extends Inner {
76+
77+
private final Handler handler;
78+
private BooleanSubscription innerSubscription = new BooleanSubscription();
79+
private Inner _inner = this;
80+
81+
public InnerHandlerThreadScheduler(Handler handler) {
82+
this.handler = handler;
83+
}
84+
85+
@Override
86+
public void unsubscribe() {
87+
innerSubscription.unsubscribe();
88+
}
89+
90+
@Override
91+
public boolean isUnsubscribed() {
92+
return innerSubscription.isUnsubscribed();
93+
}
94+
95+
@Override
96+
public void schedule(final Action1<Inner> action, long delayTime, TimeUnit unit) {
97+
handler.postDelayed(new Runnable() {
98+
@Override
99+
public void run() {
100+
if (_inner.isUnsubscribed()) {
101+
return;
102+
}
103+
action.call(_inner);
104+
}
105+
}, unit.toMillis(delayTime));
106+
}
78107

108+
@Override
109+
public void schedule(final Action1<Inner> action) {
110+
handler.postDelayed(new Runnable() {
111+
112+
@Override
113+
public void run() {
114+
if (_inner.isUnsubscribed()) {
115+
return;
116+
}
117+
action.call(_inner);
118+
}
119+
120+
}, 0L);
121+
}
122+
123+
}
124+
125+
}

‎rxjava-contrib/rxjava-android/src/test/java/rx/android/schedulers/HandlerThreadSchedulerTest.java

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,9 @@
2727
import org.robolectric.annotation.Config;
2828

2929
import rx.Scheduler;
30+
import rx.Scheduler.Inner;
3031
import rx.Subscription;
32+
import rx.util.functions.Action1;
3133
import rx.util.functions.Func2;
3234
import android.os.Handler;
3335

@@ -38,38 +40,36 @@ public class HandlerThreadSchedulerTest {
3840
@Test
3941
public void shouldScheduleImmediateActionOnHandlerThread() {
4042
final Handler handler = mock(Handler.class);
41-
final Object state = new Object();
4243
@SuppressWarnings("unchecked")
43-
final Func2<Scheduler, Object, Subscription> action = mock(Func2.class);
44+
final Action1<Inner> action = mock(Action1.class);
4445

4546
Scheduler scheduler = new HandlerThreadScheduler(handler);
46-
scheduler.schedule(state, action);
47+
scheduler.schedule(action);
4748

4849
// verify that we post to the given Handler
4950
ArgumentCaptor<Runnable> runnable = ArgumentCaptor.forClass(Runnable.class);
5051
verify(handler).postDelayed(runnable.capture(), eq(0L));
5152

5253
// verify that the given handler delegates to our action
5354
runnable.getValue().run();
54-
verify(action).call(scheduler, state);
55+
verify(action).call(any(Inner.class));
5556
}
5657

5758
@Test
5859
public void shouldScheduleDelayedActionOnHandlerThread() {
5960
final Handler handler = mock(Handler.class);
60-
final Object state = new Object();
6161
@SuppressWarnings("unchecked")
62-
final Func2<Scheduler, Object, Subscription> action = mock(Func2.class);
62+
final Action1<Inner> action = mock(Action1.class);
6363

6464
Scheduler scheduler = new HandlerThreadScheduler(handler);
65-
scheduler.schedule(state, action, 1L, TimeUnit.SECONDS);
65+
scheduler.schedule(action, 1L, TimeUnit.SECONDS);
6666

6767
// verify that we post to the given Handler
6868
ArgumentCaptor<Runnable> runnable = ArgumentCaptor.forClass(Runnable.class);
6969
verify(handler).postDelayed(runnable.capture(), eq(1000L));
7070

7171
// verify that the given handler delegates to our action
7272
runnable.getValue().run();
73-
verify(action).call(scheduler, state);
73+
verify(action).call(any(Inner.class));
7474
}
7575
}

‎rxjava-contrib/rxjava-async-util/src/main/java/rx/util/async/Async.java

Lines changed: 23 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
import rx.Observable;
2323
import rx.Scheduler;
24+
import rx.Scheduler.Inner;
2425
import rx.schedulers.Schedulers;
2526
import rx.subjects.AsyncSubject;
2627
import rx.util.async.operators.Functionals;
@@ -592,9 +593,9 @@ public static <R> Func0<Observable<R>> toAsync(final Func0<? extends R> func, fi
592593
@Override
593594
public Observable<R> call() {
594595
final AsyncSubject<R> subject = AsyncSubject.create();
595-
scheduler.schedule(new Action0() {
596+
scheduler.schedule(new Action1<Inner>() {
596597
@Override
597-
public void call() {
598+
public void call(Inner inner) {
598599
R result;
599600
try {
600601
result = func.call();
@@ -649,9 +650,9 @@ public static <T1, R> Func1<T1, Observable<R>> toAsync(final Func1<? super T1, ?
649650
@Override
650651
public Observable<R> call(final T1 t1) {
651652
final AsyncSubject<R> subject = AsyncSubject.create();
652-
scheduler.schedule(new Action0() {
653+
scheduler.schedule(new Action1<Inner>() {
653654
@Override
654-
public void call() {
655+
public void call(Inner inner) {
655656
R result;
656657
try {
657658
result = func.call(t1);
@@ -708,9 +709,9 @@ public static <T1, T2, R> Func2<T1, T2, Observable<R>> toAsync(final Func2<? sup
708709
@Override
709710
public Observable<R> call(final T1 t1, final T2 t2) {
710711
final AsyncSubject<R> subject = AsyncSubject.create();
711-
scheduler.schedule(new Action0() {
712+
scheduler.schedule(new Action1<Inner>() {
712713
@Override
713-
public void call() {
714+
public void call(Inner inner) {
714715
R result;
715716
try {
716717
result = func.call(t1, t2);
@@ -769,9 +770,9 @@ public static <T1, T2, T3, R> Func3<T1, T2, T3, Observable<R>> toAsync(final Fun
769770
@Override
770771
public Observable<R> call(final T1 t1, final T2 t2, final T3 t3) {
771772
final AsyncSubject<R> subject = AsyncSubject.create();
772-
scheduler.schedule(new Action0() {
773+
scheduler.schedule(new Action1<Inner>() {
773774
@Override
774-
public void call() {
775+
public void call(Inner inner) {
775776
R result;
776777
try {
777778
result = func.call(t1, t2, t3);
@@ -832,9 +833,9 @@ public static <T1, T2, T3, T4, R> Func4<T1, T2, T3, T4, Observable<R>> toAsync(f
832833
@Override
833834
public Observable<R> call(final T1 t1, final T2 t2, final T3 t3, final T4 t4) {
834835
final AsyncSubject<R> subject = AsyncSubject.create();
835-
scheduler.schedule(new Action0() {
836+
scheduler.schedule(new Action1<Inner>() {
836837
@Override
837-
public void call() {
838+
public void call(Inner inner) {
838839
R result;
839840
try {
840841
result = func.call(t1, t2, t3, t4);
@@ -897,9 +898,9 @@ public static <T1, T2, T3, T4, T5, R> Func5<T1, T2, T3, T4, T5, Observable<R>> t
897898
@Override
898899
public Observable<R> call(final T1 t1, final T2 t2, final T3 t3, final T4 t4, final T5 t5) {
899900
final AsyncSubject<R> subject = AsyncSubject.create();
900-
scheduler.schedule(new Action0() {
901+
scheduler.schedule(new Action1<Inner>() {
901902
@Override
902-
public void call() {
903+
public void call(Inner inner) {
903904
R result;
904905
try {
905906
result = func.call(t1, t2, t3, t4, t5);
@@ -964,9 +965,9 @@ public static <T1, T2, T3, T4, T5, T6, R> Func6<T1, T2, T3, T4, T5, T6, Observab
964965
@Override
965966
public Observable<R> call(final T1 t1, final T2 t2, final T3 t3, final T4 t4, final T5 t5, final T6 t6) {
966967
final AsyncSubject<R> subject = AsyncSubject.create();
967-
scheduler.schedule(new Action0() {
968+
scheduler.schedule(new Action1<Inner>() {
968969
@Override
969-
public void call() {
970+
public void call(Inner inner) {
970971
R result;
971972
try {
972973
result = func.call(t1, t2, t3, t4, t5, t6);
@@ -1033,9 +1034,9 @@ public static <T1, T2, T3, T4, T5, T6, T7, R> Func7<T1, T2, T3, T4, T5, T6, T7,
10331034
@Override
10341035
public Observable<R> call(final T1 t1, final T2 t2, final T3 t3, final T4 t4, final T5 t5, final T6 t6, final T7 t7) {
10351036
final AsyncSubject<R> subject = AsyncSubject.create();
1036-
scheduler.schedule(new Action0() {
1037+
scheduler.schedule(new Action1<Inner>() {
10371038
@Override
1038-
public void call() {
1039+
public void call(Inner inner) {
10391040
R result;
10401041
try {
10411042
result = func.call(t1, t2, t3, t4, t5, t6, t7);
@@ -1104,9 +1105,9 @@ public static <T1, T2, T3, T4, T5, T6, T7, T8, R> Func8<T1, T2, T3, T4, T5, T6,
11041105
@Override
11051106
public Observable<R> call(final T1 t1, final T2 t2, final T3 t3, final T4 t4, final T5 t5, final T6 t6, final T7 t7, final T8 t8) {
11061107
final AsyncSubject<R> subject = AsyncSubject.create();
1107-
scheduler.schedule(new Action0() {
1108+
scheduler.schedule(new Action1<Inner>() {
11081109
@Override
1109-
public void call() {
1110+
public void call(Inner inner) {
11101111
R result;
11111112
try {
11121113
result = func.call(t1, t2, t3, t4, t5, t6, t7, t8);
@@ -1177,9 +1178,9 @@ public static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Func9<T1, T2, T3, T4, T5,
11771178
@Override
11781179
public Observable<R> call(final T1 t1, final T2 t2, final T3 t3, final T4 t4, final T5 t5, final T6 t6, final T7 t7, final T8 t8, final T9 t9) {
11791180
final AsyncSubject<R> subject = AsyncSubject.create();
1180-
scheduler.schedule(new Action0() {
1181+
scheduler.schedule(new Action1<Inner>() {
11811182
@Override
1182-
public void call() {
1183+
public void call(Inner inner) {
11831184
R result;
11841185
try {
11851186
result = func.call(t1, t2, t3, t4, t5, t6, t7, t8, t9);
@@ -1230,9 +1231,9 @@ public static <R> FuncN<Observable<R>> toAsync(final FuncN<? extends R> func, fi
12301231
@Override
12311232
public Observable<R> call(final Object... args) {
12321233
final AsyncSubject<R> subject = AsyncSubject.create();
1233-
scheduler.schedule(new Action0() {
1234+
scheduler.schedule(new Action1<Inner>() {
12341235
@Override
1235-
public void call() {
1236+
public void call(Inner inner) {
12361237
R result;
12371238
try {
12381239
result = func.call(args);

‎rxjava-contrib/rxjava-async-util/src/main/java/rx/util/async/operators/Functionals.java

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
package rx.util.async.operators;
1818

19+
import rx.Scheduler.Inner;
1920
import rx.util.functions.Action0;
2021
import rx.util.functions.Action1;
2122

@@ -65,22 +66,22 @@ public void call() {
6566
* @param run the Runnable to run when the Action0 is called
6667
* @return the Action0 wrapping the Runnable
6768
*/
68-
public static Action0 fromRunnable(Runnable run) {
69+
public static Action1<Inner> fromRunnable(Runnable run) {
6970
if (run == null) {
7071
throw new NullPointerException("run");
7172
}
7273
return new ActionWrappingRunnable(run);
7374
}
74-
/** An Action0 which wraps and calls a Runnable. */
75-
private static final class ActionWrappingRunnable implements Action0 {
75+
/** An Action1 which wraps and calls a Runnable. */
76+
private static final class ActionWrappingRunnable implements Action1<Inner> {
7677
final Runnable run;
7778

7879
public ActionWrappingRunnable(Runnable run) {
7980
this.run = run;
8081
}
8182

8283
@Override
83-
public void call() {
84+
public void call(Inner inner) {
8485
run.run();
8586
}
8687

0 commit comments

Comments
 (0)