Skip to content

Commit c46153c

Browse files
ashleyjashleyj
committed
As per suggestions:
Added single static instance of ExecutorService for delayed posting Introduced ScheduledIOSAction to enable CompositeSubscription
1 parent 5dbb81f commit c46153c

File tree

2 files changed

+179
-36
lines changed

2 files changed

+179
-36
lines changed

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

Lines changed: 49 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,3 @@
1-
package rx.ios.schedulers;
21
/**
32
* Copyright 2013 Netflix, Inc.
43
* Copyright 2014 Ashley Williams
@@ -16,16 +15,18 @@
1615
* limitations under the License.
1716
*/
1817

18+
package rx.ios.schedulers;
1919

20-
import org.robovm.apple.foundation.NSBlockOperation;
2120
import org.robovm.apple.foundation.NSOperationQueue;
2221
import rx.Scheduler;
2322
import rx.Subscription;
2423
import rx.functions.Action0;
25-
import rx.subscriptions.BooleanSubscription;
24+
import rx.internal.util.RxThreadFactory;
25+
import rx.subscriptions.CompositeSubscription;
2626
import rx.subscriptions.Subscriptions;
2727

2828
import java.util.concurrent.Executors;
29+
import java.util.concurrent.Future;
2930
import java.util.concurrent.ScheduledExecutorService;
3031
import java.util.concurrent.TimeUnit;
3132

