/*
 * 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.
 */

package org.apache.geode.internal.cache.wan;

import static org.apache.geode.internal.cache.wan.GatewaySenderEventImpl.TransactionMetadataDisposition.EXCLUDE;
import static org.apache.geode.internal.cache.wan.GatewaySenderEventImpl.TransactionMetadataDisposition.INCLUDE_LAST_EVENT;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Arrays;

import org.apache.geode.DataSerializer;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.cache.CacheEvent;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.asyncqueue.AsyncEvent;
import org.apache.geode.cache.util.ObjectSizer;
import org.apache.geode.cache.wan.EventSequenceID;
import org.apache.geode.internal.cache.CachedDeserializable;
import org.apache.geode.internal.cache.CachedDeserializableFactory;
import org.apache.geode.internal.cache.Conflatable;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.EnumListenerEvent;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.Token;
import org.apache.geode.internal.cache.WrappedCallbackArgument;
import org.apache.geode.internal.cache.tier.sockets.Message;
import org.apache.geode.internal.lang.ObjectUtils;
import org.apache.geode.internal.offheap.OffHeapHelper;
import org.apache.geode.internal.offheap.ReferenceCountHelper;
import org.apache.geode.internal.offheap.Releasable;
import org.apache.geode.internal.offheap.StoredObject;
import org.apache.geode.internal.offheap.annotations.OffHeapIdentifier;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.offheap.annotations.Unretained;
import org.apache.geode.internal.serialization.DataSerializableFixedID;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.size.Sizeable;

/**
 * Class <code>GatewaySenderEventImpl</code> represents an event sent between
 * <code>GatewaySender</code>
 *
 *
 * @since GemFire 7.0
 *
 */
