source: java/main/src/main/java/com/framsticks/util/dispatching/Dispatching.java @ 101

package com.framsticks.util.dispatching;

import java.util.Timer;

import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.LogManager;

import com.framsticks.util.FramsticksException;

/**
 * @author Piotr Sniegowski
 */
public abstract class Dispatching {
        private static final Logger log = LogManager.getLogger(Dispatching.class);

        protected static final Timer timer = new Timer();

        public static Timer getTimer() {
                return timer;
        }

        public static boolean isThreadSafe() {
                return true;
        }

        public static <C> void dispatchIfNotActive(Dispatcher<C> dispatcher, RunAt<? extends C> runnable) {
                if (dispatcher.isActive()) {
                        runnable.runAt();
                        return;
                }
                dispatcher.dispatch(runnable);
        }

        // public static boolean assertInvokeLater(Dispatcher dispatcher, RunAt runnable) {
        //      dispatcher.invokeLater(runnable);
        //      return true;
        // }

        public static <P, C> void invokeDispatch(Dispatcher<P> dispatcher, final Dispatcher<C> finalDispatcher, final RunAt<C> runnable) {
                dispatcher.dispatch(new RunAt<P>(runnable) {
                        @Override
                        protected void runAt() {
                                finalDispatcher.dispatch(runnable);
                        }
                });
        }

        public static void sleep(double seconds) {
                log.debug("sleeping");
                try {
                        java.lang.Thread.sleep((long) (seconds * 1000));
                } catch (InterruptedException e) {

                }
                log.debug("slept");
        }

        @SuppressWarnings("unchecked")
        public static void dispatcherGuardedInvoke(Joinable joinable, RunAt<?> runnable) {
                if (joinable instanceof Dispatcher) {
                        dispatchIfNotActive(Dispatcher.class.cast(joinable), runnable);
                        return;
                }
                runnable.runAt();
        }

        public static void use(final Joinable joinable, final JoinableParent owner) {
                log.debug("using {} by {}", joinable, owner);
                if (joinable.use(owner)) {
                        log.debug("started {}", joinable);
                } else {
                        log.debug("start of {} already happened", joinable);
                }
        }

        public static void drop(final Joinable joinable, final JoinableParent owner) {
                log.debug("droping {} by {}", joinable, owner);
                if (joinable.drop(owner)) {
                        log.debug("stoped {}", joinable);
                } else {
                        log.debug("stop of {} deferred", joinable);
                }
        }

        public static void join(Joinable joinable) throws InterruptedException {
                log.debug("joining {}", joinable);
                try {
                        joinable.join();
                } catch (InterruptedException e) {
                        log.debug("failed to join {}", joinable);
                        throw e;
                }
                log.debug("joined {}", joinable);
        }

        public static void childChangedState(final JoinableParent parent, final Joinable joinable, final JoinableState state) {
                if (state.ordinal() <= JoinableState.RUNNING.ordinal()) {
                        return;
                }
                dispatcherGuardedInvoke(joinable, new RunAt<Object>(ThrowExceptionHandler.getInstance()) {
                        @Override
                        protected void runAt() {
                                log.debug("joinable {} is notifying parent {} about change to {}", joinable, parent, state);
                                parent.childChangedState(joinable, state);
                        }
                });
        }

        public static void wait(Object object, long millis) {
                try {
                        synchronized (object) {
                                object.wait(millis);
                        }
                } catch (InterruptedException e) {
                }
        }

        public static void joinAbsolutely(Joinable joinable) {
                log.debug("joining absolutely {}", joinable);
                while (true) {
                        try {
                                Dispatching.join(joinable);
                                return;
                        } catch (InterruptedException e) {
                                // throw new FramsticksException().msg("failed to join").arg("dispatcher", dispatcher).cause(e);
                        }
                        log.debug("waiting for {}", joinable);
                        wait(joinable, 500);
                }
        }

        public interface Query<T> extends ExceptionResultHandler {
                T get();
        }

        public static abstract class QueryHandler<T> implements Query<T> {
                ExceptionResultHandler handler;

                /**
                 * @param handler
                 */
                public QueryHandler(ExceptionResultHandler handler) {
                        this.handler = handler;
                }

                @Override
                public void handle(FramsticksException exception) {
                        handler.handle(exception);
                }
        }

        public static class QueryRunner<T, C> extends RunAt<C> {
                protected final Query<T> query;
                T result;
                boolean ready = false;

                /**
                 * @param query
                 */
                public QueryRunner(Query<T> query) {
                        super(query);
                        this.query = query;
                }

                @Override
                protected void runAt() {
                        result = query.get();
                        synchronized (this) {
                                ready = true;
                                this.notifyAll();
                        }
                }

                public T get() {
                        synchronized (this) {
                                while (!ready) {
                                        try {
                                                this.wait(100);
                                        } catch (InterruptedException e) {
                                        }
                                }
                        }
                        return result;
                }
        }

        public static <T, C> T get(Dispatcher<C> dispatcher, Query<T> query) {
                QueryRunner<T, C> runner = new QueryRunner<T, C>(query);
                dispatcher.dispatch(runner);
                return runner.get();
        }

        // public static class DispatcherWaiter<C, T extends Dispatcher<C> & Joinable> implements Dispatcher<C> {
        //      // protected boolean done = false;
        //      protected final T dispatcher;
        //      protected RunAt<? extends C> runnable;

        //      /**
        //       * @param joinable
        //       */
        //      public DispatcherWaiter(T dispatcher) {
        //              this.dispatcher = dispatcher;
        //      }

        //      public synchronized void waitFor() {
        //              while ((runnable == null) && (dispatcher.getState().ordinal() <= JoinableState.RUNNING.ordinal())) {
        //                      try {
        //                              this.wait();
        //                      } catch (InterruptedException e) {
        //                      }
        //              }
        //              if (runnable != null) {
        //                      runnable.run();
        //              }

        //      }

        //      @Override
        //      public boolean isActive() {
        //              return dispatcher.isActive();
        //      }

        //      @Override
        //      public synchronized void dispatch(RunAt<? extends C> runnable) {
        //              this.runnable = runnable;
        //              this.notifyAll();
        //      }

        // }

        public static class Waiter {
                protected boolean done = false;

                protected final double timeOut;
                protected final ExceptionResultHandler handler;

                /**
                 * @param timeOut
                 */
                public Waiter(double timeOut, ExceptionResultHandler handler) {
                        this.timeOut = timeOut;
                        this.handler = handler;
                }

                public synchronized void pass() {
                        done = true;
                        this.notify();
                }

                public synchronized void waitFor() {
                        long end = System.currentTimeMillis() + (int)(timeOut * 1000);
                        while ((!done) && System.currentTimeMillis() < end) {
                                try {
                                        this.wait(end - System.currentTimeMillis());
                                } catch (InterruptedException e) {
                                        break;
                                }
                        }
                        if (!done) {
                                handler.handle(new FramsticksException().msg("waiter timed out"));
                        }
                }

                public <T> Future<T> passInFuture(Class<T> type) {
                        return new FutureHandler<T>(handler) {
                                @Override
                                protected void result(T result) {
                                        Waiter.this.pass();
                                }
                        };
                }
        }


        public static <C> void synchronize(Dispatcher<C> dispatcher, double seconds) {
                final Waiter waiter = new Waiter(seconds, ThrowExceptionHandler.getInstance());
                dispatcher.dispatch(new RunAt<C>(ThrowExceptionHandler.getInstance()) {
                        @Override
                        protected void runAt() {
                                waiter.pass();
                        }
                });
                waiter.waitFor();
        }

}