3
\$\begingroup\$

We have some required services down temporarily when our server start. So we need some reconnection logic for them until they are finally up. There is a requirement to have also syncronious way for that. Here is my generic implementation for that via a single background thread:

public class RunUntilSuccess {

private static final Logger log = LoggerFactory.getLogger(RunUntilSuccess.class);

private final String processName;
private final Task task;
private final int maxAttempts;
private final int interval;
private final TimeUnit unit;
private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
private AtomicInteger count = new AtomicInteger(0);

public RunUntilSuccess(String processName, Callable<Boolean> task, int interval, TimeUnit unit) {
    this(processName, task, interval, unit, -1);
}

public RunUntilSuccess(String processName, Callable<Boolean> callable, int interval, TimeUnit unit, int maxAttempts) {
    if (callable == null) {
        throw new IllegalArgumentException("Callable cannot be null");
    }
    this.processName = processName;
    this.task = new Task(callable);
    this.interval = Math.max(0, interval);
    if (unit == null) {
        throw new IllegalArgumentException("Unit cannot be null");
    }
    this.unit = unit;
    this.maxAttempts = (maxAttempts > 0) ? maxAttempts : -1;

    start();
}

private void start() {
    log.debug("Starting task execution. " + toString());
    executor.scheduleWithFixedDelay(task, 0, interval, unit);
}

/**
 * Wait until success
 */
public void await() throws InterruptedException {
    while (!executor.awaitTermination(interval, unit)) { }
}

public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
    return executor.awaitTermination(timeout, unit);
}

private boolean isAttemptsLimitReached(int attempt){
    return (maxAttempts > 0) && (attempt >= maxAttempts);
}

@Override
public String toString() {
    final StringBuilder sb = new StringBuilder();
    sb.append(RunUntilSuccess.class.getSimpleName());
    sb.append("{processName='").append(processName).append('\'');
    sb.append(", interval=").append(interval);
    sb.append(", unit=").append(unit);
    sb.append(", maxAttempts=").append(maxAttempts);
    sb.append('}');
    return sb.toString();
}

private class Task implements Runnable {
    private final Callable<Boolean> task;

    public Task(Callable<Boolean> task) {
        this.task = task;
    }

    @Override
    public void run() {
        if (isAttemptsLimitReached(count.getAndIncrement())) {
            log.debug("Task execution finished unsuccessfully after " + count + " attempts.");
            executor.shutdown();
        } else {
            log.debug("Attempt #" + count);
            try {
                if (Boolean.TRUE.equals(task.call())) {
                    log.debug("Attempt #" + count + " was successful. Task execution finished.");
                    executor.shutdown();
                } else {
                    throw new FalseExecution();
                }

            } catch (Exception e) {
                log.error("Attempt #" + count + " failed due to:", e);
            }
        }
    }
}

public static class FalseExecution extends RuntimeException {

}

usage:

Callable<Boolean> connector = new Callable<Boolean>() {
        @Override
        public Boolean call() throws Exception {
            // connect to ...
            return true;
        }
    };

    RunUntilSuccess initializer = new RunUntilSuccess("name", connector, reconnectTimeoutSeconds, TimeUnit.SECONDS);
            // initializer.await() after it for sync way

Is it a common task and how generic this code? Thanks!

\$\endgroup\$
1
\$\begingroup\$

design you callable something like below - and you can use Futures.get() to get the result back.

public Boolean call() {

        boolean flag  = false;
        while(!(isAttemptsLimitReached(count.getAndIncrement()) || flag)){

         try{

         //try  connection (with timeout)

         }catch(Exception e){
         // show exception 
         }

         //continue loop
      }

        return flag;
}
\$\endgroup\$
  • \$\begingroup\$ might be a good idea to decide on how to spell boolean (capital B or not?) and then stick with it for consistency. \$\endgroup\$ – Adam Dec 28 '11 at 12:17
  • \$\begingroup\$ @codesparkle this is a pointer code and not "code to be copied" ! \$\endgroup\$ – Nrj Dec 28 '11 at 12:53
  • \$\begingroup\$ it was just a suggestion to improve readability... \$\endgroup\$ – Adam Dec 28 '11 at 14:34
  • \$\begingroup\$ @Nrj, thanks, seems for clear code with Future \$\endgroup\$ – yetanothercoder Dec 28 '11 at 18:41
2
\$\begingroup\$

Your use of executor.awaitTermination() seems odd to me. I would instead have used a CountdownLatch to trigger success in a blocked thread. But I also agree with @Craig P. Motlin that a Future makes sense.

Besides that specific comment, you might consider some asynchronous notification technique for the dependency to declare that it's alive. That might be overkill for your scenario, but it's worked for me. Jini service registry, JMS messages, Hazelcast, plain UDP multicast, etc.

\$\endgroup\$
1
\$\begingroup\$

When you submit a task to an Executor you get back a Future, which feels like the appropriate API here since you want to block in one thread on a delayed computation.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.