/**
 * Copyright 2014 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package rx.schedulers;

import javafx.animation.KeyFrame;
import javafx.animation.Timeline;
import javafx.application.Platform;
import javafx.event.ActionEvent;
import javafx.event.EventHandler;
import javafx.util.Duration;
import rx.Scheduler;
import rx.Subscription;
import rx.functions.Action0;
import rx.subscriptions.BooleanSubscription;
import rx.subscriptions.CompositeSubscription;
import rx.subscriptions.SerialSubscription;
import rx.subscriptions.Subscriptions;

import java.awt.*;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;

/**
 * Executes work on the JavaFx UI thread.
 * This scheduler should only be used with actions that execute quickly.
 */
public final class JavaFxScheduler extends Scheduler {
    private static final JavaFxScheduler INSTANCE = new JavaFxScheduler();

    /* package for unit test */JavaFxScheduler() {
    }

    public static JavaFxScheduler getInstance() {
        return INSTANCE;
    }

    private static void assertThatTheDelayIsValidForTheJavaFxTimer(long delay) {
        if (delay < 0 || delay > Integer.MAX_VALUE) {
            throw new IllegalArgumentException(String.format("The JavaFx timer only accepts non-negative delays up to %d milliseconds.", Integer.MAX_VALUE));
        }
    }

    @Override
    public Worker createWorker() {
        return new InnerJavaFxScheduler();
    }

    private static class InnerJavaFxScheduler extends Worker implements Runnable {

        private final CompositeSubscription tracking = new CompositeSubscription();

        /** Allows cheaper trampolining than invokeLater(). Accessed from EDT only. */
        private final Queue<Runnable> queue = new ConcurrentLinkedQueue<>();
        /** Allows cheaper trampolining than invokeLater(). Accessed from EDT only. */
        private int wip;

        @Override
        public void unsubscribe() {
            tracking.unsubscribe();
        }

        @Override
        public boolean isUnsubscribed() {
            return tracking.isUnsubscribed();
        }

        @Override
        public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
            long delay = unit.toMillis(delayTime);
            assertThatTheDelayIsValidForTheJavaFxTimer(delay);

            class DualAction implements EventHandler<ActionEvent>, Subscription, Runnable {

                private Timeline timeline;

                final SerialSubscription subs = new SerialSubscription();
                boolean nonDelayed;

                private void setTimer(Timeline timeline) {
                    this.timeline = timeline;
                }

                @Override
                public void handle(ActionEvent event) {
                    run();
                }

                @Override
                public void run() {
                    if (nonDelayed) {
                        try {
                            if (tracking.isUnsubscribed() || isUnsubscribed()) {
                                return;
                            }
                            action.call();
                        } finally {
                            subs.unsubscribe();
                        }
                    } else {
                        timeline.stop();
                        timeline = null;
                        nonDelayed = true;
                        trampoline(this);
                    }
                }

                @Override
                public boolean isUnsubscribed() {
                    return subs.isUnsubscribed();
                }

                @Override
                public void unsubscribe() {
                    subs.unsubscribe();
                }
                public void set(Subscription s) {
                    subs.set(s);
                }
            }

            final DualAction executeOnce = new DualAction();
            tracking.add(executeOnce);

            //final Timeline timer = new Timeline((int) delay, executeOnce);
            final Timeline timer = new Timeline(new KeyFrame(Duration.millis(delay)));

            executeOnce.setTimer(timer);
            timer.play();

            executeOnce.set(Subscriptions.create(() -> {
                timer.stop();
                tracking.remove(executeOnce);
            }));

            return executeOnce;
        }

        @Override
        public Subscription schedule(final Action0 action) {
            final BooleanSubscription s = BooleanSubscription.create();
            Runnable runnable = () -> {
                if (tracking.isUnsubscribed() || s.isUnsubscribed()) {
                    return;
                }
                action.call();
                tracking.remove(s);
            };

            if (Platform.isFxApplicationThread()) {
                if (trampoline(runnable)) {
                    return Subscriptions.unsubscribed();
                }
                else {
                    queue.offer(runnable);
                    EventQueue.invokeLater(this);
                }
            }

            tracking.add(s);
            // wrap for returning so it also removes it from the 'innerSubscription'
            return Subscriptions.create(() -> {
                s.unsubscribe();
                tracking.remove(s);
            });
        }
        /**
         * Uses a fast-path/slow path trampolining and tries to run
         * the given runnable directly.
         * @param runnable
         * @return true if the fast path was taken
         */
        boolean trampoline(Runnable runnable) {
            // fast path: if wip increments from 0 to 1
            if (wip == 0) {
                wip = 1;
                runnable.run();
                // but a recursive schedule happened
                if (--wip > 0) {
                    do {
                        Runnable r = queue.poll();
                        r.run();
                    } while (--wip > 0);
                }
                return true;
            }
            queue.offer(runnable);
            run();
            return false;
        }
        @Override
        public void run() {
            if (wip++ == 0) {
                do {
                    Runnable r = queue.poll();
                    r.run();
                } while (--wip > 0);
            }
        }
    }
}

@@ -1,598 +1,4 @@
-/**%0A * Copyright 2014 Netflix, Inc.%0A *%0A * Licensed under the Apache License, Version 2.0 (the %22License%22);%0A * you may not use this file except in compliance with the License.%0A * You may obtain a copy of the License at%0A *%0A * http://www.apache.org/licenses/LICENSE-2.0%0A *%0A * Unless required by applicable law or agreed to in writing, software%0A * distributed under the License is distributed on an %22AS IS%22 BASIS,%0A * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A * See the License for the specific language governing permissions and%0A * limitations under the License.%0A */%0A
 pack
@@ -471,27 +471,8 @@
 s;%0A%0A
-import java.awt.*;%0A
 impo
@@ -2183,16 +2183,27 @@
 delay = 
+Math.max(0,
 unit.toM
@@ -2218,16 +2218,17 @@
 layTime)
+)
 ;%0A      
@@ -2369,33 +2369,32 @@
 ion, Runnable %7B%0A
-%0A
                 
@@ -2420,17 +2420,16 @@
 meline;%0A
-%0A
         
@@ -3948,85 +3948,8 @@
 );%0A%0A
-            //final Timeline timer = new Timeline((int) delay, executeOnce);%0A
     
@@ -4027,21 +4027,33 @@
 s(delay)
+, executeOnce
 ));%0A
-%0A
         
@@ -4489,32 +4489,58 @@
 nable = () -%3E %7B%0A
+                try %7B%0A    
                 
@@ -4586,32 +4586,36 @@
 subscribed()) %7B%0A
+    
                 
@@ -4634,34 +4634,42 @@
                 
-%7D%0A
+    %7D%0A    
                 
@@ -4703,41 +4703,120 @@
     
-tracking.remove(s);%0A            %7D
+%7D finally %7B%0A                    tracking.remove(s);%0A                %7D%0A            %7D;%0A            tracking.add(s)
 ;%0A%0A 
@@ -5075,25 +5075,20 @@
     
-EventQueue.invoke
+Platform.run
 Late
@@ -5133,37 +5133,8 @@
  %7D%0A%0A
-            tracking.add(s);%0A
     
@@ -5263,59 +5263,8 @@
 ) -%3E
- %7B%0A                s.unsubscribe();%0A               
  tra
@@ -5278,31 +5278,16 @@
 emove(s)
-;%0A            %7D
 );%0A     

Fix action execution timing