api_monitor_competition

package test;

import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

import com.google.common.base.Preconditions;
 
public class ApiMonitor {
	private final ScheduledExecutorService periodicPingScheduler;
	private final LoanApi api;
	private final ExecutorService pingExecutor;
	private final ApiMonitorConfig config;
	private final FailureTracker failureTracker;

	private ScheduledFuture<?> runningTask;

	public ApiMonitor(LoanApi api, ScheduledExecutorService periodicPingScheduler, ExecutorService pingExecutor,
			ApiMonitorConfig config, FailureTracker failureTracker) {
		this.api = api;
		this.periodicPingScheduler = periodicPingScheduler;
		this.pingExecutor = pingExecutor;
		this.config = config;
		this.failureTracker = failureTracker;
	}

	public void start() {
		Preconditions.checkArgument(runningTask == null, "This ApiMonitor was already started in the past");
		runningTask = periodicPingScheduler.scheduleWithFixedDelay(this::check, 0, config.checkDelay, config.checkDelayUnit);
	}

	public void stop() {
		runningTask.cancel(true);
	}

	private void check() {
		Future<?> execution = pingExecutor.submit(this::callApi);
		try {
			execution.get(config.pingTimeout, config.pingTimeoutUnit);
			failureTracker.success();
		} catch (InterruptedException e) {
			// interrupted -> stop monitoring
			stop();
		} catch (RuntimeException | ExecutionException | TimeoutException e) {
			failureTracker.failure();
		} finally {
			execution.cancel(true);
		}
	}

	private void callApi() {
		api.ping();
	}

	public static enum ApiStatus {
		Up, Down;
	}

	public static class FailureTracker {
		private final AtomicInteger consecutiveFailures = new AtomicInteger(0);
		private final ApiMonitorConfig config;
		private final Consumer<ApiStatus> apiStatusListener;

		public FailureTracker(ApiMonitorConfig config, Consumer<ApiStatus> apiStatusListener) {
			this.config = config;
			this.apiStatusListener = apiStatusListener;
		}

		public ApiStatus getApiStatus() {
			if (consecutiveFailures.get() < config.failureThreshold) {
				return ApiStatus.Up;
			} else {
				return ApiStatus.Down;
			}
		}

		private void success() {
			consecutiveFailures.set(0);
			publishStatus();
		}

		private void failure() {
			consecutiveFailures.incrementAndGet();
			publishStatus();
		}

		private void publishStatus() {
			apiStatusListener.accept(getApiStatus());
		}
	}

	public static class ApiMonitorConfig {
		public TimeUnit pingTimeoutUnit = TimeUnit.MILLISECONDS;
		public long pingTimeout = 500;
		public TimeUnit checkDelayUnit = TimeUnit.MILLISECONDS;
		public long checkDelay = 1000;
		public int failureThreshold = 3;
	}
	
	public static void main(String... args) throws Exception {
		LoanApi api = () -> {
			try {
				if (Math.random() < 0.3) {
					System.out.println("ping fail");
					throw new RuntimeException("ping failed");
				} else {
					int sleep = new Random().nextInt(1000);
					System.out.println("sleep " + sleep + "ms");
					Thread.sleep(sleep);
				}
			} catch (InterruptedException e) {
			}
		};

		ApiMonitorConfig cfg = new ApiMonitorConfig();
		FailureTracker tracker = new FailureTracker(cfg, (status) -> System.out.println("  api is " + status));
		ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
		ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
		ApiMonitor monitor = new ApiMonitor(api, scheduler, fixedThreadPool, cfg, tracker);
		
		monitor.start();
		Thread.sleep(30000);
		monitor.stop();
		
		scheduler.shutdownNow();
		fixedThreadPool.shutdownNow();
	}
}