IOSSchedulers for RoboVM

rxjava-contrib/rxjava-ios/build.gradle

@@ -0,0 +1,39 @@
+apply plugin: 'osgi'
+
+dependencies {
+    compile project(':rxjava-core')
+
+    // testing
+    provided 'junit:junit-dep:4.10'
+    compile 'org.robovm:robovm-rt:0.0.12'
+    compile 'org.robovm:robovm-objc:0.0.12'
+    compile 'org.robovm:robovm-cocoatouch:0.0.12'
+}
+
+javadoc {
+    options {
+        doclet = "org.benjchristensen.doclet.DocletExclude"
+        docletpath = [rootProject.file('./gradle/doclet-exclude.jar')]
+        stylesheetFile = rootProject.file('./gradle/javadocStyleSheet.css')
+        windowTitle = "RxJava iOS Javadoc ${project.version}"
+    }
+    options.addStringOption('top').value = '<h2 class="title" style="padding-top:40px">RxJava iOS</h2>'
+}
+
+jar {
+    manifest {
+        name = 'rxjava-ios'
+        instruction 'Bundle-Vendor', 'Netflix'
+        instruction 'Bundle-DocURL', 'https://github.com/Netflix/RxJava'
+        instruction 'Import-Package', '!org.junit,!junit.framework,!org.mockito.*,*'
+        instruction 'Fragment-Host', 'com.netflix.rxjava.core'
+    }
+}
+
+test {
+    testLogging {
+        exceptionFormat "full"
+        events "started"
+        displayGranularity 2
+    }
+}

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

@@ -0,0 +1,122 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * Copyright 2014 Ashley Williams
+ *
+ * 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 rx.ios.schedulers;
+
+import org.robovm.apple.foundation.NSOperationQueue;
+import rx.Scheduler;
+import rx.Subscription;
+import rx.functions.Action0;
+import rx.internal.util.RxThreadFactory;
+import rx.subscriptions.CompositeSubscription;
+import rx.subscriptions.Subscriptions;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Schedules actions to run on an iOS Handler thread.
+ */
+public class HandlerThreadScheduler extends Scheduler {
+
+    private final NSOperationQueue operationQueue;
+    private static final String THREAD_PREFIX = "RxiOSScheduledExecutorPool-";
+
+
+    public HandlerThreadScheduler(NSOperationQueue operationQueue) {
+        this.operationQueue = operationQueue;
+    }
+
+    @Override
+    public Worker createWorker() {
+        return new InnerHandlerThreadScheduler(operationQueue);
+    }
+
+
+    private static class InnerHandlerThreadScheduler extends Worker {
+
+        private final NSOperationQueue operationQueue;
+        private CompositeSubscription innerSubscription = new CompositeSubscription();
+
+
+        public InnerHandlerThreadScheduler(NSOperationQueue operationQueue) {
+            this.operationQueue = operationQueue;
+        }
+
+        @Override
+        public void unsubscribe() {
+            innerSubscription.unsubscribe();
+        }
+
+        @Override
+        public boolean isUnsubscribed() {
+            return innerSubscription.isUnsubscribed();
+        }
+
+        @Override
+        public Subscription schedule(final Action0 action) {
+            return schedule(action, 0, null);
+        }
+
+        @Override
+        public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
+            return scheduledAction(action, delayTime, unit);
+        }
+
+        public Subscription scheduledAction(final Action0 action, long delay, TimeUnit unit) {
+
+            if (innerSubscription.isUnsubscribed()) {
+                return Subscriptions.empty();
+            }
+
+            final ScheduledIOSAction scheduledAction = new ScheduledIOSAction(action, operationQueue);
+            final ScheduledExecutorService executor = IOSScheduledExecutorPool.getInstance();
+
+            Future<?> future;
+            if (delay <= 0) {
+                future = executor.submit(scheduledAction);
+            } else {
+                future = executor.schedule(scheduledAction, delay, unit);
+            }
+
+            scheduledAction.add(Subscriptions.from(future));
+            scheduledAction.addParent(innerSubscription);
+
+            return scheduledAction;
+        }
+    }
+
+
+    private static final class IOSScheduledExecutorPool {
+
+        private static final RxThreadFactory THREAD_FACTORY = new RxThreadFactory(THREAD_PREFIX);
+
+        private static IOSScheduledExecutorPool INSTANCE = new IOSScheduledExecutorPool();
+        private final ScheduledExecutorService executorService;
+
+        private IOSScheduledExecutorPool() {
+            executorService = Executors.newScheduledThreadPool(1, THREAD_FACTORY);
+        }
+
+        public static ScheduledExecutorService getInstance() {
+            return INSTANCE.executorService;
+        }
+    }
+
+}

rxjava-contrib/rxjava-ios/src/main/java/rx/ios/schedulers/IOSSchedulers.java

