package org.infinispan.commands.triangle;

import static org.infinispan.commands.write.ValueMatcher.MATCH_ALWAYS;

import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;

import org.infinispan.commands.functional.AbstractWriteKeyCommand;
import org.infinispan.commands.functional.ReadWriteKeyCommand;
import org.infinispan.commands.functional.ReadWriteKeyValueCommand;
import org.infinispan.commands.functional.WriteOnlyKeyCommand;
import org.infinispan.commands.functional.WriteOnlyKeyValueCommand;
import org.infinispan.commands.write.WriteCommand;
import org.infinispan.commons.marshall.MarshallUtil;
import org.infinispan.metadata.Metadata;
import org.infinispan.util.ByteString;

/**
 * A single key {@link BackupWriteCommand} for single key functional commands.
 *
 * @author Pedro Ruivo
 * @since 9.0
 */
public class SingleKeyFunctionalBackupWriteCommand extends FunctionalBackupWriteCommand {

   public static final byte COMMAND_ID = 77;
   private static final Operation[] CACHED_OPERATION = Operation.values();

   private Operation operation;
   private Object key;
   private Object value;
   private Object prevValue;
   private Metadata prevMetadata;

   //for testing
   @SuppressWarnings("unused")
   public SingleKeyFunctionalBackupWriteCommand() {
      super(null);
   }

   public SingleKeyFunctionalBackupWriteCommand(ByteString cacheName) {
      super(cacheName);
   }

   private static Operation valueOf(int index) {
      return CACHED_OPERATION[index];
   }

   @Override
   public byte getCommandId() {
      return COMMAND_ID;
   }

   public void setReadWriteKeyCommand(ReadWriteKeyCommand command) {
      this.operation = Operation.READ_WRITE;
      setCommonFields(command);
      this.function = command.getFunction();
   }

   public void setReadWriteKeyValueCommand(ReadWriteKeyValueCommand command) {
      this.operation = Operation.READ_WRITE_KEY_VALUE;
      setCommonFields(command);
      this.function = command.getBiFunction();
      this.value = command.getArgument();
      this.prevValue = command.getPrevValue();
      this.prevMetadata = command.getPrevMetadata();
   }

   public void setWriteOnlyKeyValueCommand(WriteOnlyKeyValueCommand command) {
      this.operation = Operation.WRITE_ONLY_KEY_VALUE;
      setCommonFields(command);
      this.function = command.getBiConsumer();
      this.value = command.getArgument();
   }

   public void setWriteOnlyKeyCommand(WriteOnlyKeyCommand command) {
      this.operation = Operation.WRITE_ONLY;
      setCommonFields(command);
      this.function = command.getConsumer();
   }

   @Override
   public void writeTo(ObjectOutput output) throws IOException {
      writeBase(output);
      writeFunctionAndParams(output);
      MarshallUtil.marshallEnum(operation, output);
      output.writeObject(key);
      switch (operation) {
         case READ_WRITE_KEY_VALUE:
            output.writeObject(prevValue);
            output.writeObject(prevMetadata);
         case WRITE_ONLY_KEY_VALUE:
            output.writeObject(value);
         case READ_WRITE:
         case WRITE_ONLY:
         default:
      }
   }

   @Override
   public void readFrom(ObjectInput input) throws IOException, ClassNotFoundException {
      readBase(input);
      readFunctionAndParams(input);
      operation = MarshallUtil.unmarshallEnum(input, SingleKeyFunctionalBackupWriteCommand::valueOf);
      key = input.readObject();
      switch (operation) {
         case READ_WRITE_KEY_VALUE:
            prevValue = input.readObject();
            prevMetadata = (Metadata) input.readObject();
         case WRITE_ONLY_KEY_VALUE:
            value = input.readObject();
         case READ_WRITE:
         case WRITE_ONLY:
         default:
      }
   }

   @Override
   WriteCommand createWriteCommand() {
      switch (operation) {
         case READ_WRITE:
            //noinspection unchecked
            return new ReadWriteKeyCommand(key, (Function) function, segmentId, getCommandInvocationId(), MATCH_ALWAYS,
                  params, keyDataConversion, valueDataConversion);
         case READ_WRITE_KEY_VALUE:
            //noinspection unchecked
            ReadWriteKeyValueCommand cmd = new ReadWriteKeyValueCommand(key, value, (BiFunction) function, segmentId,
                  getCommandInvocationId(), MATCH_ALWAYS, params, keyDataConversion, valueDataConversion);
            cmd.setPrevValueAndMetadata(prevValue, prevMetadata);
            return cmd;
         case WRITE_ONLY:
            //noinspection unchecked
            return new WriteOnlyKeyCommand(key, (Consumer) function, segmentId, getCommandInvocationId(), MATCH_ALWAYS,
                  params, keyDataConversion, valueDataConversion);
         case WRITE_ONLY_KEY_VALUE:
            //noinspection unchecked
            return new WriteOnlyKeyValueCommand(key, value, (BiConsumer) function, segmentId, getCommandInvocationId(),
                  MATCH_ALWAYS, params, keyDataConversion, valueDataConversion);
         default:
            throw new IllegalStateException("Unknown operation " + operation);
      }
   }

   private <C extends AbstractWriteKeyCommand> void setCommonFields(C command) {
      setCommonAttributesFromCommand(command);
      setFunctionalCommand(command);
      this.key = command.getKey();
   }

   private enum Operation {
      READ_WRITE_KEY_VALUE,
      READ_WRITE,
      WRITE_ONLY_KEY_VALUE,
      WRITE_ONLY
   }
}
