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

package com.framsticks.util.dispatching;

import org.apache.log4j.Logger;

import com.framsticks.util.FramsticksException;

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

        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 " + joinable + " by " + owner);
                if (joinable.use(owner)) {
                        log.debug("started " + joinable);
                } else {
                        log.debug("start of " + joinable + " already happened");
                }
        }

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

        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) {
                dispatcherGuardedInvoke(joinable, new RunAt<Object>(ThrowExceptionHandler.getInstance()) {
                        @Override
                        protected void runAt() {
                                log.debug("joinable " + joinable + " is notifying parent " + parent + " about change to " + 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> {
                T get();
        }

        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) {
                        //TODO TEH
                        super(ThrowExceptionHandler.getInstance());
                        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 Waiter {
                protected boolean done = false;

                protected final double timeOut;

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

                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) {
                                throw new FramsticksException().msg("waiter timed out");
                        }
                }
        }


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

}