package rx.concurrency;

import android.os.Handler;
import rx.Scheduler;
import rx.Subscription;
import rx.operators.AtomicObservableSubscription;
import rx.util.functions.Func2;

import java.util.concurrent.TimeUnit;

/**
 * Schedules actions to run on an Android Handler thread.
 */
public class HandlerThreadScheduler extends Scheduler {

    private final Handler handler;

    public HandlerThreadScheduler(Handler handler) {
        this.handler = handler;
    }

    @Override
    public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action) {
        final AtomicObservableSubscription subscription = new AtomicObservableSubscription();
        final Scheduler _scheduler = this;

        handler.post(new Runnable() {
            @Override
            public void run() {
                subscription.wrap(action.call(_scheduler, state));
            }
        });
        return subscription;
    }

    @Override
    public <T> Subscription schedule(final T state, final Func2<Scheduler, T, Subscription> action, long delayTime, TimeUnit unit) {
        if (delayTime == 0) {
            return schedule(state, action);
        } else {
            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;
        }
    }
}



@@ -56,165 +56,552 @@
 ort 
-rx.Scheduler;%0Aimport rx.Subscription;%0Aimport rx.operators.AtomicObservableSubscription;%0Aimport rx.util.functions.Func2;%0A%0Aimport java.util.concurrent.TimeUnit
+org.junit.Test;%0Aimport org.junit.runner.RunWith;%0Aimport org.mockito.ArgumentCaptor;%0Aimport rx.Scheduler;%0Aimport rx.Subscription;%0Aimport rx.android.testsupport.AndroidTestRunner;%0Aimport rx.operators.AtomicObservableSubscription;%0Aimport rx.util.functions.Func2;%0A%0Aimport java.util.concurrent.TimeUnit;%0A%0Aimport static org.mockito.Matchers.any;%0Aimport static org.mockito.Matchers.anyLong;%0Aimport static org.mockito.Matchers.eq;%0Aimport static org.mockito.Mockito.mock;%0Aimport static org.mockito.Mockito.never;%0Aimport static org.mockito.Mockito.verify
 ;%0A%0A/
@@ -1998,13 +1998,2286 @@
  %7D%0A    %7D
+%0A%0A    @RunWith(AndroidTestRunner.class)%0A    public static final class UnitTest %7B%0A%0A        @Test%0A        public void shouldScheduleActionOnHandlerThread() %7B%0A            final Handler handler = mock(Handler.class);%0A            final Object state = new Object();%0A            final Func2%3CScheduler, Object, Subscription%3E action = mock(Func2.class);%0A%0A            Scheduler scheduler = new HandlerThreadScheduler(handler);%0A            scheduler.schedule(state, action);%0A%0A            // verify that we post to the given Handler%0A            ArgumentCaptor%3CRunnable%3E runnable = ArgumentCaptor.forClass(Runnable.class);%0A            verify(handler).post(runnable.capture());%0A%0A            // verify that the given handler delegates to our action%0A            runnable.getValue().run();%0A            verify(action).call(scheduler, state);%0A        %7D%0A%0A        @Test%0A        public void shouldScheduleDelayedActionOnHandlerThread() %7B%0A            final Handler handler = mock(Handler.class);%0A            final Object state = new Object();%0A            final Func2%3CScheduler, Object, Subscription%3E action = mock(Func2.class);%0A%0A            Scheduler scheduler = new HandlerThreadScheduler(handler);%0A            scheduler.schedule(state, action, 1L, TimeUnit.SECONDS);%0A%0A            // verify that we post to the given Handler%0A            ArgumentCaptor%3CRunnable%3E runnable = ArgumentCaptor.forClass(Runnable.class);%0A            verify(handler).postDelayed(runnable.capture(), eq(1000L));%0A%0A            // verify that the given handler delegates to our action%0A            runnable.getValue().run();%0A            verify(action).call(scheduler, state);%0A        %7D%0A%0A        @Test%0A        public void scheduleDelayedActionShouldForwardToNormalPostIfDelayIsZero() %7B%0A            final Handler handler = mock(Handler.class);%0A            final Object state = new Object();%0A            final Func2%3CScheduler, Object, Subscription%3E action = mock(Func2.class);%0A%0A            Scheduler scheduler = new HandlerThreadScheduler(handler);%0A            scheduler.schedule(state, action, 0L, TimeUnit.SECONDS);%0A%0A            // verify that we post to the given Handler%0A            verify(handler).post(any(Runnable.class));%0A            verify(handler, never()).postDelayed(any(Runnable.class), anyLong());%0A        %7D%0A%0A    %7D
 %0A%7D%0A%0A%0A

Add unit tests for HandlerThreadScheduler