content: /*
   * Copyright (C) 2020 ActiveJ LLC.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   * http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package io.activej.service.adapter;
  
  import io.activej.async.service.ReactiveService;
  import io.activej.common.annotation.StaticFactories;
  import io.activej.common.service.BlockingService;
  import io.activej.eventloop.Eventloop;
  import io.activej.inject.binding.OptionalDependency;
  import io.activej.net.ReactiveServer;
  import io.activej.reactor.net.BlockingSocketServer;
  import io.activej.service.Service;
  import org.slf4j.Logger;
  
  import javax.sql.DataSource;
  import java.io.IOException;
  import java.sql.Connection;
  import java.util.*;
  import java.util.concurrent.*;
  
  import static org.slf4j.LoggerFactory.getLogger;
  
  @SuppressWarnings("WeakerAccess")
  @StaticFactories(ServiceAdapter.class)
  public class ServiceAdapters {
  	private static final Logger logger = getLogger(ServiceAdapters.class);
  
  	public abstract static class SimpleServiceAdapter<S> implements ServiceAdapter<S> {
  		private final boolean startConcurrently;
  		private final boolean stopConcurrently;
  
  		protected SimpleServiceAdapter(boolean startConcurrently, boolean stopConcurrently) {
  			this.startConcurrently = startConcurrently;
  			this.stopConcurrently = stopConcurrently;
  		}
  
  		protected SimpleServiceAdapter() {
  			this(true, true);
  		}
  
  		protected abstract void start(S instance) throws Exception;
  
  		protected abstract void stop(S instance) throws Exception;
  
  		@Override
  		public final CompletableFuture<?> start(S instance, Executor executor) {
  			CompletableFuture<?> future = new CompletableFuture<>();
  			(startConcurrently ? executor : (Executor) Runnable::run).execute(() -> {
  				try {
  					start(instance);
  					future.complete(null);
  				} catch (Exception e) {
  					future.completeExceptionally(e);
  				}
  			});
  			return future;
  		}
  
  		@Override
  		public final CompletableFuture<?> stop(S instance, Executor executor) {
  			CompletableFuture<?> future = new CompletableFuture<>();
  			(stopConcurrently ? executor : (Executor) Runnable::run).execute(() -> {
  				try {
  					stop(instance);
  					future.complete(null);
  				} catch (Exception e) {
  					future.completeExceptionally(e);
  				}
  			});
  			return future;
  		}
  	}
  
  	public static ServiceAdapter<Service> forService() {
  		return new ServiceAdapter<>() {
  			@Override
  			public CompletableFuture<?> start(Service instance, Executor executor) {
  				return instance.start();
  			}
  
  			@Override
  			public CompletableFuture<?> stop(Service instance, Executor executor) {
  				return instance.stop();
  			}
  		};
  	}
  
  	/**
  	 * Returns factory which transforms blocking Service to asynchronous non-blocking CompletableFuture. It runs blocking operations from other thread from
  	 * executor.
  	 */
  	public static ServiceAdapter<BlockingService> forBlockingService() {
  		return new SimpleServiceAdapter<>() {
  			@Override
  			protected void start(BlockingService instance) throws Exception {
  				instance.start();
  			}
  
  			@Override
  			protected void stop(BlockingService instance) throws Exception {
  				instance.stop();
  			}
  		};
  	}
  
  	public static <T> ServiceAdapter<OptionalDependency<T>> forOptionalDependency(ServiceAdapter<T> adapter) {
  		return new ServiceAdapter<>() {
  			@Override
  			public CompletableFuture<?> start(OptionalDependency<T> optional, Executor executor) {
  				if (optional.isPresent()) {
  					T instance = optional.get();
  					return adapter.start(instance, executor);
  				}
  				return CompletableFuture.completedFuture(null);
  			}
  
  			@Override
  			public CompletableFuture<?> stop(OptionalDependency<T> optional, Executor executor) {
  				if (optional.isPresent()) {
  					T instance = optional.get();
  					return adapter.stop(instance, executor);
  				}
  				return CompletableFuture.completedFuture(null);
  			}
  		};
  	}
  
  	/**
  	 * Returns factory which transforms Timer to CompletableFuture. On start it does nothing, on stop it cancel timer.
  	 */
  	public static ServiceAdapter<Timer> forTimer() {
  		return new SimpleServiceAdapter<>(false, false) {
  			@Override
  			protected void start(Timer instance) {
  			}
  
  			@Override
  			protected void stop(Timer instance) {
  				instance.cancel();
  			}
  		};
  	}
  
  	/**
  	 * Returns factory which transforms ExecutorService to CompletableFuture. On starting it doing nothing, on stopping it shuts down ExecutorService.
  	 */
  	public static ServiceAdapter<ExecutorService> forExecutorService() {
  		return new SimpleServiceAdapter<>(false, true) {
  			@Override
  			protected void start(ExecutorService instance) {
  			}
  
  			@Override
  			protected void stop(ExecutorService instance) throws Exception {
  				instance.shutdown();
  				if (!instance.isTerminated()) {
  					logger.warn("Awaiting termination of {} ...", instance);
  					//noinspection ResultOfMethodCallIgnored
  					instance.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
  					logger.info("Instance {} has been terminated", instance);
  				}
  			}
  		};
  	}
  
  	/**
  	 * Returns factory which transforms AutoCloseable object to CompletableFuture. On starting it doing nothing, on stopping it close AutoCloseable.
  	 */
  	public static ServiceAdapter<AutoCloseable> forAutoCloseable() {
  		return new SimpleServiceAdapter<>(false, true) {
  			@Override
  			protected void start(AutoCloseable instance) {
  			}
  
  			@Override
  			protected void stop(AutoCloseable instance) throws Exception {
  				instance.close();
  			}
  		};
  	}
  
  	/**
  	 * Returns factory which transforms DataSource object to CompletableFuture. On start it checks connecting , on stop it closes DataSource.
  	 */
  	public static ServiceAdapter<DataSource> forDataSource() {
  		return new SimpleServiceAdapter<>(true, false) {
  			@Override
  			protected void start(DataSource instance) throws Exception {
  				Connection connection = instance.getConnection();
  				connection.close();
  			}
  
  			@Override
  			protected void stop(DataSource instance) {
  			}
  		};
  	}
  
  	public static ServiceAdapter<ReactiveService> forReactiveService() {
  		return new ServiceAdapter<>() {
  			@Override
  			public CompletableFuture<?> start(ReactiveService instance, Executor executor) {
  				CompletableFuture<Object> future = new CompletableFuture<>();
  				instance.getReactor().execute(() -> {
  					try {
  						instance.start()
  							.whenResult(future::complete)
  							.whenException(future::completeExceptionally);
  					} catch (Exception e) {
  						future.completeExceptionally(e);
  					}
  				});
  				return future;
  			}
  
  			@Override
  			public CompletableFuture<?> stop(ReactiveService instance, Executor executor) {
  				CompletableFuture<Object> future = new CompletableFuture<>();
  				instance.getReactor().execute(() -> {
  					try {
  						instance.stop()
  							.whenResult(future::complete)
  							.whenException(future::completeExceptionally);
  					} catch (Exception e) {
  						future.completeExceptionally(e);
  					}
  				});
  				return future;
  			}
  		};
  	}
  
  	public static ServiceAdapter<ReactiveServer> forReactiveServer() {
  		return new ServiceAdapter<>() {
  			@Override
  			public CompletableFuture<?> start(ReactiveServer instance, Executor executor) {
  				CompletableFuture<?> future = new CompletableFuture<>();
  				instance.getReactor().execute(() -> {
  					try {
  						instance.listen();
  						future.complete(null);
  					} catch (IOException e) {
  						future.completeExceptionally(e);
  					}
  				});
  				return future;
  			}
  
  			@Override
  			public CompletableFuture<?> stop(ReactiveServer instance, Executor executor) {
  				CompletableFuture<Object> future = new CompletableFuture<>();
  				instance.getReactor().execute(() -> instance.close()
  					.whenResult(future::complete)
  					.whenException(future::completeExceptionally));
  				return future;
  			}
  		};
  	}
  
  	public static ServiceAdapter<Eventloop> forEventloop(ThreadFactory threadFactory) {
  		return new ServiceAdapter<>() {
  			@Override
  			public CompletableFuture<?> start(Eventloop eventloop, Executor executor) {
  				CompletableFuture<?> future = new CompletableFuture<>();
  				threadFactory.newThread(() -> {
  					eventloop.keepAlive(true);
  					future.complete(null);
  					eventloop.run();
  				}).start();
  				return future;
  			}
  
  			@Override
  			public CompletableFuture<?> stop(Eventloop eventloop, Executor executor) {
  				Thread eventloopThread = eventloop.getEventloopThread();
  				if (eventloopThread == null) {
  					// already stopped
  					return CompletableFuture.completedFuture(null);
  				}
  				CompletableFuture<?> future = new CompletableFuture<>();
  				eventloop.execute(() -> {
  					eventloop.keepAlive(false);
  					logStopping(eventloop);
  					Eventloop.logger.info("Waiting for {}", eventloop);
  				});
  				executor.execute(() -> {
  					try {
  						eventloopThread.join();
  						future.complete(null);
  					} catch (InterruptedException e) {
  						Thread.currentThread().interrupt();
  						future.completeExceptionally(e);
  					}
  				});
  				return future;
  			}
  
  			private void logStopping(Eventloop eventloop) {
  				eventloop.delayBackground(1000L, () -> {
  					if (eventloop.getEventloopThread() != null) {
  						Eventloop.logger.info("...Waiting for {}", eventloop);
  						logStopping(eventloop);
  					}
  				});
  			}
  		};
  	}
  
  	public static ServiceAdapter<Eventloop> forEventloop() {
  		ThreadFactory threadFactory = Executors.defaultThreadFactory();
  		return forEventloop(r -> {
  			Thread thread = threadFactory.newThread(r);
  			thread.setName("eventloop: " + thread.getName());
  			return thread;
  		});
  	}
  
  	public static ServiceAdapter<BlockingSocketServer> forBlockingSocketServer() {
  		return new ServiceAdapters.SimpleServiceAdapter<>() {
  			@Override
  			protected void start(BlockingSocketServer instance) throws Exception {
  				instance.start();
  			}
  
  			@Override
  			protected void stop(BlockingSocketServer instance) throws Exception {
  				instance.stop();
  			}
  		};
  	}
  
  	public static <T> ServiceAdapter<T> immediateServiceAdapter() {
  		return new SimpleServiceAdapter<>(false, false) {
  			@Override
  			protected void start(T instance) {
  			}
  
  			@Override
  			protected void stop(T instance) {
  			}
  		};
  	}
  
  	@SafeVarargs
  	public static <T, S extends ServiceAdapter<? super T>> ServiceAdapter<T> combinedAdapter(S... startOrder) {
  		return combinedAdapter(List.of(startOrder));
  	}
  
  	public static <T> ServiceAdapter<T> combinedAdapter(List<? extends ServiceAdapter<? super T>> startOrder) {
  		List<? extends ServiceAdapter<? super T>> stopOrder = new ArrayList<>(startOrder);
  		Collections.reverse(stopOrder);
  		return combinedAdapter(startOrder, stopOrder);
  	}
  
  	@FunctionalInterface
  	public interface Action<T> {
  		CompletableFuture<?> doAction(ServiceAdapter<T> serviceAdapter, T instance, Executor executor);
  	}
  
  	public static <T> ServiceAdapter<T> combinedAdapter(
  		List<? extends ServiceAdapter<? super T>> startOrder, List<? extends ServiceAdapter<? super T>> stopOrder
  	) {
  		return new ServiceAdapter<>() {
  			@SuppressWarnings("unchecked")
  			private void doAction(
  				T instance, Executor executor, Iterator<? extends ServiceAdapter<? super T>> iterator,
  				CompletableFuture<?> future, Action<T> action
  			) {
  				if (iterator.hasNext()) {
  					action.doAction((ServiceAdapter<T>) iterator.next(), instance, executor)
  						.whenCompleteAsync(($, e) -> {
  							if (e == null) {
  								doAction(instance, executor, iterator, future, action);
  							} else if (e instanceof InterruptedException) {
  								Thread.currentThread().interrupt();
  								future.completeExceptionally(e);
  							} else if (e instanceof ExecutionException) {
  								future.completeExceptionally(e.getCause());
  							}
  						}, Runnable::run);
  				} else {
  					future.complete(null);
  				}
  			}
  
  			@Override
  			public CompletableFuture<Void> start(T instance, Executor executor) {
  				CompletableFuture<Void> future = new CompletableFuture<>();
  				doAction(instance, executor, startOrder.iterator(), future,
  					ServiceAdapter::start);
  				return future;
  			}
  
  			@Override
  			public CompletableFuture<Void> stop(T instance, Executor executor) {
  				CompletableFuture<Void> future = new CompletableFuture<>();
  				doAction(instance, executor, stopOrder.iterator(), future,
  					ServiceAdapter::stop);
  				return future;
  			}
  		};
  	}
  }
  