package org.infinispan.upgrade;

import static org.infinispan.util.logging.Log.CONTAINER;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.infinispan.Cache;
import org.infinispan.commons.util.ServiceFinder;
import org.infinispan.commons.util.Util;
import org.infinispan.configuration.global.GlobalConfiguration;
import org.infinispan.factories.annotations.Inject;
import org.infinispan.factories.annotations.SurvivesRestarts;
import org.infinispan.factories.scopes.Scope;
import org.infinispan.factories.scopes.Scopes;
import org.infinispan.jmx.annotations.MBean;
import org.infinispan.jmx.annotations.ManagedOperation;
import org.infinispan.jmx.annotations.Parameter;
import org.infinispan.commons.time.TimeService;
import org.infinispan.util.logging.Log;
import org.infinispan.util.logging.LogFactory;

/**
 * RollingUpgradeManager handles the synchronization of data between Infinispan
 * clusters when performing rolling upgrades.
 *
 * @author Manik Surtani
 * @author Tristan Tarrant
 * @since 5.2
 */
@MBean(objectName = "RollingUpgradeManager", description = "Handles the migration of data when upgrading between versions.")
@Scope(value = Scopes.NAMED_CACHE)
@SurvivesRestarts
public class RollingUpgradeManager {
   private static final Log log = LogFactory.getLog(RollingUpgradeManager.class);
   private final Set<SourceMigrator> sourceMigrators = new HashSet<>(2);
   @Inject Cache<Object, Object> cache;
   @Inject TimeService timeService;
   @Inject GlobalConfiguration globalConfiguration;

   @ManagedOperation(
         description = "Synchronizes data from source clusters to target clusters with the specified migrator.",
         displayName = "Synchronizes data from source clusters to target clusters with the specified migrator."
   )
   public long synchronizeData(@Parameter(name="migratorName", description="Specifies the name of the migrator to use. Set hotrod as the value unless using custom migrators.") String migratorName) throws Exception {
      TargetMigrator migrator = getMigrator(migratorName);
      long start = timeService.time();
      long count = migrator.synchronizeData(cache);
      log.entriesMigrated(count, cache.getName(), Util.prettyPrintTime(timeService.timeDuration(start, TimeUnit.MILLISECONDS)));
      return count;

   }

   @ManagedOperation(
           description = "Synchronizes data from source clusters to target clusters with the specified migrator.",
           displayName = "Synchronizes data from source clusters to target clusters with the specified migrator."
   )
   public long synchronizeData(@Parameter(name = "migratorName", description = "Specifies the name of the migrator to use. Set hotrod as the value unless using custom migrators.") String migratorName,
                               @Parameter(name = "readBatch", description = "Specifies how many entries to read at a time from source clusters. Default is 10000.") int readBatch,
                               @Parameter(name = "threads", description = "Specifies the number of threads to use per node when writing data to target clusters. Defaults to number of available processors.") int threads) throws Exception {
      TargetMigrator migrator = getMigrator(migratorName);
      long start = timeService.time();
      long count = migrator.synchronizeData(cache, readBatch, threads);
      log.entriesMigrated(count, cache.getName(), Util.prettyPrintTime(timeService.timeDuration(start, TimeUnit.MILLISECONDS)));
      return count;
   }

   @ManagedOperation(
         description = "Disconnects target clusters from source clusters.",
         displayName = "Disconnects target clusters from source clusters."
   )
   public void disconnectSource(@Parameter(name="migratorName", description="Specifies the name of the migrator to use. Set hotrod as the value unless using custom migrators.") String migratorName) throws Exception {
      TargetMigrator migrator = getMigrator(migratorName);
      migrator.disconnectSource(cache);
   }

   private TargetMigrator getMigrator(String name) throws Exception {
      ClassLoader cl = globalConfiguration.classLoader();
      for (TargetMigrator m : ServiceFinder.load(TargetMigrator.class, cl)) {
         if (name.equalsIgnoreCase(m.getName())) {
            return m;
         }
      }
      throw CONTAINER.unknownMigrator(name);
   }

   /**
    * Registers a migrator for a specific data format or endpoint. In the Infinispan ecosystem, we'd
    * typically have one Migrator implementation for Hot Rod, one for memcached, one for REST and
    * one for embedded/in-VM mode, and these would typically be added to the upgrade manager on
    * first access via any of these protocols.
    *
    * @param migrator
    */
   public void addSourceMigrator(SourceMigrator migrator) {
      sourceMigrators.add(migrator);
   }
}
