package org.infinispan.remoting.inboundhandler;

import static org.infinispan.factories.KnownComponentNames.BLOCKING_EXECUTOR;
import static org.infinispan.factories.KnownComponentNames.NON_BLOCKING_EXECUTOR;
import static org.infinispan.remoting.inboundhandler.BasePerCacheInboundInvocationHandler.MBEAN_COMPONENT_NAME;
import static org.infinispan.util.logging.Log.CLUSTER;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.LongAdder;

import org.infinispan.commands.ReplicableCommand;
import org.infinispan.commands.TopologyAffectedCommand;
import org.infinispan.commands.remote.CacheRpcCommand;
import org.infinispan.commands.remote.ClusteredGetAllCommand;
import org.infinispan.commands.remote.ClusteredGetCommand;
import org.infinispan.commands.remote.SingleRpcCommand;
import org.infinispan.commons.CacheException;
import org.infinispan.configuration.cache.Configuration;
import org.infinispan.factories.ComponentRegistry;
import org.infinispan.factories.annotations.ComponentName;
import org.infinispan.factories.annotations.Inject;
import org.infinispan.factories.annotations.Start;
import org.infinispan.factories.annotations.Stop;
import org.infinispan.factories.scopes.Scope;
import org.infinispan.factories.scopes.Scopes;
import org.infinispan.jmx.annotations.DataType;
import org.infinispan.jmx.annotations.MBean;
import org.infinispan.jmx.annotations.ManagedAttribute;
import org.infinispan.jmx.annotations.ManagedOperation;
import org.infinispan.remoting.inboundhandler.action.ReadyAction;
import org.infinispan.remoting.responses.CacheNotFoundResponse;
import org.infinispan.remoting.responses.ExceptionResponse;
import org.infinispan.remoting.responses.Response;
import org.infinispan.remoting.responses.ResponseGenerator;
import org.infinispan.statetransfer.OutdatedTopologyException;
import org.infinispan.statetransfer.StateTransferLock;
import org.infinispan.util.concurrent.BlockingRunnable;
import org.infinispan.util.concurrent.BlockingTaskAwareExecutorService;
import org.infinispan.util.concurrent.CompletableFutures;
import org.infinispan.util.concurrent.CompletionStages;
import org.infinispan.util.logging.Log;
import org.infinispan.util.logging.LogFactory;

/**
 * Implementation with the default handling methods and utilities methods.
 *
 * @author Pedro Ruivo
 * @since 7.1
 */
@Scope(Scopes.NAMED_CACHE)
@MBean(objectName = MBEAN_COMPONENT_NAME, description = "Handles all the remote requests.")
public abstract class BasePerCacheInboundInvocationHandler implements PerCacheInboundInvocationHandler {
   private static final Log log = LogFactory.getLog(BasePerCacheInboundInvocationHandler.class);

   public static final String MBEAN_COMPONENT_NAME = "InboundInvocationHandler";
   private static final int NO_TOPOLOGY_COMMAND = Integer.MIN_VALUE;

   // TODO: To be removed with https://issues.redhat.com/browse/ISPN-11483
   @Inject @ComponentName(BLOCKING_EXECUTOR)
   protected BlockingTaskAwareExecutorService blockingExecutor;
   @Inject @ComponentName(NON_BLOCKING_EXECUTOR)
   protected BlockingTaskAwareExecutorService nonBlockingExecutor;
   @Inject StateTransferLock stateTransferLock;
   @Inject ResponseGenerator responseGenerator;
   @Inject ComponentRegistry componentRegistry;
   @Inject protected Configuration configuration;

   private volatile boolean stopped = false;
   private volatile int firstTopologyAsMember = Integer.MAX_VALUE;

   private final LongAdder xSiteReceived = new LongAdder();
   private volatile boolean statisticsEnabled = false;

   private static int extractCommandTopologyId(SingleRpcCommand command) {
      ReplicableCommand innerCmd = command.getCommand();
      if (innerCmd instanceof TopologyAffectedCommand) {
         return ((TopologyAffectedCommand) innerCmd).getTopologyId();
      }
      return NO_TOPOLOGY_COMMAND;
   }

   static int extractCommandTopologyId(CacheRpcCommand command) {
      switch (command.getCommandId()) {
         case SingleRpcCommand.COMMAND_ID:
            return extractCommandTopologyId((SingleRpcCommand) command);
         case ClusteredGetCommand.COMMAND_ID:
         case ClusteredGetAllCommand.COMMAND_ID:
            // These commands are topology aware but we don't block them here - topologyId logic
            // is handled in StateTransferInterceptor
            return NO_TOPOLOGY_COMMAND;
         default:
            if (command instanceof TopologyAffectedCommand) {
               return ((TopologyAffectedCommand) command).getTopologyId();
            }
      }
      return NO_TOPOLOGY_COMMAND;
   }

   @Start
   public void start() {
      this.stopped = false;
      setStatisticsEnabled(configuration.statistics().enabled());
   }

   @Stop
   public void stop() {
      this.stopped = true;
   }

   public boolean isStopped() {
      return stopped;
   }

   final CompletableFuture<Response> invokeCommand(CacheRpcCommand cmd) throws Throwable {
      if (log.isTraceEnabled()) {
         log.tracef("Calling perform() on %s", cmd);
      }
      CompletableFuture<?> future = cmd.invokeAsync(componentRegistry).toCompletableFuture();
      if (CompletionStages.isCompletedSuccessfully(future)) {
         Object obj = future.join();
         Response response = responseGenerator.getResponse(cmd, obj);
         if (response == null) {
            return CompletableFutures.completedNull();
         }
         return CompletableFuture.completedFuture(response);
      }
      return future.handle((rv, throwable) -> {
         CompletableFutures.rethrowExceptionIfPresent(throwable);

         return responseGenerator.getResponse(cmd, rv);
      });
   }