public class GatewaySenderEventImpl
    implements AsyncEvent, DataSerializableFixedID, Conflatable, Sizeable, Releasable {
  private static final long serialVersionUID = -5690172020872255422L;
  protected static final Object TOKEN_NULL = new Object();

  // It should use current version. But it was hard-coded to be 0x11, i.e. GEODE_120_ORDINAL,
  // by mistake since 120 to pre-190
  protected static final short VERSION = KnownVersion.getCurrentVersion().ordinal();

  protected EnumListenerEvent operation;

  protected Object substituteValue;

  /**
   * The action to be taken (e.g. AFTER_CREATE)
   */
  protected int action;

  /**
   * The operation detail of EntryEvent (e.g. LOAD, PUTALL etc.)
   */
  protected int operationDetail;

  /**
   * The number of parts for the <code>Message</code>
   *
   * @see Message
   */
  protected int numberOfParts;

  /**
   * The identifier of this event
   */
  protected EventID id;

  /**
   * The <code>Region</code> that was updated
   */
  private transient LocalRegion region;

  /**
   * The name of the region being affected by this event
   */
  protected String regionPath;

  /**
   * The key being affected by this event
   */
  protected Object key;

  /**
   * The serialized new value for this event's key. May not be computed at construction time.
   */
  protected volatile byte[] value;

  /**
   * The "object" form of the value. Will be null after this object is deserialized.
   */
  @Retained(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
  protected transient Object valueObj;
  protected transient boolean valueObjReleased;

  private transient boolean serializedValueNotAvailable;

  /**
   * Whether the value is a serialized object or just a byte[]
   */
  protected byte valueIsObject;

  /**
   * The callback argument for this event
   */
  protected GatewaySenderEventCallbackArgument callbackArgument;

  /**
   * The version timestamp
   */
  protected long versionTimeStamp;

  /**
   * Whether this event is a possible duplicate
   */
  protected boolean possibleDuplicate;

  /**
   * Whether this event is acknowledged after the ack received by AckReaderThread. As of now this is
   * getting used for PDX related GatewaySenderEvent. But can be extended for for other
   * GatewaySenderEvent.
   */
  protected volatile boolean isAcked;

  /**
   * Whether this event is dispatched by dispatcher. As of now this is getting used for PDX related
   * GatewaySenderEvent. But can be extended for for other GatewaySenderEvent.
   */
  protected volatile boolean isDispatched;
  /**
   * The creation timestamp in ms
   */
  protected long creationTime;

  /**
   * For ParalledGatewaySender we need bucketId of the PartitionRegion on which the update operation
   * was applied.
   */
  protected int bucketId;

  protected Long shadowKey = -1L;

  protected boolean isInitialized;

  private transient boolean isConcurrencyConflict = false;

  private short version;

  private boolean isLastEventInTransaction = false;

  private TransactionId transactionId = null;


  /**
   * Is this thread in the process of serializing this event?
   */
  public static final ThreadLocal<Boolean> isSerializingValue =
      ThreadLocal.withInitial(() -> Boolean.FALSE);

  private static final int CREATE_ACTION = 0;

  private static final int UPDATE_ACTION = 1;

  private static final int DESTROY_ACTION = 2;

  private static final int VERSION_ACTION = 3;

  public static final int UPDATE_ACTION_NO_GENERATE_CALLBACKS = 4;

  private static final int INVALIDATE_ACTION = 5;
  /**
   * Static constants for Operation detail of EntryEvent.
   */
  private static final int OP_DETAIL_NONE = 10;

  private static final int OP_DETAIL_LOCAL_LOAD = 11;

  private static final int OP_DETAIL_NET_LOAD = 12;

  private static final int OP_DETAIL_PUTALL = 13;

  private static final int OP_DETAIL_REMOVEALL = 14;

  private static final int DEFAULT_SERIALIZED_VALUE_SIZE = -1;

  private volatile int serializedValueSize = DEFAULT_SERIALIZED_VALUE_SIZE;


  /**
   * No-arg constructor for data serialization.
   *
   * @see DataSerializer
   */
  public GatewaySenderEventImpl() {}

  /**
   * @param operation The operation for this event (e.g. AFTER_CREATE)
   * @param event The <code>CacheEvent</code> on which this <code>GatewayEventImpl</code> is based
   * @param substituteValue The value to be enqueued instead of the value in the event.
   * @param transactionMetadataDisposition indicating the inclusion of transaction metadata.
   *
   */
  @Retained
  public GatewaySenderEventImpl(EnumListenerEvent operation, CacheEvent<?, ?> event,
      Object substituteValue, final TransactionMetadataDisposition transactionMetadataDisposition)
      throws IOException {
    this(operation, event, substituteValue, true, transactionMetadataDisposition);
  }

  @Retained
  public GatewaySenderEventImpl(EnumListenerEvent operation, CacheEvent<?, ?> event,
      Object substituteValue) throws IOException {
    this(operation, event, substituteValue, true, EXCLUDE);
  }

  @Retained
  public GatewaySenderEventImpl(EnumListenerEvent operation, CacheEvent<?, ?> event,
      Object substituteValue, boolean initialize, int bucketId,
      final TransactionMetadataDisposition transactionMetadataDisposition) throws IOException {
    this(operation, event, substituteValue, initialize, transactionMetadataDisposition);
    this.bucketId = bucketId;
  }

  /**
   * @param operation The operation for this event (e.g. AFTER_CREATE)
   * @param ce The <code>CacheEvent</code> on which this <code>GatewayEventImpl</code> is based
   * @param substituteValue The value to be enqueued instead of the value in the event.
   * @param initialize Whether to initialize this instance
   * @param transactionMetadataDisposition indicating the inclusion of transaction metadata.
   */
  @Retained
  public GatewaySenderEventImpl(EnumListenerEvent operation, CacheEvent<?, ?> ce,
      Object substituteValue, boolean initialize,
      final TransactionMetadataDisposition transactionMetadataDisposition) throws IOException {
    // Set the operation and event
    final EntryEventImpl event = (EntryEventImpl) ce;
    this.operation = operation;
    this.substituteValue = substituteValue;

    // Initialize the region name. This is being done here because the event
    // can get serialized/deserialized (for some reason) between the time
    // it is set above and used (in initialize). If this happens, the
    // region is null because it is a transient field of the event.
    region = (LocalRegion) event.getRegion();
    regionPath = region.getFullPath();

    // Initialize the unique id
    initializeId(event);

    // Initialize possible duplicate
    possibleDuplicate = event.isPossibleDuplicate();

    // Initialize ack and dispatch status of events
    isAcked = false;
    isDispatched = false;


    // Initialize the creation timestamp
    creationTime = System.currentTimeMillis();

    if (event.getVersionTag() != null && event.getVersionTag().hasValidVersion()) {
      versionTimeStamp = event.getVersionTag().getVersionTimeStamp();
    }

    // Set key
    // System.out.println("this._entryEvent: " + event);
    // System.out.println("this._entryEvent.getKey(): " +
    // event.getKey());
    key = event.getKey();

    initializeValue(event);

    // Set the callback arg
    callbackArgument = (GatewaySenderEventCallbackArgument) event.getRawCallbackArgument();

    // Initialize the action and number of parts (called after _callbackArgument
    // is set above)
    initializeAction(this.operation, event);

    // initialize the operation detail
    initializeOperationDetail(event.getOperation());

    setShadowKey(event.getTailKey());

    if (initialize) {
      initialize();
    }
    isConcurrencyConflict = event.isConcurrencyConflict();

    if (transactionMetadataDisposition != EXCLUDE) {
      transactionId = event.getTransactionId();
      isLastEventInTransaction =
          transactionMetadataDisposition == INCLUDE_LAST_EVENT && null != transactionId;
    }
  }

  /**
   * Used to create a heap copy of an offHeap event. Note that this constructor produces an instance
   * that does not need to be released.
   */
  protected GatewaySenderEventImpl(GatewaySenderEventImpl offHeapEvent) {
    operation = offHeapEvent.operation;
    action = offHeapEvent.action;
    numberOfParts = offHeapEvent.numberOfParts;
    id = offHeapEvent.id;
    region = offHeapEvent.region;
    regionPath = offHeapEvent.regionPath;
    key = offHeapEvent.key;
    callbackArgument = offHeapEvent.callbackArgument;
    versionTimeStamp = offHeapEvent.versionTimeStamp;
    possibleDuplicate = offHeapEvent.possibleDuplicate;
    isAcked = offHeapEvent.isAcked;
    isDispatched = offHeapEvent.isDispatched;
    creationTime = offHeapEvent.creationTime;
    bucketId = offHeapEvent.bucketId;
    shadowKey = offHeapEvent.shadowKey;
    isInitialized = offHeapEvent.isInitialized;

    valueObj = null;
    valueObjReleased = false;
    valueIsObject = offHeapEvent.valueIsObject;
    value = offHeapEvent.getSerializedValue();
    transactionId = offHeapEvent.transactionId;
    isLastEventInTransaction = offHeapEvent.isLastEventInTransaction;
  }

  /**
   * Returns this event's action
   *
   * @return this event's action
   */
  public int getAction() {
    return action;
  }

  /**
   * Returns this event's operation
   *
   * @return this event's operation
   */
  @Override
  public Operation getOperation() {
    Operation op = null;
    switch (action) {
      case CREATE_ACTION:
        switch (operationDetail) {
          case OP_DETAIL_LOCAL_LOAD:
            op = Operation.LOCAL_LOAD_CREATE;
            break;
          case OP_DETAIL_NET_LOAD:
            op = Operation.NET_LOAD_CREATE;
            break;
          case OP_DETAIL_PUTALL:
            op = Operation.PUTALL_CREATE;
            break;
          case OP_DETAIL_NONE:
            op = Operation.CREATE;
            break;
          // if operationDetail is none of the above, then default should be NONE
          default:
            op = Operation.CREATE;
            break;
        }
        break;
      case UPDATE_ACTION:
        switch (operationDetail) {
          case OP_DETAIL_LOCAL_LOAD:
            op = Operation.LOCAL_LOAD_UPDATE;
            break;
          case OP_DETAIL_NET_LOAD:
            op = Operation.NET_LOAD_UPDATE;
            break;
          case OP_DETAIL_PUTALL:
            op = Operation.PUTALL_UPDATE;
            break;
          case OP_DETAIL_NONE:
            op = Operation.UPDATE;
            break;
          // if operationDetail is none of the above, then default should be NONE
          default:
            op = Operation.UPDATE;
            break;
        }
        break;
      case DESTROY_ACTION:
        if (operationDetail == OP_DETAIL_REMOVEALL) {
          op = Operation.REMOVEALL_DESTROY;
        } else {
          op = Operation.DESTROY;
        }
        break;
      case VERSION_ACTION:
        op = Operation.UPDATE_VERSION_STAMP;
        break;
      case INVALIDATE_ACTION:
        op = Operation.INVALIDATE;
        break;
    }
    return op;
  }

  public Object getSubstituteValue() {
    return substituteValue;
  }

  public EnumListenerEvent getEnumListenerEvent() {
    return operation;
  }

  /**
   * Return this event's region name
   *
   * @return this event's region name
   */
  public String getRegionPath() {
    return regionPath;
  }

  public boolean isInitialized() {
    return isInitialized;
  }

  /**
   * Returns this event's key
   *
   * @return this event's key
   */
  @Override
  public Object getKey() {
    // TODO:Asif : Ideally would like to have throw exception if the key
    // is TOKEN_UN_INITIALIZED, but for the time being trying to retain the GFE
    // behaviour
    // of returning null if getKey is invoked on un-initialized gateway event
    return isInitialized() ? key : null;
  }

  /**
   * Returns whether this event's value is a serialized object
   *
   * @return whether this event's value is a serialized object
   */
  public byte getValueIsObject() {
    return valueIsObject;
  }

  /**
   * Return this event's callback argument
   *
   * @return this event's callback argument
   */
  @Override
  public Object getCallbackArgument() {
    Object result = getSenderCallbackArgument();
    while (result instanceof WrappedCallbackArgument) {
      WrappedCallbackArgument wca = (WrappedCallbackArgument) result;
      result = wca.getOriginalCallbackArg();
    }
    return result;
  }

  public GatewaySenderEventCallbackArgument getSenderCallbackArgument() {
    return callbackArgument;
  }

  /**
   * Return this event's number of parts
   *
   * @return this event's number of parts
   */
  public int getNumberOfParts() {
    return numberOfParts;
  }

  /**
   * Return the currently held form of the object. May return a retained OFF_HEAP_REFERENCE.
   */
  @Retained
  public Object getRawValue() {
    @Retained(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
    Object result = value;
    if (result == null) {
      result = substituteValue;
      if (result == null) {
        result = valueObj;
        if (result instanceof StoredObject && ((StoredObject) result).hasRefCount()) {
          if (valueObjReleased) {
            result = null;
          } else {
            StoredObject ohref = (StoredObject) result;
            if (!ohref.retain()) {
              result = null;
            } else if (valueObjReleased) {
              ohref.release();
              result = null;
            }
          }
        }
      }
    }
    return result;
  }

  /**
   * Return this event's deserialized value
   *
   * @return this event's deserialized value
   */
  @Override
  public Object getDeserializedValue() {
    if (valueIsObject == 0x00) {
      Object result = value;
      if (result == null) {
        @Unretained(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
        Object so = valueObj;
        if (valueObjReleased) {
          throw new IllegalStateException(
              "Value is no longer available. getDeserializedValue must be called before processEvents returns.");
        }
        if (so instanceof StoredObject) {
          return ((StoredObject) so).getValueAsDeserializedHeapObject();
        } else {
          throw new IllegalStateException(
              "expected valueObj field to be an instance of StoredObject but it was " + so);
        }
      }
      return result;
    } else {
      Object vo = valueObj;
      if (vo != null) {
        if (vo instanceof StoredObject) {
          @Unretained(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
          StoredObject so = (StoredObject) vo;
          return so.getValueAsDeserializedHeapObject();
        } else {
          return vo; // it is already deserialized
        }
      } else {
        if (value != null) {
          Object result = EntryEventImpl.deserialize(value);
          valueObj = result;
          return result;
        } else if (substituteValue != null) {
          // If the substitute value is set, return it.
          return substituteValue;
        } else {
          if (valueObjReleased) {
            throw new IllegalStateException(
                "Value is no longer available. getDeserializedValue must be called before processEvents returns.");
          }
          // both value and valueObj are null but we did not free it.
          return null;
        }
      }
    }
  }

  /**
   * Returns the value in the form of a String. This should be used by code that wants to log the
   * value. This is a debugging exception.
   */
  public String getValueAsString(boolean deserialize) {
    Object v = value;
    if (v == null) {
      v = substituteValue;
    }
    if (deserialize) {
      try {
        v = getDeserializedValue();
      } catch (Exception | InternalGemFireError e) {
        return "Could not convert value to string because " + e;
      }
    }
    if (v == null) {
      @Unretained(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
      Object ov = valueObj;
      if (ov instanceof CachedDeserializable) {
        return ((CachedDeserializable) ov).getStringForm();
      }
    }
    if (v != null) {
      if (v instanceof byte[]) {
        byte[] bav = (byte[]) v;
        // Using Arrays.toString(bav) can cause us to run out of memory
        return "byte[" + bav.length + "]";
      } else {
        String valueStr;
        try {
          valueStr = v.toString();
        } catch (Exception e) {
          valueStr = "Could not convert value to string because " + e;
        }
        return valueStr;
      }
    } else {
      return "";
    }
  }

  public boolean isSerializedValueNotAvailable() {
    return serializedValueNotAvailable;
  }

  /**
   * If the value owned of this event is just bytes return that byte array; otherwise serialize the
   * value object and return the serialized bytes. Use {@link #getValueIsObject()} to determine if
   * the result is raw or serialized bytes.
   */
  @Override
  public byte[] getSerializedValue() {
    byte[] result = value;
    if (result == null) {
      if (substituteValue != null) {
        // The substitute value is set. Serialize it
        isSerializingValue.set(Boolean.TRUE);
        result = EntryEventImpl.serialize(substituteValue);
        isSerializingValue.set(Boolean.FALSE);
        return result;
      }
      @Unretained(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
      Object vo = valueObj;
      if (vo instanceof StoredObject) {
        synchronized (this) {
          result = value;
          if (result == null) {
            StoredObject so = (StoredObject) vo;
            result = so.getValueAsHeapByteArray();
            value = result;
          }
        }
      } else {
        synchronized (this) {
          result = value;
          if (result == null && vo != null && !(vo instanceof Token)) {
            isSerializingValue.set(Boolean.TRUE);
            result = EntryEventImpl.serialize(vo);
            isSerializingValue.set(Boolean.FALSE);
            value = result;
          } else if (result == null) {
            if (valueObjReleased) {
              serializedValueNotAvailable = true;
              throw new IllegalStateException(
                  "Value is no longer available. getSerializedValue must be called before processEvents returns.");
            }
          }
        }
      }
    }
    return result;
  }

  public void setPossibleDuplicate(boolean possibleDuplicate) {
    this.possibleDuplicate = possibleDuplicate;
  }

  @Override
  public boolean getPossibleDuplicate() {
    return possibleDuplicate;
  }

  public long getCreationTime() {
    return creationTime;
  }

  @Override
  public int getDSFID() {
    return GATEWAY_SENDER_EVENT_IMPL;
  }

  @Override
  public void toData(DataOutput out,
      SerializationContext context) throws IOException {
    toDataPre_GEODE_1_15_0_0(out, context);
    out.writeInt(operationDetail);
  }

  public void toDataPre_GEODE_1_15_0_0(DataOutput out,
      SerializationContext context) throws IOException {
    toDataPre_GEODE_1_14_0_0(out, context);
    boolean hasTransaction = transactionId != null;
    DataSerializer.writeBoolean(hasTransaction, out);
    if (hasTransaction) {
      DataSerializer.writeBoolean(isLastEventInTransaction, out);
      context.getSerializer().writeObject(transactionId, out);
    }
  }

  public void toDataPre_GEODE_1_14_0_0(DataOutput out,
      SerializationContext context) throws IOException {
    toDataPre_GEODE_1_9_0_0(out, context);
    DataSerializer.writeBoolean(isConcurrencyConflict, out);
  }

  public void toDataPre_GEODE_1_9_0_0(DataOutput out, SerializationContext context)
      throws IOException {
    // Make sure we are initialized before we serialize.
    initialize();
    out.writeShort(VERSION);
    out.writeInt(action);
    out.writeInt(numberOfParts);
    // out.writeUTF(this._id);
    context.getSerializer().writeObject(id, out);
    DataSerializer.writeString(regionPath, out);
    out.writeByte(valueIsObject);
    serializeKey(out, context);
    DataSerializer.writeByteArray(getSerializedValue(), out);
    context.getSerializer().writeObject(callbackArgument, out);
    out.writeBoolean(possibleDuplicate);
    out.writeLong(creationTime);
    out.writeInt(bucketId);
    out.writeLong(shadowKey);
    out.writeLong(getVersionTimeStamp());
  }

  protected void serializeKey(DataOutput out,
      SerializationContext context) throws IOException {
    context.getSerializer().writeObject(key, out);
  }

  @Override
  public void fromData(DataInput in,
      DeserializationContext context) throws IOException, ClassNotFoundException {
    fromDataPre_GEODE_1_15_0_0(in, context);
    if (version >= KnownVersion.GEODE_1_15_0.ordinal()) {
      operationDetail = in.readInt();
    }
  }

  public void fromDataPre_GEODE_1_15_0_0(DataInput in, DeserializationContext context)
      throws IOException, ClassNotFoundException {
    fromDataPre_GEODE_1_14_0_0(in, context);
    if (version >= KnownVersion.GEODE_1_14_0.ordinal()) {
      boolean hasTransaction = DataSerializer.readBoolean(in);
      if (hasTransaction) {
        isLastEventInTransaction = DataSerializer.readBoolean(in);
        transactionId = context.getDeserializer().readObject(in);
      }
    }
  }

  public void fromDataPre_GEODE_1_14_0_0(DataInput in, DeserializationContext context)
      throws IOException, ClassNotFoundException {
    fromDataPre_GEODE_1_9_0_0(in, context);
    if (version >= KnownVersion.GEODE_1_9_0.ordinal()) {
      isConcurrencyConflict = DataSerializer.readBoolean(in);
    }
  }

  public void fromDataPre_GEODE_1_9_0_0(DataInput in, DeserializationContext context)
      throws IOException, ClassNotFoundException {
    version = in.readShort();
    isInitialized = true;
    action = in.readInt();
    numberOfParts = in.readInt();
    id = context.getDeserializer().readObject(in);
    regionPath = DataSerializer.readString(in);
    valueIsObject = in.readByte();
    deserializeKey(in, context);
    value = DataSerializer.readByteArray(in);
    callbackArgument = context.getDeserializer().readObject(in);
    possibleDuplicate = in.readBoolean();
    creationTime = in.readLong();
    bucketId = in.readInt();
    shadowKey = in.readLong();
    versionTimeStamp = in.readLong();
  }

  protected void deserializeKey(DataInput in,
      DeserializationContext context) throws IOException, ClassNotFoundException {
    key = context.getDeserializer().readObject(in);
  }

  @Override
  public String toString() {
    return "GatewaySenderEventImpl[" + "id=" + id + ";action="
        + action + ";operation=" + getOperation() + ";region="
        + regionPath + ";key=" + key + ";value="
        + getValueAsString(true) + ";valueIsObject=" + valueIsObject
        + ";numberOfParts=" + numberOfParts + ";callbackArgument="
        + callbackArgument + ";possibleDuplicate=" + possibleDuplicate
        + ";creationTime=" + creationTime + ";shadowKey="
        + shadowKey + ";timeStamp=" + versionTimeStamp
        + ";acked=" + isAcked + ";dispatched=" + isDispatched
        + ";bucketId=" + bucketId
        + ";isConcurrencyConflict=" + isConcurrencyConflict
        + ";transactionId=" + transactionId
        + ";isLastEventInTransaction=" + isLastEventInTransaction
        + "]";
  }

  public String toSmallString() {
    return "GatewaySenderEventImpl[" + "id=" + id + ";operation="
        + getOperation() + ";region=" + regionPath + ";key="
        + key + ";shadowKey=" + shadowKey + ";bucketId="
        + bucketId + "]";
  }

  public static boolean isSerializingValue() {
    return isSerializingValue.get();
  }

  // public static boolean isDeserializingValue() {
  // return ((Boolean)isDeserializingValue.get()).booleanValue();
  // }

  // / Conflatable interface methods ///

  /**
   * Determines whether or not to conflate this message. This method will answer true IFF the
   * message's operation is AFTER_UPDATE and its region has enabled are conflation. Otherwise, this
   * method will answer false. Messages whose operation is AFTER_CREATE, AFTER_DESTROY,
   * AFTER_INVALIDATE or AFTER_REGION_DESTROY are not conflated.
   *
   * @return Whether to conflate this message
   */
  @Override
  public boolean shouldBeConflated() {
    // If the message is an update, it may be conflatable. If it is a
    // create, destroy, invalidate or destroy-region, it is not conflatable.
    // Only updates are conflated.
    return isUpdate();
  }

  @Override
  public String getRegionToConflate() {
    return regionPath;
  }

  @Override
  public Object getKeyToConflate() {
    return key;
  }

  @Override
  public Object getValueToConflate() {
    // Since all the uses of this are for logging
    // changing it to return the string form of the value
    // instead of the actual value.
    return getValueAsString(true);
  }

  @Override
  public void setLatestValue(Object value) {
    // Currently this method is never used.
    // If someone does want to use it in the future
    // then the implementation needs to be updated
    // to correctly update value, valueObj, and valueIsObject
    throw new UnsupportedOperationException();
  }

  // / End Conflatable interface methods ///

  /**
   * Returns whether this <code>GatewayEvent</code> represents an update.
   *
   * @return whether this <code>GatewayEvent</code> represents an update
   */
  protected boolean isUpdate() {
    // This event can be in one of three states:
    // - in memory primary (initialized)
    // - in memory secondary (not initialized)
    // - evicted to disk, read back in (initialized)
    // In the first case, both the operation and action are set.
    // In the second case, only the operation is set.
    // In the third case, only the action is set.
    return operation == null ? action == UPDATE_ACTION
        : operation == EnumListenerEvent.AFTER_UPDATE;
  }

  /**
   * Returns whether this <code>GatewayEvent</code> represents a create.
   *
   * @return whether this <code>GatewayEvent</code> represents a create
   */
  protected boolean isCreate() {
    // See the comment in isUpdate() for additional details
    return operation == null ? action == CREATE_ACTION
        : operation == EnumListenerEvent.AFTER_CREATE;
  }

  /**
   * Returns whether this <code>GatewayEvent</code> represents a destroy.
   *
   * @return whether this <code>GatewayEvent</code> represents a destroy
   */
  protected boolean isDestroy() {
    // See the comment in isUpdate() for additional details
    return operation == null ? action == DESTROY_ACTION
        : operation == EnumListenerEvent.AFTER_DESTROY;
  }

  /**
   * Initialize the unique identifier for this event. This id is used by the receiving
   * <code>Gateway</code> to keep track of which events have been processed. Duplicates can be
   * dropped.
   */
  private void initializeId(EntryEventImpl event) {
    // CS43_HA
    id = event.getEventId();
    // TODO:ASIF :Once stabilized remove the check below
    if (id == null) {
      throw new IllegalStateException(
          "No event id is available for this gateway event.");
    }

  }

  /**
   * Initialize this instance. Get the useful parts of the input operation and event.
   */
  public void initialize() {
    if (isInitialized()) {
      return;
    }
    isInitialized = true;
  }


  // Initializes the value object. This function need a relook because the
  // serialization of the value looks unnecessary.
  @Retained(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
  protected void initializeValue(EntryEventImpl event) {
    // Set the value to be a byte[] representation of either the value or
    // substituteValue (if set).
    if (substituteValue == null) {
      // If the value is already serialized, use it.
      valueIsObject = 0x01;
      /*
       * so ends up being stored in this.valueObj
       */
      @Retained(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
      StoredObject so;
      {
        ReferenceCountHelper.setReferenceCountOwner(this);
        so = event.getOffHeapNewValue();
        ReferenceCountHelper.setReferenceCountOwner(null);
      }

      if (so != null) {
        // Since GatewaySenderEventImpl instances can live for a long time in the gateway region
        // queue we do not want the StoredObject to be one that keeps the heap form cached.
        so = so.getStoredObjectWithoutHeapForm();
        valueObj = so;
        if (!so.isSerialized()) {
          valueIsObject = 0x00;
        }
      } else if (event.getCachedSerializedNewValue() != null) {
        // We want this to have lower precedence than StoredObject so that the gateway
        // can share a reference to the off-heap value.
        value = event.getCachedSerializedNewValue();
      } else {
        final Object newValue = event.getRawNewValue();
        assert !(newValue instanceof StoredObject); // since we already called getOffHeapNewValue()
                                                    // and it returned null
        if (newValue instanceof CachedDeserializable) {
          value = ((CachedDeserializable) newValue).getSerializedValue();
        } else if (newValue instanceof byte[]) {
          // The value is byte[]. Set _valueIsObject flag to 0x00 (not an object)
          value = (byte[]) newValue;
          valueIsObject = 0x00;
        } else {
          // The value is an object. It will be serialized later when getSerializedValue is called.
          valueObj = newValue;
          // to prevent bug 48281 we need to serialize it now
          getSerializedValue();
          valueObj = null;
        }
      }
    } else {
      // The substituteValue is set. Use it.
      if (substituteValue instanceof byte[]) {
        // The substituteValue is byte[]. Set valueIsObject flag to 0x00 (not an object)
        value = (byte[]) substituteValue;
        valueIsObject = 0x00;
      } else if (substituteValue == TOKEN_NULL) {
        // The substituteValue represents null. Set the value and substituteValue to null.
        value = null;
        substituteValue = null;
        valueIsObject = 0x01;
      } else {
        // The substituteValue is an object. Leave it as is.
        valueIsObject = 0x01;
      }
    }
  }

  /**
   * Initialize this event's action and number of parts
   *
   * @param operation The operation from which to initialize this event's action and number of parts
   */
  protected void initializeAction(EnumListenerEvent operation, EntryEventImpl event) {
    if (operation == EnumListenerEvent.AFTER_CREATE) {
      // Initialize after create action
      action = CREATE_ACTION;

      // Initialize number of parts
      // part 1 = action
      // part 2 = posDup flag
      // part 3 = regionName
      // part 4 = eventId
      // part 5 = key
      // part 6 = value (create and update only)
      // part 7 = whether callbackArgument is non-null
      // part 8 = callbackArgument (if non-null)
      // part 9 = versionTimeStamp;
      numberOfParts = (callbackArgument == null) ? 8 : 9;
    } else if (operation == EnumListenerEvent.AFTER_UPDATE) {
      // Initialize after update action
      action = UPDATE_ACTION;

      // Initialize number of parts
      numberOfParts = (callbackArgument == null) ? 8 : 9;
    } else if (operation == EnumListenerEvent.AFTER_DESTROY) {
      // Initialize after destroy action
      action = DESTROY_ACTION;

      // Initialize number of parts
      // Since there is no value, there is one less part
      numberOfParts = (callbackArgument == null) ? 7 : 8;
    } else if (operation == EnumListenerEvent.TIMESTAMP_UPDATE) {
      // Initialize after destroy action
      action = VERSION_ACTION;

      // Initialize number of parts
      // Since there is no value, there is one less part
      numberOfParts = (callbackArgument == null) ? 7 : 8;
    } else if (operation == EnumListenerEvent.AFTER_INVALIDATE) {
      // Initialize after invalidate action
      action = INVALIDATE_ACTION;

      // Initialize number of parts
      // Since there is no value, there is one less part
      numberOfParts = (callbackArgument == null) ? 7 : 8;
    } else if (operation == EnumListenerEvent.AFTER_UPDATE_WITH_GENERATE_CALLBACKS) {
      if (event.isGenerateCallbacks()) {
        action = UPDATE_ACTION;
      } else {
        action = UPDATE_ACTION_NO_GENERATE_CALLBACKS;
      }
      numberOfParts = (callbackArgument == null) ? 8 : 9;
    }
  }

  private void initializeOperationDetail(Operation operation) {
    if (operation.isLocalLoad()) {
      operationDetail = OP_DETAIL_LOCAL_LOAD;
    } else if (operation.isNetLoad()) {
      operationDetail = OP_DETAIL_NET_LOAD;
    } else if (operation.isPutAll()) {
      operationDetail = OP_DETAIL_PUTALL;
    } else if (operation.isRemoveAll()) {
      operationDetail = OP_DETAIL_REMOVEALL;
    } else {
      operationDetail = OP_DETAIL_NONE;
    }
  }

  @Override
  public EventID getEventId() {
    return id;
  }

  /**
   * Return the EventSequenceID of the Event
   *
   */
  @Override
  public EventSequenceID getEventSequenceID() {
    return new EventSequenceID(id.getMembershipID(), id.getThreadID(), id.getSequenceID());
  }

  public long getVersionTimeStamp() {
    return versionTimeStamp;
  }

  @Override
  public int getSizeInBytes() {
    // Calculate the size of this event. This is used for overflow to disk.

    // The sizes of the following variables are calculated:
    //
    // - the value (byte[])
    // - the original callback argument (Object)
    // - primitive and object instance variable references
    //
    // The sizes of the following variables are not calculated:

    // - the key because it is a reference
    // - the region and regionName because they are references
    // - the operation because it is a reference
    // - the entry event because it is nulled prior to calling this method
    // - the transactionId because it is is a reference

    // The size of instances of the following internal datatypes were estimated
    // using a NullDataOutputStream and hardcoded into this method:

    // - the id (an instance of EventId)
    // - the callbackArgument (an instance of GatewayEventCallbackArgument)

    int size = 0;

    // Add this event overhead
    size += Sizeable.PER_OBJECT_OVERHEAD;

    // Add object references
    // _id reference = 4 bytes
    // _region reference = 4 bytes
    // _regionName reference = 4 bytes
    // _key reference = 4 bytes
    // _callbackArgument reference = 4 bytes
    // _operation reference = 4 bytes
    // _entryEvent reference = 4 bytes
    // _transactionId reference = 4 bytes
    size += 32;

    // Add primitive references
    // int _action = 4 bytes
    // int _numberOfParts = 4 bytes
    // byte _valueIsObject = 1 byte
    // boolean _possibleDuplicate = 1 byte
    // int bucketId = 4 bytes
    // long shadowKey = 8 bytes
    // long creationTime = 8 bytes
    // boolean _hasTransaction = 1 byte
    // boolean _isLastEventInTransaction = 1 byte
    size += 32;

    // Add the id (an instance of EventId)
    // The hardcoded value below was estimated using a NullDataOutputStream
    size += Sizeable.PER_OBJECT_OVERHEAD + 56;

    // The value (a byte[])
    size += getSerializedValueSize();

    // The callback argument (a GatewayEventCallbackArgument wrapping an Object
    // which is the original callback argument)
    // The hardcoded value below represents the GatewayEventCallbackArgument
    // and was estimated using a NullDataOutputStream
    size += Sizeable.PER_OBJECT_OVERHEAD + 194;
    // The sizeOf call gets the size of the input callback argument.
    size += Sizeable.PER_OBJECT_OVERHEAD + sizeOf(getCallbackArgument());

    // the version timestamp
    size += 8;

    return size;
  }

  private int sizeOf(Object obj) {
    int size = 0;
    if (obj == null) {
      return size;
    }
    if (obj instanceof String) {
      size = ObjectSizer.DEFAULT.sizeof(obj);
    } else if (obj instanceof Integer) {
      size = 4; // estimate
    } else if (obj instanceof Long) {
      size = 8; // estimate
    } else {
      size = CachedDeserializableFactory.calcMemSize(obj) - Sizeable.PER_OBJECT_OVERHEAD;
    }
    return size;
  }


  // Asif: If the GatewayEvent serializes to a node where the region itself may
  // not be present or the
  // region is not created yet , and if the gateway event queue is persistent,
  // then even if
  // we try to set the region in the fromData , we may still get null. Though
  // the product is
  // not using this method anywhere still not comfortable changing the Interface
  // so
  // modifying the implementation a bit.

  @Override
  public Region<?, ?> getRegion() {
    // The region will be null mostly for the other node where the gateway event
    // is serialized
    return region != null ? region
        : CacheFactory.getAnyInstance().getRegion(regionPath);
  }

  public int getBucketId() {
    return bucketId;
  }

  public boolean isConcurrencyConflict() {
    return isConcurrencyConflict;
  }

  /**
   * @param tailKey the tailKey to set
   */
  public void setShadowKey(Long tailKey) {
    shadowKey = tailKey;
  }

  /**
   * @return the tailKey
   */
  public Long getShadowKey() {
    return shadowKey;
  }

  public boolean isLastEventInTransaction() {
    return isLastEventInTransaction;
  }

  public TransactionId getTransactionId() {
    return transactionId;
  }

  public boolean equals(Object obj) {
    if (this == obj) {
      return true;
    }

    if (!(obj instanceof GatewaySenderEventImpl)) {
      return false;
    }

    GatewaySenderEventImpl that = (GatewaySenderEventImpl) obj;

    return shadowKey.equals(that.shadowKey)
        && id.equals(that.id)
        && bucketId == that.bucketId
        && action == that.action
        && regionPath.equals(that.regionPath)
        && key.equals(that.key)
        && Arrays.equals(value, that.value);
  }

  public int hashCode() {
    int hashCode = 17;
    hashCode = 37 * hashCode + ObjectUtils.hashCode(shadowKey);
    hashCode = 37 * hashCode + ObjectUtils.hashCode(id);
    hashCode = 37 * hashCode + bucketId;
    hashCode = 37 * hashCode + action;
    hashCode = 37 * hashCode + ObjectUtils.hashCode(regionPath);
    hashCode = 37 * hashCode + ObjectUtils.hashCode(key);
    hashCode = 37 * hashCode + (value == null ? 0 : Arrays.hashCode(value));
    return hashCode;
  }

  @Override
  public KnownVersion[] getSerializationVersions() {
    return new KnownVersion[] {KnownVersion.GEODE_1_9_0, KnownVersion.GEODE_1_14_0,
        KnownVersion.GEODE_1_15_0};
  }

  public int getSerializedValueSize() {
    int localSerializedValueSize = serializedValueSize;
    if (localSerializedValueSize != DEFAULT_SERIALIZED_VALUE_SIZE) {
      return localSerializedValueSize;
    }
    @Unretained(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
    Object vo = valueObj;
    if (vo instanceof StoredObject) {
      localSerializedValueSize = ((StoredObject) vo).getSizeInBytes();
    } else {
      if (substituteValue != null) {
        localSerializedValueSize = sizeOf(substituteValue);
      } else {
        localSerializedValueSize = CachedDeserializableFactory.calcMemSize(getSerializedValue());
      }
    }
    serializedValueSize = localSerializedValueSize;
    return localSerializedValueSize;
  }

  @Override
  @Released(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
  public synchronized void release() {
    @Released(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE)
    Object vo = valueObj;
    if (OffHeapHelper.releaseAndTrackOwner(vo, this)) {
      valueObj = null;
      valueObjReleased = true;
    }
  }

  public static void release(
      @Released(OffHeapIdentifier.GATEWAY_SENDER_EVENT_IMPL_VALUE) Object o) {
    if (o instanceof GatewaySenderEventImpl) {
      ((GatewaySenderEventImpl) o).release();
    }
  }

  /**
   * Make a heap copy of this off-heap event and return it. A copy only needs to be made if the
   * event's value is stored off-heap. If it is already on the java heap then just return "this". If
   * it was stored off-heap and is no longer available (because it was released) then return null.
   */
  public GatewaySenderEventImpl makeHeapCopyIfOffHeap() {
    if (value != null || substituteValue != null) {
      // we have the value stored on the heap so return this
      return this;
    } else {
      Object v = valueObj;
      if (v == null) {
        if (valueObjReleased) {
          // this means that the original off heap value was freed
          return null;
        } else {
          return this;
        }
      }
      if (v instanceof StoredObject && ((StoredObject) v).hasRefCount()) {
        try {
          return makeCopy();
        } catch (IllegalStateException ex) {
          // this means that the original off heap value was freed
          return null;
        }
      } else {
        // the valueObj does not use refCounts so just return this.
        return this;
      }
    }
  }

  protected GatewaySenderEventImpl makeCopy() {
    return new GatewaySenderEventImpl(this);
  }

  public void setAcked(boolean acked) {
    isAcked = acked;
  }

  public enum TransactionMetadataDisposition {
    /**
     * Transaction metadata should be excluded from the event.
     */
    EXCLUDE,
    /**
     * Transaction metadata should be included in the event.
     */
    INCLUDE,
    /**
     * Transaction metadata should be included in the event and this is the last event in the
     * transaction.
     */
    INCLUDE_LAST_EVENT,
  }
}