Scheduler.scheduleRecursive

Make recursive scheduling a first-class citizen without needing to refer to "this" which doesn't work inside lambdas (Java8/Scala/etc).

        Schedulers.newThread().scheduleRecursive(new Action1<Recurse>() {
            @OverRide
            public void call(Recurse r) {
                System.out.println("do stuff on thread: " + Thread.currentThread() + " Time: " + System.nanoTime());
                r.schedule(100, TimeUnit.MILLISECONDS);
            }
        });

Author: benjchristensen

rxjava-core/src/main/java/rx/Scheduler.java

@@ -94,6 +94,57 @@ public void call(Inner inner) {
         return schedule(recursiveAction, initialDelay, unit);
     }
 
+    public final Subscription scheduleRecursive(final Action1<Recurse> action) {
+        return schedule(new Action1<Inner>() {
+
+            @Override
+            public void call(Inner inner) {
+                action.call(new Recurse(inner, action));
+            }
+
+        });
+    }
+
+    public static final class Recurse {
+        final Action1<Recurse> action;
+        final Inner inner;
+
+        private Recurse(Inner inner, Action1<Recurse> action) {
+            this.inner = inner;
+            this.action = action;
+        }
+
+        /**
+         * Schedule the current function for execution immediately.
+         */
+        public final void schedule() {
+            final Recurse self = this;
+            inner.schedule(new Action1<Inner>() {
+
+                @Override
+                public void call(Inner _inner) {
+                    action.call(self);
+                }
+
+            });
+        }
+
+        /**
+         * Schedule the current function for execution in the future.
+         */
+        public final void schedule(long delay, TimeUnit unit) {
+            final Recurse self = this;
+            inner.schedule(new Action1<Inner>() {
+
+                @Override
+                public void call(Inner _inner) {
+                    action.call(self);
+                }
+
+            }, delay, unit);
+        }
+    }
+
     public abstract static class Inner implements Subscription {
 
         /**