   final StateTransferLock getStateTransferLock() {
      return stateTransferLock;
   }

   final ExceptionResponse exceptionHandlingCommand(CacheRpcCommand command, Throwable throwable) {
      CLUSTER.exceptionHandlingCommand(command, throwable);
      if (throwable instanceof Exception) {
         return new ExceptionResponse(((Exception) throwable));
      } else {
         return new ExceptionResponse(new CacheException("Problems invoking command.", throwable));
      }
   }

   final ExceptionResponse outdatedTopology(OutdatedTopologyException exception) {
      log.tracef("Topology changed, retrying: %s", exception);
      return new ExceptionResponse(exception);
   }

   final Response interruptedException(CacheRpcCommand command) {
      CLUSTER.debugf("Shutdown while handling command %s", command);
      return CacheNotFoundResponse.INSTANCE;
   }

   final void unexpectedDeliverMode(ReplicableCommand command, DeliverOrder deliverOrder) {
      throw new IllegalArgumentException(String.format("Unexpected deliver mode %s for command%s", deliverOrder, command));
   }

   final void handleRunnable(BlockingRunnable runnable, boolean onExecutorService) {
      // This means it is blocking and not preserve order per executeOnExecutorService
      if (onExecutorService) {
         blockingExecutor.execute(runnable);
      } else {
         runnable.run();
      }
   }

   public final boolean isCommandSentBeforeFirstTopology(int commandTopologyId) {
      if (0 <= commandTopologyId && commandTopologyId < firstTopologyAsMember) {
         if (log.isTraceEnabled()) {
            log.tracef("Ignoring command sent before the local node was a member (command topology id is %d, first topology as member is %d)", commandTopologyId, firstTopologyAsMember);
         }
         return true;
      }
      return false;
   }

   final BlockingRunnable createDefaultRunnable(CacheRpcCommand command, Reply reply, int commandTopologyId,
         boolean waitTransactionalData, boolean onExecutorService,
         boolean sync) {
      return new DefaultTopologyRunnable(this, command, reply,
                                         TopologyMode.create(onExecutorService, waitTransactionalData),
                                         commandTopologyId, sync);
   }

   final BlockingRunnable createDefaultRunnable(final CacheRpcCommand command, final Reply reply,
         final int commandTopologyId, TopologyMode topologyMode,
         boolean sync) {
      return new DefaultTopologyRunnable(this, command, reply, topologyMode, commandTopologyId, sync);
   }

   final boolean executeOnExecutorService(DeliverOrder order, CacheRpcCommand command) {
      return !order.preserveOrder() && command.canBlock();
   }

   final BlockingRunnable createReadyActionRunnable(CacheRpcCommand command, Reply reply, int commandTopologyId,
         boolean sync, ReadyAction readyAction) {
      if (readyAction != null) {
         return createNonNullReadyActionRunnable(command, reply, commandTopologyId, sync, readyAction);
      } else {
         return new DefaultTopologyRunnable(this, command, reply, TopologyMode.READY_TX_DATA, commandTopologyId, sync);
      }
   }

   @Override
   public void registerXSiteCommandReceiver() {
      if (statisticsEnabled) {
         xSiteReceived.increment();
      }
   }

   @Override
   public boolean getStatisticsEnabled() {
      return isStatisticsEnabled();
   }

   @Override
   @ManagedOperation(description = "Resets statistics gathered by this component", displayName = "Reset statistics")
   public void resetStatistics() {
      xSiteReceived.reset();
   }

   @ManagedAttribute(description = "Enables or disables the gathering of statistics by this component",
         displayName = "Statistics enabled",
         dataType = DataType.TRAIT,
         writable = true)
   public boolean isStatisticsEnabled() {
      return statisticsEnabled;
   }

   @Override
   public void setStatisticsEnabled(boolean enabled) {
      this.statisticsEnabled = enabled;
   }

   @ManagedAttribute(description = "Returns the number of cross-site requests received by this node",
         displayName = "Cross-Site Requests Received")
   public long getXSiteRequestsReceived() {
      return statisticsEnabled ? xSiteReceived.sum() : 0;
   }

   private BlockingRunnable createNonNullReadyActionRunnable(CacheRpcCommand command, Reply reply, int commandTopologyId, boolean sync, ReadyAction readyAction) {
      readyAction.addListener(this::checkForReadyTasks);
      return new DefaultTopologyRunnable(this, command, reply, TopologyMode.READY_TX_DATA, commandTopologyId, sync) {
         @Override
         public boolean isReady() {
            return super.isReady() && readyAction.isReady();
         }

         @Override
         protected void onException(Throwable throwable) {
            super.onException(throwable);
            readyAction.onException();
         }

         @Override
         protected void onFinally() {
            super.onFinally();
            readyAction.onFinally();
         }
      };
   }

   @Override
   public void setFirstTopologyAsMember(int firstTopologyAsMember) {
      this.firstTopologyAsMember = firstTopologyAsMember;
   }

   @Override
   public int getFirstTopologyAsMember() {
         return firstTopologyAsMember;
   }

   @Override
   public void checkForReadyTasks() {
      blockingExecutor.checkForReadyTasks();
      nonBlockingExecutor.checkForReadyTasks();
   }
}
