qbit/core/src/main/java/io/advantageous/qbit/reactive/Reactor.java
public CallbackCoordinator coordinateWithTimeout(final CallbackCoordinator coordinator,
                                                     final long startTime,
                                                     final long timeoutDuration,
                                                     final TimeUnit timeUnit,
                                                     final Runnable timeOutHandler,
                                                     final Runnable finishedHandler) {

        final long timeoutDurationMS = timeUnit.toMillis(timeoutDuration);

        final long theStartTime = startTime == -1 ? currentTime : startTime;


        final CallbackCoordinator wrapper = new CallbackCoordinator() {

            AtomicBoolean done = new AtomicBoolean();

            @Override
            public boolean checkComplete() {
                if (done.get()) {
                    return true;
                }

                if (coordinator.checkComplete()) {
                    done.set(true);
                }

                return done.get();

            }


            public boolean timedOut(long now) {


                if (startTime() == -1 || timeOutDuration() == -1) {
                    return false;
                }
                if ((now - startTime()) > timeOutDuration()) {

                    if (!done.get()) {
                        timeOutHandler.run();
                        done.set(true);
                    }
                    return true;
                } else {
                    return false;
                }
            }

            @Override
            public long timeOutDuration() {
                return coordinator.timeOutDuration() == -1 ? timeoutDurationMS : coordinator.timeOutDuration();
            }

            @Override
            public long startTime() {
                return coordinator.startTime() == -1 ? theStartTime : coordinator.startTime();
            }

            public void finished() {
                if (checkComplete()) {
                    removeCoordinator(this);
                }
                if (finishedHandler != null) {
                    finishedHandler.run();
                }
                coordinator.finished();
            }

            public void cancel() {
                done.set(true);
                removeCoordinator(this);
                coordinator.cancel();
            }
        };


        this.coordinatorQueue.add(wrapper);
        return wrapper;
    }
public CallbackCoordinator coordinateWithTimeout(final CallbackCoordinator coordinator,
                                                     final long startTime,
                                                     final long timeoutDuration,
                                                     final TimeUnit timeUnit,
                                                     final Runnable timeOutHandler,
                                                     final Runnable finishedHandler) {

        final long timeoutDurationMS = timeUnit.toMillis(timeoutDuration);

        final long theStartTime = startTime == -1 ? currentTime : startTime;


        final CallbackCoordinator wrapper = new CallbackCoordinator() {

            AtomicBoolean done = new AtomicBoolean();

            @Override
            public boolean checkComplete() {
                if (done.get()) {
                    return true;
                }

                if (coordinator.checkComplete()) {
                    done.set(true);
                }

                return done.get();

            }


            public boolean timedOut(long now) {


                if (startTime() == -1 || timeOutDuration() == -1) {
                    return false;
                }
                if ((now - startTime()) > timeOutDuration()) {

                    if (!done.get()) {
                        timeOutHandler.run();
                        done.set(true);
                    }
                    return true;
                } else {
                    return false;
                }
            }

            @Override
            public long timeOutDuration() {
                return coordinator.timeOutDuration() == -1 ? timeoutDurationMS : coordinator.timeOutDuration();
            }

            @Override
            public long startTime() {
                return coordinator.startTime() == -1 ? theStartTime : coordinator.startTime();
            }

            public void finished() {
                if (checkComplete()) {
                    removeCoordinator(this);
                }
                if (finishedHandler != null) {
                    finishedHandler.run();
                }
                coordinator.finished();
            }

            public void cancel() {
                done.set(true);
                removeCoordinator(this);
                coordinator.cancel();
            }
        };


        this.coordinatorQueue.add(wrapper);
        return wrapper;
    }
public CallbackCoordinator coordinateWithTimeout ( final CallbackCoordinator coordinator , final long startTime , final long timeoutDuration , final TimeUnit timeUnit , final Runnable timeOutHandler , final Runnable finishedHandler ) { final long timeoutDurationMS = timeUnit . toMillis ( timeoutDuration ) ; final long theStartTime = startTime == - 1 ? currentTime : startTime ; final CallbackCoordinator wrapper = new CallbackCoordinator ( ) { AtomicBoolean done = new AtomicBoolean ( ) ; @ Override public boolean checkComplete ( ) { if ( done . get ( ) ) { return true ; } if ( coordinator . checkComplete ( ) ) { done . set ( true ) ; } return done . get ( ) ; } public boolean timedOut ( long now ) { if ( startTime ( ) == - 1 || timeOutDuration ( ) == - 1 ) { return false ; } if ( ( now - startTime ( ) ) > timeOutDuration ( ) ) { if ( ! done . get ( ) ) { timeOutHandler . run ( ) ; done . set ( true ) ; } return true ; } else { return false ; } } @ Override public long timeOutDuration ( ) { return coordinator . timeOutDuration ( ) == - 1 ? timeoutDurationMS : coordinator . timeOutDuration ( ) ; } @ Override public long startTime ( ) { return coordinator . startTime ( ) == - 1 ? theStartTime : coordinator . startTime ( ) ; } public void finished ( ) { if ( checkComplete ( ) ) { removeCoordinator ( this ) ; } if ( finishedHandler != null ) { finishedHandler . run ( ) ; } coordinator . finished ( ) ; } public void cancel ( ) { done . set ( true ) ; removeCoordinator ( this ) ; coordinator . cancel ( ) ; } } ; this . coordinatorQueue . add ( wrapper ) ; return wrapper ; }
Utility method to create a coordinator.

@param coordinator     coordinator
@param startTime       startTime
@param timeoutDuration timeoutDuration
@param timeUnit        timeUnit
@param timeOutHandler  timeOutHandler
@param finishedHandler finishedHandler
@return callback coordinator
Utility method to create a coordinator .