From df66fa199ece86374e14ac64adf71b3158097d71 Mon Sep 17 00:00:00 2001
From: Ben Christensen <benjchristensen@netflix.com>
Date: Fri, 14 Feb 2014 12:40:15 -0800
Subject: [PATCH] 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);
            }
        });
---
 rxjava-core/src/main/java/rx/Scheduler.java | 51 +++++++++++++++++++++
 1 file changed, 51 insertions(+)

diff --git a/rxjava-core/src/main/java/rx/Scheduler.java b/rxjava-core/src/main/java/rx/Scheduler.java
index fd93a92972..29d5e70f99 100644
--- a/rxjava-core/src/main/java/rx/Scheduler.java
+++ b/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 {
 
         /**