Initial support for scheduling on Android Handler threads

rxjava-contrib/rxjava-android/build.gradle

@@ -0,0 +1,59 @@
+apply plugin: 'java'
+apply plugin: 'eclipse'
+apply plugin: 'idea'
+apply plugin: 'osgi'
+
+sourceCompatibility = JavaVersion.VERSION_1_6
+targetCompatibility = JavaVersion.VERSION_1_6
+
+dependencies {
+ compile project(':rxjava-core')
+ provided 'junit:junit-dep:4.10'
+ provided 'org.mockito:mockito-core:1.8.5'
+ provided 'org.robolectric:robolectric:2.1.1'
+ provided 'com.google.android:android:4.0.1.2'
+}
+
+eclipse {
+ classpath {
+ // include 'provided' dependencies on the classpath
+ plusConfigurations += configurations.provided
+
+ downloadSources = true
+ downloadJavadoc = true
+ }
+}
+
+idea {
+	module {
+ // include 'provided' dependencies on the classpath
+ scopes.PROVIDED.plus += configurations.provided
+	}
+}
+
+javadoc {
+ options {
+ doclet = "org.benjchristensen.doclet.DocletExclude"
+ docletpath = [rootProject.file('./gradle/doclet-exclude.jar')]
+ stylesheetFile = rootProject.file('./gradle/javadocStyleSheet.css')
+ windowTitle = "RxJava Android Javadoc ${project.version}"
+ }
+ options.addStringOption('top').value = '<h2 class="title" style="padding-top:40px">RxJava Android</h2>'
+}
+
+jar {
+ manifest {
+ name = 'rxjava-android'
+ instruction 'Bundle-Vendor', 'Netflix'
+ instruction 'Bundle-DocURL', 'https://github.com/Netflix/RxJava'
+ instruction 'Import-Package', '!org.junit,!junit.framework,!org.mockito.*,*'
+ }
+}
+
+test {
+ testLogging {
+ exceptionFormat "full"
+ events "started"
+ displayGranularity 2
+ }
+}

rxjava-contrib/rxjava-android/src/main/java/rx/android/concurrency/AndroidSchedulers.java

@@ -0,0 +1,36 @@
+package rx.android.concurrency;
+
+import android.os.Handler;
+import android.os.Looper;
+import rx.Scheduler;
+
+/**
+ * Schedulers that have Android specific functionality
+ */
+public class AndroidSchedulers {
+
+ private static final Scheduler MAIN_THREAD_SCHEDULER =
+ new HandlerThreadScheduler(new Handler(Looper.getMainLooper()));
+
+ private AndroidSchedulers(){
+
+ }
+
+ /**
+ * {@link Scheduler} which uses the provided {@link Handler} to execute an action
+ * @param handler The handler that will be used when executing the action
+ * @return A handler based scheduler
+ */
+ public static Scheduler handlerThread(final Handler handler) {
+ return new HandlerThreadScheduler(handler);
+ }
+
+ /**
+ * {@link Scheduler} which will execute an action on the main Android UI thread.
+ *
+ * @return A Main {@link Looper} based scheduler
+ */
+ public static Scheduler mainThread() {
+ return MAIN_THREAD_SCHEDULER;
+ }
+}

rxjava-contrib/rxjava-android/src/main/java/rx/android/concurrency/HandlerThreadScheduler.java

@@ -0,0 +1,113 @@
+package rx.android.concurrency;
+
+import android.os.Handler;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.robolectric.RobolectricTestRunner;
+import org.robolectric.annotation.Config;
+import rx.Scheduler;
+import rx.Subscription;
+import rx.operators.AtomicObservableSubscription;
+import rx.util.functions.Func2;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Schedules actions to run on an Android Handler thread.
+ */
+public class HandlerThreadScheduler extends Scheduler {
+
+ private final Handler handler;
+
+ /**
+ * Constructs a {@link HandlerThreadScheduler} using the given {@link Handler}
+ * @param handler {@link Handler} to use when scheduling actions
+ */
+ public HandlerThreadScheduler(Handler handler) {
+ this.handler = handler;
+ }
+
+ /**
+ * Calls {@link HandlerThreadScheduler#schedule(Object, rx.util.functions.Func2, long, java.util.concurrent.TimeUnit)}
+ * with a delay of zero milliseconds.
+ *
+ * See {@link #schedule(Object, rx.util.functions.Func2, long, java.util.concurrent.TimeUnit)}
+ */
+ @Override
+ public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action) {
+ return schedule(state, action, 0L, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Calls {@link Handler#postDelayed(Runnable, long)} with a runnable that executes the given action.
+ * @param state
+ * State to pass into the action.
+ * @param action
+ * Action to schedule.
+ * @param delayTime
+ * Time the action is to be delayed before executing.
+ * @param unit
+ * Time unit of the delay time.
+ * @return A Subscription from which one can unsubscribe from.
+ */
+ @Override
+ public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit) {
+ final AtomicObservableSubscription subscription = new AtomicObservableSubscription();
+ final Scheduler _scheduler = this;
+ handler.postDelayed(new Runnable() {
+ @Override
+ public void run() {
+ subscription.wrap(action.call(_scheduler, state));
+ }
+ }, unit.toMillis(delayTime));
+ return subscription;
+ }
+
+ @RunWith(RobolectricTestRunner.class)
+ @Config(manifest=Config.NONE)
+ public static final class UnitTest {
+
+ @Test
+ public void shouldScheduleImmediateActionOnHandlerThread() {
+ final Handler handler = mock(Handler.class);
+ final Object state = new Object();
+ final Func2<Scheduler, Object, Subscription> action = mock(Func2.class);
+
+ Scheduler scheduler = new HandlerThreadScheduler(handler);
+ scheduler.schedule(state, action);
+
+ // verify that we post to the given Handler
+ ArgumentCaptor<Runnable> runnable = ArgumentCaptor.forClass(Runnable.class);
+ verify(handler).postDelayed(runnable.capture(), eq(0L));
+
+ // verify that the given handler delegates to our action
+ runnable.getValue().run();
+ verify(action).call(scheduler, state);
+ }
+
+ @Test
+ public void shouldScheduleDelayedActionOnHandlerThread() {
+ final Handler handler = mock(Handler.class);
+ final Object state = new Object();
+ final Func2<Scheduler, Object, Subscription> action = mock(Func2.class);
+
+ Scheduler scheduler = new HandlerThreadScheduler(handler);
+ scheduler.schedule(state, action, 1L, TimeUnit.SECONDS);
+
+ // verify that we post to the given Handler
+ ArgumentCaptor<Runnable> runnable = ArgumentCaptor.forClass(Runnable.class);
+ verify(handler).postDelayed(runnable.capture(), eq(1000L));
+
+ // verify that the given handler delegates to our action
+ runnable.getValue().run();
+ verify(action).call(scheduler, state);
+ }
+ }
+}
+
+

settings.gradle

@@ -4,4 +4,5 @@ include 'rxjava-core', \
 'language-adaptors:rxjava-jruby', \
 'language-adaptors:rxjava-clojure', \
 'language-adaptors:rxjava-scala', \
-'rxjava-contrib:rxjava-swing'
+'rxjava-contrib:rxjava-swing', \
+'rxjava-contrib:rxjava-android'