@@ -35,6 +36,8 @@
3536
public class HandlerThreadScheduler extends Scheduler {
3637

3738
private final NSOperationQueue operationQueue;
39+
private static final String THREAD_PREFIX = "RxiOSScheduledExecutorPool-";
40+
3841

3942
public HandlerThreadScheduler(NSOperationQueue operationQueue) {
4043
this.operationQueue = operationQueue;
@@ -49,7 +52,7 @@ public Worker createWorker() {
4952
private static class InnerHandlerThreadScheduler extends Worker {
5053

5154
private final NSOperationQueue operationQueue;
52-
private BooleanSubscription innerSubscription = new BooleanSubscription();
55+
private CompositeSubscription innerSubscription = new CompositeSubscription();
5356

5457

5558
public InnerHandlerThreadScheduler(NSOperationQueue operationQueue) {
@@ -67,43 +70,53 @@ public boolean isUnsubscribed() {
6770
}
6871

6972
@Override
70-
public Subscription schedule(Action0 action0) {
71-
return schedule(action0, 0, TimeUnit.MILLISECONDS);
73+
public Subscription schedule(final Action0 action) {
74+
return schedule(action, 0, null);
7275
}
7376

7477
@Override
7578
public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
79+
return scheduledAction(action, delayTime, unit);
80+
}
81+
82+
public Subscription scheduledAction(final Action0 action, long delay, TimeUnit unit) {
7683

77-
ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
78-
final NSBlockOperation runOperation = new NSBlockOperation();
79-
80-
executor.schedule(new Runnable() {
81-
@Override
82-
public void run() {
83-
if (isUnsubscribed()) {
84-
return;
85-
}
86-
/* Runnable for action */
87-
final Runnable actionRunner = new Runnable() {
88-
@Override
89-
public void run() {
90-
action.call();
91-
}
92-
};
93-
94-
runOperation.addExecutionBlock$(actionRunner);
95-
96-
/* Add operation to operation queue*/
97-
operationQueue.addOperation(runOperation);
98-
}
99-
}, delayTime, unit);
100-
101-
return Subscriptions.create(new Action0() {
102-
@Override
103-
public void call() {
104-
runOperation.cancel();
105-
}
106-
});
84+
if (innerSubscription.isUnsubscribed()) {
85+
return Subscriptions.empty();
86+
}
87+
88+
final ScheduledIOSAction scheduledAction = new ScheduledIOSAction(action, operationQueue);
89+
final ScheduledExecutorService executor = IOSScheduledExecutorPool.getInstance();
90+
91+
Future<?> future;
92+
if (delay <= 0) {
93+
future = executor.submit(scheduledAction);
94+
} else {
95+
future = executor.schedule(scheduledAction, delay, unit);
96+
}
97+
98+
scheduledAction.add(Subscriptions.from(future));
99+
scheduledAction.addParent(innerSubscription);
100+
101+
return scheduledAction;
107102
}
108103
}
104+
105+
106+
private static final class IOSScheduledExecutorPool {
107+
108+
private static final RxThreadFactory THREAD_FACTORY = new RxThreadFactory(THREAD_PREFIX);
109+
110+
private static IOSScheduledExecutorPool INSTANCE = new IOSScheduledExecutorPool();
111+
private final ScheduledExecutorService executorService;
112+
113+
private IOSScheduledExecutorPool() {
114+
executorService = Executors.newScheduledThreadPool(1, THREAD_FACTORY);
115+
}
116+
117+
public static ScheduledExecutorService getInstance() {
118+
return INSTANCE.executorService;
119+
}
120+
}
121+
109122
}

‎rxjava-contrib/rxjava-ios/src/main/java/rx/ios/schedulers/ScheduledIOSAction.java‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/**
2+
* Copyright 2014 Netflix, Inc.
3+
* Copyright 2014 Ashley Williams
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package rx.ios.schedulers;
19+
20+
import org.robovm.apple.foundation.NSBlockOperation;
21+
import org.robovm.apple.foundation.NSOperationQueue;
22+
import rx.Subscription;
23+
import rx.functions.Action0;
24+
import rx.subscriptions.CompositeSubscription;
25+
26+
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
27+
28+
/**
29+
* Based on {@code ScheduledAction} - A {@code Runnable} that executes an {@code Action0}
30+
* that can be cancelled.
31+
*/
32+
final class ScheduledIOSAction implements Runnable, Subscription {
33+
final CompositeSubscription cancel;
34+
final Action0 action;
35+
NSBlockOperation nsBlockOperation;
36+
final NSOperationQueue operationQueue;
37+
volatile int once;
38+
static final AtomicIntegerFieldUpdater<ScheduledIOSAction> ONCE_UPDATER
39+
= AtomicIntegerFieldUpdater.newUpdater(ScheduledIOSAction.class, "once");
40+
41+
public ScheduledIOSAction(Action0 action, NSOperationQueue operationQueue) {
42+
this.action = action;
43+
this.operationQueue = operationQueue;
44+
this.cancel = new CompositeSubscription();
45+
46+
nsBlockOperation = new NSBlockOperation();
47+
}
48+
49+
@Override
50+
public void run() {
51+
try {
52+
53+
final Runnable actionRunner = new Runnable() {
54+
@Override
55+
public void run() {
56+
action.call();
57+
}
58+
};
59+
60+
nsBlockOperation.addExecutionBlock$(actionRunner);
61+
62+
/* Add operation to operation queue*/
63+
operationQueue.addOperation(nsBlockOperation);
64+
65+
} finally {
66+
unsubscribe();
67+
}
68+
}
69+
70+
@Override
71+
public boolean isUnsubscribed() {
72+
return cancel.isUnsubscribed();
73+
}
74+
75+
@Override
76+
public void unsubscribe() {
77+
if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
78+
nsBlockOperation.cancel();
79+
cancel.unsubscribe();
80+
System.err.println("cancelled");
81+
}
82+
}
83+
84+
/**
85+
* Adds a {@code Subscription} to the {@link CompositeSubscription} to be later cancelled on unsubscribe
86+
*
87+
* @param s subscription to add
88+
*/
89+
public void add(Subscription s) {
90+
cancel.add(s);
91+
}
92+
93+
/**
94+
* Adds a parent {@link rx.subscriptions.CompositeSubscription} to this {@code ScheduledIOSAction} so when
95+
* the action is cancelled or terminates, it can remove itself from this parent
96+
* @param parent the parent {@code CompositeSubscription} to add
97+
*/
98+
public void addParent(CompositeSubscription parent) {
99+
cancel.add(new Remover(this, parent));
100+
}
101+
102+
103+
/**
104+
* Remove a child subscription from a composite when unsubscribing
105+
*/
106+
private static final class Remover implements Subscription {
107+
final Subscription s;
108+
final CompositeSubscription parent;
109+
volatile int once;
110+
static final AtomicIntegerFieldUpdater<Remover> ONCE_UPDATER
111+
= AtomicIntegerFieldUpdater.newUpdater(Remover.class, "once");
112+
113+
public Remover(Subscription s, CompositeSubscription parent) {
114+
this.s = s;
115+
this.parent = parent;
116+
}
117+
118+
@Override
119+
public boolean isUnsubscribed() {
120+
return s.isUnsubscribed();
121+
}
122+
123+
@Override
124+
public void unsubscribe() {
125+
if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
126+
parent.remove(s);
127+
}
128+
}
129+
}
130+
}

0 commit comments

Comments
 (0)