package org.apache.commons.jcs3.auxiliary.remote;
   
   import java.io.IOException;
   
   /*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
   * to you 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.
   */
  
  import java.util.List;
  import java.util.ListIterator;
  import java.util.concurrent.atomic.AtomicBoolean;
  
  import org.apache.commons.jcs3.auxiliary.remote.behavior.IRemoteCacheAttributes;
  import org.apache.commons.jcs3.auxiliary.remote.server.behavior.RemoteType;
  import org.apache.commons.jcs3.engine.CacheStatus;
  import org.apache.commons.jcs3.engine.behavior.ICache;
  import org.apache.commons.jcs3.engine.behavior.IElementSerializer;
  import org.apache.commons.jcs3.engine.logging.behavior.ICacheEventLogger;
  import org.apache.commons.jcs3.log.Log;
  import org.apache.commons.jcs3.log.LogManager;
  
  /**
   * Used to provide access to multiple services under nowait protection. Factory should construct
   * NoWaitFacade to give to the composite cache out of caches it constructs from the varies manager
   * to lateral services.
   * <p>
   * Typically, we only connect to one remote server per facade. We use a list of one
   * RemoteCacheNoWait.
   */
  public class RemoteCacheNoWaitFacade<K, V>
      extends AbstractRemoteCacheNoWaitFacade<K, V>
  {
      /** log instance */
      private static final Log log = LogManager.getLog( RemoteCacheNoWaitFacade.class );
  
      /** Provide factory instance to RemoteCacheFailoverRunner */
      private final RemoteCacheFactory cacheFactory;
  
      /** Attempt to restore primary connection (switched off for testing) */
      protected boolean attemptRestorePrimary = true;
  
      /** Time in ms to sleep between failover attempts */
      private static final long idlePeriod = 20000L;
  
      /**
       * Constructs with the given remote cache, and fires events to any listeners.
       * <p>
       * @param noWaits
       * @param rca
       * @param cacheEventLogger
       * @param elementSerializer
       * @param cacheFactory
       */
      public RemoteCacheNoWaitFacade( final List<RemoteCacheNoWait<K,V>> noWaits,
                                      final IRemoteCacheAttributes rca,
                                      final ICacheEventLogger cacheEventLogger,
                                      final IElementSerializer elementSerializer,
                                      final RemoteCacheFactory cacheFactory)
      {
          super( noWaits, rca, cacheEventLogger, elementSerializer );
          this.cacheFactory = cacheFactory;
      }
  
      /**
       * Begin the failover process if this is a local cache. Clustered remote caches do not failover.
       * <p>
       * @param rcnw The no wait in error.
       */
      @Override
      protected void failover( final RemoteCacheNoWait<K, V> rcnw )
      {
          log.debug( "in failover for {0}", rcnw );
  
          if ( getAuxiliaryCacheAttributes().getRemoteType() == RemoteType.LOCAL )
          {
              if ( rcnw.getStatus() == CacheStatus.ERROR )
              {
                  // start failover, primary recovery process
                  final Thread runner = new Thread(this::connectAndRestore);
                  runner.setDaemon( true );
                  runner.start();
  
                  if ( getCacheEventLogger() != null )
                  {
                     getCacheEventLogger().logApplicationEvent( "RemoteCacheNoWaitFacade", "InitiatedFailover",
                                                                rcnw + " was in error." );
                 }
             }
             else
             {
                 log.info( "The noWait is not in error" );
             }
         }
     }
 
     /**
      * The thread tries to establish a connection with a failover
      * server, if any are defined. Once a failover connection is made, it will
      * attempt to replace the failover with the primary remote server.
      * <p>
      * It works by switching out the RemoteCacheNoWait inside the Facade.
      * <p>
      * Client (i.e.) the CompositeCache has reference to a RemoteCacheNoWaitFacade.
      * This facade is created by the RemoteCacheFactory. The factory maintains a set
      * of managers, one for each remote server. Typically, there will only be one
      * manager.
      * <p>
      * If you use multiple remote servers, you may want to set one or more as
      * failovers. If a local cache cannot connect to the primary server, or looses
      * its connection to the primary server, it will attempt to restore that
      * Connection in the background. If failovers are defined, the Failover runner
      * will try to connect to a failover until the primary is restored.
      * If no failovers are defined, this will exit automatically.
      *
      * @since 3.1
      */
     protected void connectAndRestore()
     {
         final IRemoteCacheAttributes rca0 = getAuxiliaryCacheAttributes();
         // Each RemoteCacheManager corresponds to one remote connection.
         final List<RemoteLocation> failovers = rca0.getFailovers();
         // we should probably check to see if there are any failovers,
         // even though the caller should have already.
 
         if ( failovers == null )
         {
             log.warn( "Remote is misconfigured, failovers was null." );
             return;
         }
         if ( failovers.size() == 1 )
         {
             // if there is only the primary, return out of this
             log.info( "No failovers defined, exiting failover runner." );
             return;
         }
 
         final AtomicBoolean allright = new AtomicBoolean(false);
 
         do
         {
             log.info( "Remote cache FAILOVER RUNNING." );
 
             // there is no active listener
             if ( !allright.get() )
             {
                 // Monitor each RemoteCacheManager instance one after the other.
                 final int fidx = rca0.getFailoverIndex();
                 log.debug( "fidx = {0} failovers.size = {1}", rca0::getFailoverIndex, failovers::size);
 
                 // If we don't check the primary, if it gets connected in the
                 // background,
                 // we will disconnect it only to put it right back
                 final ListIterator<RemoteLocation> i = failovers.listIterator(fidx); // + 1; // +1 skips the primary
                 log.debug( "starting at failover i = {0}", i );
 
                 // try them one at a time until successful
                 while (i.hasNext() && !allright.get())
                 {
                     final int failoverIndex = i.nextIndex();
                     final RemoteLocation server = i.next();
                     log.debug("Trying server [{0}] at failover index i = {1}", server, failoverIndex);
 
                     final RemoteCacheAttributes rca = (RemoteCacheAttributes) rca0.clone();
                     rca.setRemoteLocation(server);
                     final RemoteCacheManager rcm = cacheFactory.getManager( rca );
 
                     log.debug( "RemoteCacheAttributes for failover = {0}", rca );
 
                     if (rcm != null)
                     {
                         // add a listener if there are none, need to tell rca
                         // what number it is at
                         final ICache<K, V> ic = rcm.getCache( rca );
                         if ( ic.getStatus() == CacheStatus.ALIVE )
                         {
                             // may need to do this more gracefully
                             log.debug( "resetting no wait" );
                             restorePrimaryServer((RemoteCacheNoWait<K, V>) ic);
                             rca0.setFailoverIndex(failoverIndex);
 
                             log.debug("setting ALLRIGHT to true");
                             if (i.hasPrevious())
                             {
                                 log.debug("Moving to Primary Recovery Mode, failover index = {0}", failoverIndex);
                             }
                             else
                             {
                                 log.debug("No need to connect to failover, the primary server is back up.");
                             }
 
                             allright.set(true);
 
                             log.info( "CONNECTED to host = [{0}]", rca::getRemoteLocation);
                         }
                     }
                 }
             }
             // end if !allright
             // get here if while index >0 and allright, meaning that we are
             // connected to some backup server.
             else
             {
                 log.debug( "ALLRIGHT is true " );
                 log.info( "Failover runner is in primary recovery mode. "
                         + "Failover index = {0} Will now try to reconnect to "
                         + "primary server.", rca0::getFailoverIndex);
             }
 
             // Exit loop if in test mode
             if (allright.get() && !attemptRestorePrimary)
             {
                 break;
             }
 
             boolean primaryRestoredSuccessfully = false;
             // if we are not connected to the primary, try.
             if (rca0.getFailoverIndex() > 0)
             {
                 primaryRestoredSuccessfully = restorePrimary();
                 log.debug( "Primary recovery success state = {0}",
                         primaryRestoredSuccessfully );
             }
 
             if (!primaryRestoredSuccessfully)
             {
                 // Time driven mode: sleep between each round of recovery attempt.
                 try
                 {
                     log.warn( "Failed to reconnect to primary server. "
                             + "Cache failover runner is going to sleep for "
                             + "{0} milliseconds.", idlePeriod );
                     Thread.sleep( idlePeriod );
                 }
                 catch ( final InterruptedException ex )
                 {
                     // ignore;
                 }
             }
 
             // try to bring the listener back to the primary
         }
         while (rca0.getFailoverIndex() > 0 || !allright.get());
         // continue if the primary is not restored or if things are not allright.
 
         if ( log.isInfoEnabled() )
         {
             final int failoverIndex = rca0.getFailoverIndex();
             log.info( "Exiting failover runner. Failover index = {0}", failoverIndex);
 
             if ( failoverIndex <= 0 )
             {
                 log.info( "Failover index is <= 0, meaning we are not connected to a failover server." );
             }
             else
             {
                 log.info( "Failover index is > 0, meaning we are connected to a failover server." );
             }
         }
     }
 
     /**
      * Try to restore the primary server.
      * <p>
      * Once primary is restored the failover listener must be deregistered.
      * <p>
      * The primary server is the first server defines in the FailoverServers
      * list.
      *
      * @return boolean value indicating whether the restoration was successful
      */
     private boolean restorePrimary()
     {
         final IRemoteCacheAttributes rca0 = getAuxiliaryCacheAttributes();
         // try to move back to the primary
         final RemoteLocation server = rca0.getFailovers().get(0);
 
         log.info( "Trying to restore connection to primary remote server "
                 + "[{0}]", server );
 
         final RemoteCacheAttributes rca = (RemoteCacheAttributes) rca0.clone();
         rca.setRemoteLocation(server);
         final RemoteCacheManager rcm = cacheFactory.getManager( rca );
 
         if (rcm != null)
         {
             // add a listener if there are none, need to tell rca what number it
             // is at
             final ICache<K, V> ic = rcm.getCache( rca );
             // by default the listener id should be 0, else it will be the
             // listener
             // Originally associated with the remote cache. either way is fine.
             // We just don't want the listener id from a failover being used.
             // If the remote server was rebooted this could be a problem if new
             // locals were also added.
 
             if ( ic.getStatus() == CacheStatus.ALIVE )
             {
                 try
                 {
                     // we could have more than one listener registered right
                     // now.
                     // this will not result in a loop, only duplication
                     // stop duplicate listening.
                     if (getPrimaryServer() != null && getPrimaryServer().getStatus() == CacheStatus.ALIVE )
                     {
                         final int fidx = rca0.getFailoverIndex();
 
                         if ( fidx > 0 )
                         {
                             final RemoteLocation serverOld = rca0.getFailovers().get(fidx);
 
                             log.debug( "Failover Index = {0} the server at that "
                                     + "index is [{1}]", fidx, serverOld );
 
                             if ( serverOld != null )
                             {
                                 // create attributes that reflect the
                                 // previous failed over configuration.
                                 final RemoteCacheAttributes rcaOld = (RemoteCacheAttributes) rca0.clone();
                                 rcaOld.setRemoteLocation(serverOld);
                                 final RemoteCacheManager rcmOld = cacheFactory.getManager( rcaOld );
 
                                 if ( rcmOld != null )
                                 {
                                     // manager can remove by name if
                                     // necessary
                                     rcmOld.removeRemoteCacheListener( rcaOld );
                                 }
                                 log.info( "Successfully deregistered from "
                                         + "FAILOVER remote server = {0}", serverOld );
                             }
                         }
                         else if ( fidx == 0 )
                         {
                             // this should never happen. If there are no
                             // failovers this shouldn't get called.
                             if ( log.isDebugEnabled() )
                             {
                                 log.debug( "No need to restore primary, it is already restored." );
                                 return true;
                             }
                         }
                         else {
                             // this should never happen
                             log.warn( "Failover index is less than 0, this shouldn't happen" );
                         }
                     }
                 }
                 catch ( final IOException e )
                 {
                     // TODO, should try again, or somehow stop the listener
                     log.error("Trouble trying to deregister old failover "
                             + "listener prior to restoring the primary = {0}",
                             server, e );
                 }
 
                 // Restore primary
                 // may need to do this more gracefully, letting the failover finish in the background
                 final RemoteCacheNoWait<K, V> failoverNoWait = getPrimaryServer();
 
                 // swap in a new one
                 restorePrimaryServer((RemoteCacheNoWait<K, V>) ic);
                 rca0.setFailoverIndex( 0 );
 
                 final String message = "Successfully reconnected to PRIMARY "
                         + "remote server. Substituted primary for "
                         + "failoverNoWait [" + failoverNoWait + "]";
                 log.info( message );
 
                 if (getCacheEventLogger() != null)
                 {
                     getCacheEventLogger().logApplicationEvent(
                             "RemoteCacheFailoverRunner", "RestoredPrimary",
                             message );
                 }
                 return true;
             }
         }
 
         // else all right
         // if the failover index was at 0 here, we would be in a bad
         // situation, unless there were just
         // no failovers configured.
         log.debug( "Primary server status in error, not connected." );
 
         return false;
     }
 }