considers async option in HandlerScheduler.scheduleDirect

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

@@ -32,13 +32,18 @@ final class HandlerScheduler extends Scheduler {
     }
 
     @Override
+    @SuppressLint("NewApi") // Async will only be true when the API is available to call.
     public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
         if (run == null) throw new NullPointerException("run == null");
         if (unit == null) throw new NullPointerException("unit == null");
 
         run = RxJavaPlugins.onSchedule(run);
         ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
-        handler.postDelayed(scheduled, unit.toMillis(delay));
+        Message message = Message.obtain(handler, scheduled);
+        if (async) {
+            message.setAsynchronous(true);
+        }
+        handler.sendMessageDelayed(message, unit.toMillis(delay));
         return scheduled;
     }
 

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

@@ -15,6 +15,7 @@
 
 import android.os.Handler;
 import android.os.Looper;
+import android.os.Message;
 import io.reactivex.Scheduler;
 import io.reactivex.Scheduler.Worker;
 import io.reactivex.android.testutil.CountingRunnable;
@@ -34,6 +35,7 @@
 import org.robolectric.ParameterizedRobolectricTestRunner;
 import org.robolectric.annotation.Config;
 import org.robolectric.shadows.ShadowLooper;
+import org.robolectric.shadows.ShadowMessageQueue;
 
 import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.concurrent.TimeUnit.SECONDS;
@@ -43,6 +45,7 @@
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.robolectric.Shadows.shadowOf;
 import static org.robolectric.shadows.ShadowLooper.pauseMainLooper;
 import static org.robolectric.shadows.ShadowLooper.runUiThreadTasks;
 import static org.robolectric.shadows.ShadowLooper.runUiThreadTasksIncludingDelayedTasks;
@@ -61,9 +64,11 @@ public static Collection<Object[]> data() {
     }
 
     private Scheduler scheduler;
+    private boolean async;
 
     public HandlerSchedulerTest(boolean async) {
         this.scheduler = new HandlerScheduler(new Handler(Looper.getMainLooper()), async);
+        this.async = async;
     }
 
     @Before
@@ -774,6 +779,47 @@ public void workerSchedulePeriodicallyInputValidation() {
         }
     }
 
+    @Test
+    public void directScheduleSetAsync() {
+        ShadowMessageQueue mainMessageQueue = shadowOf(Looper.getMainLooper().getQueue());
+
+        scheduler.scheduleDirect(new Runnable() {
+            @Override public void run() {
+            }
+        });
+
+        Message message = mainMessageQueue.getHead();
+        assertEquals(async, message.isAsynchronous());
+    }
+
+    @Test
+    public void workerScheduleSetAsync() {
+        ShadowMessageQueue mainMessageQueue = shadowOf(Looper.getMainLooper().getQueue());
+
+        Worker worker = scheduler.createWorker();
+        worker.schedule(new Runnable() {
+            @Override public void run() {
+            }
+        });
+
+        Message message = mainMessageQueue.getHead();
+        assertEquals(async, message.isAsynchronous());
+    }
+
+    @Test
+    public void workerSchedulePeriodicallySetAsync() {
+        ShadowMessageQueue mainMessageQueue = shadowOf(Looper.getMainLooper().getQueue());
+
+        Worker worker = scheduler.createWorker();
+        worker.schedulePeriodically(new Runnable() {
+            @Override public void run() {
+            }
+        }, 1, 1, MINUTES);
+
+        Message message = mainMessageQueue.getHead();
+        assertEquals(async, message.isAsynchronous());
+    }
+
     private static void idleMainLooper(long amount, TimeUnit unit) {
         // TODO delete this when https://github.com/robolectric/robolectric/pull/2592 is released.
         ShadowLooper.idleMainLooper(unit.toMillis(amount));