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

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) {
                if (state.ordinal() <= JoinableState.RUNNING.ordinal()) {
                        return;
                }
                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> 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();
        }

}