Add an option to post async messages

This functionality was made public in API 22 but same API has existed since API 16. This prevents locking the message queue for vsync events.

Author: ZacSweers

rxandroid/src/main/java/io/reactivex/android/schedulers/AndroidSchedulers.java

@@ -13,20 +13,21 @@
  */
 package io.reactivex.android.schedulers;
 
+import android.annotation.SuppressLint;
+import android.os.Build;
 import android.os.Handler;
 import android.os.Looper;
-
-import java.util.concurrent.Callable;
-
+import android.os.Message;
 import io.reactivex.Scheduler;
 import io.reactivex.android.plugins.RxAndroidPlugins;
+import java.util.concurrent.Callable;
 
 /** Android-specific Schedulers. */
 public final class AndroidSchedulers {
 
     private static final class MainHolder {
-
-        static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));
+        static final Scheduler DEFAULT
+            = new HandlerScheduler(new Handler(Looper.getMainLooper()), false);
     }
 
     private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
@@ -43,8 +44,32 @@ public static Scheduler mainThread() {
 
     /** A {@link Scheduler} which executes actions on {@code looper}. */
     public static Scheduler from(Looper looper) {
+        return from(looper, false);
+    }
+
+    /**
+     * A {@link Scheduler} which executes actions on {@code looper}.
+     *
+     * @param async if true, the scheduler will use async messaging on API >= 16 to avoid VSYNC
+     *              locking. On API < 16 this value is ignored.
+     * @see Message#setAsynchronous(boolean)
+     */
+    @SuppressLint("NewApi") // Checking for an @hide API.
+    public static Scheduler from(Looper looper, boolean async) {
         if (looper == null) throw new NullPointerException("looper == null");
-        return new HandlerScheduler(new Handler(looper));
+        if (Build.VERSION.SDK_INT < 16) {
+            async = false;
+        } else if (async && Build.VERSION.SDK_INT < 22) {
+            // Confirm that the method is available on this API level despite being @hide.
+            Message message = Message.obtain();
+            try {
+                message.setAsynchronous(true);
+            } catch (NoSuchMethodError e) {
+                async = false;
+            }
+            message.recycle();
+        }
+        return new HandlerScheduler(new Handler(looper), async);
     }
 
     private AndroidSchedulers() {

rxandroid/src/main/java/io/reactivex/android/schedulers/HandlerScheduler.java

@@ -13,6 +13,7 @@
  */
 package io.reactivex.android.schedulers;
 
+import android.annotation.SuppressLint;
 import android.os.Handler;
 import android.os.Message;
 import io.reactivex.Scheduler;
@@ -23,9 +24,11 @@
 
 final class HandlerScheduler extends Scheduler {
     private final Handler handler;
+    private final boolean async;
 
-    HandlerScheduler(Handler handler) {
+    HandlerScheduler(Handler handler, boolean async) {
         this.handler = handler;
+        this.async = async;
     }
 
     @Override
@@ -41,19 +44,22 @@ public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
 
     @Override
     public Worker createWorker() {
-        return new HandlerWorker(handler);
+        return new HandlerWorker(handler, async);
     }
 
     private static final class HandlerWorker extends Worker {
         private final Handler handler;
+        private final boolean async;
 
         private volatile boolean disposed;
 
-        HandlerWorker(Handler handler) {
+        HandlerWorker(Handler handler, boolean async) {
             this.handler = handler;
+            this.async = async;
         }
 
         @Override
+        @SuppressLint("NewApi") // Async will only be true when the API is available to call.
         public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
             if (run == null) throw new NullPointerException("run == null");
             if (unit == null) throw new NullPointerException("unit == null");
@@ -69,6 +75,10 @@ public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
             Message message = Message.obtain(handler, scheduled);
             message.obj = this; // Used as token for batch disposal of this worker's runnables.
 
+            if (async) {
+                message.setAsynchronous(true);
+            }
+
             handler.sendMessageDelayed(message, unit.toMillis(delay));
 
             // Re-check disposed state for removing in case we were racing a call to dispose().

rxandroid/src/test/java/io/reactivex/android/schedulers/AndroidSchedulersTest.java

@@ -13,7 +13,9 @@
  */
 package io.reactivex.android.schedulers;
 
+import android.os.Build;
 import android.os.Looper;
+import android.os.Message;
 import io.reactivex.Scheduler;
 import io.reactivex.android.plugins.RxAndroidPlugins;
 import io.reactivex.android.testutil.EmptyScheduler;
@@ -25,11 +27,16 @@
 import org.junit.runner.RunWith;
 import org.robolectric.RobolectricTestRunner;
 import org.robolectric.annotation.Config;
+import org.robolectric.shadows.ShadowLooper;
+import org.robolectric.shadows.ShadowMessageQueue;
+import org.robolectric.util.ReflectionHelpers;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.fail;
+import static org.robolectric.Shadows.shadowOf;
 
 @RunWith(RobolectricTestRunner.class)
 @Config(manifest=Config.NONE)
@@ -68,8 +75,36 @@ public void fromNullThrows() {
         }
     }
 
+    @Test
+    public void fromNullThrowsTwoArg() {
+        try {
+            AndroidSchedulers.from(null, false);
+            fail();
+        } catch (NullPointerException e) {
+            assertEquals("looper == null", e.getMessage());
+        }
+    }
+
     @Test
     public void fromReturnsUsableScheduler() {
         assertNotNull(AndroidSchedulers.from(Looper.getMainLooper()));
     }
+
+    @Test
+    public void asyncIgnoredPre16() {
+        ReflectionHelpers.setStaticField(Build.VERSION.class, "SDK_INT", 14);
+
+        ShadowLooper mainLooper = ShadowLooper.getShadowMainLooper();
+        mainLooper.pause();
+        ShadowMessageQueue mainMessageQueue = shadowOf(Looper.getMainLooper().getQueue());
+
+        Scheduler main = AndroidSchedulers.from(Looper.getMainLooper(), true);
+        main.scheduleDirect(new Runnable() {
+            @Override public void run() {
+            }
+        });
+
+        Message message = mainMessageQueue.getHead();
+        assertFalse(message.isAsynchronous());
+    }
 }

rxandroid/src/test/java/io/reactivex/android/schedulers/HandlerSchedulerTest.java

@@ -22,14 +22,16 @@
 import io.reactivex.functions.Consumer;
 import io.reactivex.functions.Function;
 import io.reactivex.plugins.RxJavaPlugins;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.robolectric.RobolectricTestRunner;
+import org.robolectric.ParameterizedRobolectricTestRunner;
 import org.robolectric.annotation.Config;
 import org.robolectric.shadows.ShadowLooper;
 
@@ -46,9 +48,24 @@
 import static org.robolectric.shadows.ShadowLooper.runUiThreadTasksIncludingDelayedTasks;
 import static org.robolectric.shadows.ShadowLooper.unPauseMainLooper;
 
-@RunWith(RobolectricTestRunner.class)
-@Config(manifest=Config.NONE)
+@RunWith(ParameterizedRobolectricTestRunner.class)
+@Config(manifest=Config.NONE, sdk = 16)
 public final class HandlerSchedulerTest {
+
+    @ParameterizedRobolectricTestRunner.Parameters(name = "async = {0}")
+    public static Collection<Object[]> data() {
+        return Arrays.asList(new Object[][]{
+            {true},
+            {false}
+        });
+    }
+
+    private Scheduler scheduler;
+
+    public HandlerSchedulerTest(boolean async) {
+        this.scheduler = new HandlerScheduler(new Handler(Looper.getMainLooper()), async);
+    }
+
     @Before
     public void setUp() {
         RxJavaPlugins.reset();
@@ -61,8 +78,6 @@ public void tearDown() {
         unPauseMainLooper();
     }
 
-    private Scheduler scheduler = new HandlerScheduler(new Handler(Looper.getMainLooper()));
-
     @Test
     public void directScheduleOncePostsImmediately() {
         CountingRunnable counter = new CountingRunnable();