@@ -0,0 +1,40 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ * Copyright 2014 Ashley Williams
+ *
+ * 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 rx.ios.schedulers;
+
+import rx.Scheduler;
+import org.robovm.apple.foundation.NSOperationQueue;
+
+
+public class IOSSchedulers {
+
+    private static final Scheduler MAIN_THREAD_SCHEDULER =
+            new HandlerThreadScheduler((NSOperationQueue) NSOperationQueue.getMainQueue());
+
+    private IOSSchedulers(){
+
+    }
+
+
+    public static Scheduler handlerThread(final NSOperationQueue operationQueue) {
+        return new HandlerThreadScheduler(operationQueue);
+    }
+
+    public static Scheduler mainThread() {
+        return MAIN_THREAD_SCHEDULER;
+    }
+}

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

@@ -0,0 +1,130 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * Copyright 2014 Ashley Williams
+ *
+ * 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 rx.ios.schedulers;
+
+import org.robovm.apple.foundation.NSBlockOperation;
+import org.robovm.apple.foundation.NSOperationQueue;
+import rx.Subscription;
+import rx.functions.Action0;
+import rx.subscriptions.CompositeSubscription;
+
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
+/**
+ * Based on {@code ScheduledAction} - A {@code Runnable} that executes an {@code Action0}
+ * that can be cancelled.
+ */
+final class ScheduledIOSAction implements Runnable, Subscription {
+    final CompositeSubscription cancel;
+    final Action0 action;
+    NSBlockOperation nsBlockOperation;
+    final NSOperationQueue operationQueue;
+    volatile int once;
+    static final AtomicIntegerFieldUpdater<ScheduledIOSAction> ONCE_UPDATER
+            = AtomicIntegerFieldUpdater.newUpdater(ScheduledIOSAction.class, "once");
+
+    public ScheduledIOSAction(Action0 action, NSOperationQueue operationQueue) {
+        this.action = action;
+        this.operationQueue = operationQueue;
+        this.cancel = new CompositeSubscription();
+
+        nsBlockOperation = new NSBlockOperation();
+    }
+
+    @Override
+    public void run() {
+        try {
+
+            final Runnable actionRunner = new Runnable() {
+                @Override
+                public void run() {
+                    action.call();
+                }
+            };
+
+            nsBlockOperation.addExecutionBlock$(actionRunner);
+
+            /* Add operation to operation queue*/
+            operationQueue.addOperation(nsBlockOperation);
+
+        } finally {
+            unsubscribe();
+        }
+    }
+
+    @Override
+    public boolean isUnsubscribed() {
+        return cancel.isUnsubscribed();
+    }
+
+    @Override
+    public void unsubscribe() {
+        if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
+            nsBlockOperation.cancel();
+            cancel.unsubscribe();
+            System.err.println("cancelled");
+        }
+    }
+
+    /**
+     * Adds a {@code Subscription} to the {@link CompositeSubscription} to be later cancelled on unsubscribe
+     *
+     * @param s subscription to add
+     */
+    public void add(Subscription s) {
+        cancel.add(s);
+    }
+
+    /**
+     * Adds a parent {@link rx.subscriptions.CompositeSubscription} to this {@code ScheduledIOSAction} so when
+     * the action is cancelled or terminates, it can remove itself from this parent
+     * @param parent the parent {@code CompositeSubscription} to add
+     */
+    public void addParent(CompositeSubscription parent) {
+        cancel.add(new Remover(this, parent));
+    }
+
+
+    /**
+     * Remove a child subscription from a composite when unsubscribing
+     */
+    private static final class Remover implements Subscription {
+        final Subscription s;
+        final CompositeSubscription parent;
+        volatile int once;
+        static final AtomicIntegerFieldUpdater<Remover> ONCE_UPDATER
+                = AtomicIntegerFieldUpdater.newUpdater(Remover.class, "once");
+
+        public Remover(Subscription s, CompositeSubscription parent) {
+            this.s = s;
+            this.parent = parent;
+        }
+
+        @Override
+        public boolean isUnsubscribed() {
+            return s.isUnsubscribed();
+        }
+
+        @Override
+        public void unsubscribe() {
+            if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
+                parent.remove(s);
+            }
+        }
+    }
+}

settings.gradle

@@ -7,6 +7,7 @@ include 'rxjava-core', \
 'language-adaptors:rxjava-kotlin', \
 'rxjava-contrib:rxjava-swing', \
 //'rxjava-contrib:rxjava-javafx', \
+'rxjava-contrib:rxjava-ios', \
 'rxjava-contrib:rxjava-android', \
 'rxjava-contrib:rxjava-android-samples-build-wrapper', \
 'rxjava-contrib:rxjava-apache-http', \