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

import static org.apache.geode.cache.Region.SEPARATOR_CHAR;
import static org.apache.geode.internal.lang.SystemProperty.getProductIntegerProperty;
import static org.apache.geode.internal.lang.SystemPropertyHelper.HA_REGION_QUEUE_EXPIRY_TIME_PROPERTY;
import static org.apache.geode.internal.lang.SystemPropertyHelper.THREAD_ID_EXPIRY_TIME_PROPERTY;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.Queue;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.DataSerializable;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.CustomExpiry;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.ExpirationAction;
import org.apache.geode.cache.ExpirationAttributes;
import org.apache.geode.cache.MirrorType;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.query.internal.CqQueryVsdStats;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.InternalCqQuery;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.cache.CacheServerImpl;
import org.apache.geode.internal.cache.CachedDeserializable;
import org.apache.geode.internal.cache.Conflatable;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.HARegion;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.RegionQueue;
import org.apache.geode.internal.cache.VMCachedDeserializable;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.ClientMarkerMessageImpl;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessage;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessageImpl;
import org.apache.geode.internal.cache.tier.sockets.ClientUpdateMessageImpl.CqNameToOp;
import org.apache.geode.internal.cache.tier.sockets.HAEventWrapper;
import org.apache.geode.internal.cache.tier.sockets.Handshake;
import org.apache.geode.internal.logging.log4j.LogMarker;
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.serialization.Versioning;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.util.BlobHelper;
import org.apache.geode.internal.util.concurrent.StoppableCondition;
import org.apache.geode.internal.util.concurrent.StoppableReentrantLock;
import org.apache.geode.internal.util.concurrent.StoppableReentrantReadWriteLock;
import org.apache.geode.internal.util.concurrent.StoppableReentrantReadWriteLock.StoppableWriteLock;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * An implementation of Queue using Gemfire Region as the underlying datastructure. The key will be
 * a counter(long) and value will be the offered Object in the queue. For example, an entry in this
 * region (key = 5; value = obj1) would mean that the Object obj1 is at the 5th position in the
 * queue.
 *
 * This class has a field idsAvailable which is guraded by a ReentrantReadWriteLock. The peek
 * threads which do not modify the idsAvailable LinkedhashSet take read lock , thereby increasing
 * the concurrency of peek operations. The threads like take, remove, QRM ,put& expiry take a write
 * lock while operating on the set. <BR>
 * <B>This class is performant for multiple dispatchers that are trying to do non blocking peek </B>
 * <br>
 * For Blocking operations the object should be of type BlockingHARegionQueue. This class has just a
 * ReentrantLock . Condition object of this lock is used to issue signal to the waiting peek & take
 * threads for put operation. As a result , concurrency of peek operations is not possible. This
 * class is optimized for a single peek thread .
 *
 * 30 May 2008: 5.7 onwards the underlying GemFire Region will continue to have key as counter(long)
 * but the value will be a wrapper object(HAEventWrapper) which will be a key in a separate data
 * structure called haContainer (an implementation of Map). The value against this wrapper will be
 * the offered object in the queue. The purpose of this modification is to allow multiple
 * ha-region-queues share their offered values without storing separate copies in memory, upon GII.
 *
 * (See BlockingHARegionQueue)
 *
 * @since GemFire 4.3
 */
public class HARegionQueue implements RegionQueue {
  private static final Logger logger = LogService.getLogger();

  /** The {@code Region} backing this queue */
  protected HARegion region;

  /**
   * The key into the {@code Region} used when putting entries onto the queue. The counter uses
   * incrementAndGet so counter will always be started from 1
   */
  protected final AtomicLong tailKey = new AtomicLong(0);

  /**
   * Map whose key is the ThreadIdentifier object value is the LastDispatchedAndCurrentEvents
   * object. Every add operation will be identified by the ThreadIdentifier object & the position
   * recorded in the LastDispatchedAndCurrentEvents object.
   */
  protected final ConcurrentMap eventsMap = new ConcurrentHashMap<>();

  /**
   * The {@code Map} mapping the regionName->key to the queue key. This index allows fast updating
   * of entries in the queue for conflation.
   */
  protected volatile Map indexes = Collections.unmodifiableMap(new HashMap<>());

  private final StoppableReentrantReadWriteLock rwLock;

  private final StoppableReentrantReadWriteLock.StoppableReadLock readLock;

  private final StoppableWriteLock writeLock;

  /** The name of the {@code Region} backing this queue */
  private final String regionName;

  /** The ClientProxyMembershipID associated with the ha queue */
  private final ClientProxyMembershipID clientProxyID;

  /**
   * The statistics for this queue
   */
  public HARegionQueueStats stats;

  /**
   * Accesses to this set must be protected via the rwLock.
   */
  protected LinkedHashSet idsAvailable;

  /**
   * Map of HA queue region-name and value as a MapWrapper object (whose underlying map contains
   * ThreadIdentifier as key & value as the last dispatched sequence ID)
   */
  @MakeNotStatic
  static ConcurrentMap dispatchedMessagesMap;

  /**
   * A MapWrapper object whose underlying map contains ThreadIdentifier as key & value as the last
   * dispatched sequence ID
   */
  protected MapWrapper threadIdToSeqId;

  /**
   * A sequence violation can occur , if an HARegionQueue receives events thru GII & the same event
   * also arrives via Gemfire Put in that local VM. If the HARegionQueue does not receive any data
   * via GII , then there should not be any violation. If there is data arriving thru GII, such
   * violations can be expected , but should be analyzed thoroughly.
   */
  protected final AtomicBoolean puttingGIIDataInQueue = new AtomicBoolean();

  /**
   * flag indicating whether interest has been registered for this queue. This is used to prevent
   * race conditions between secondaries and primary for interest registration.
   */
  private boolean hasRegisteredInterest;

  /**
   * a thread local to store the counters corresponding to the events peeked by a particular thread.
   * When {@code remove()} will be called, these events stored in thread-local will be destroyed.
   */
  protected static final ThreadLocal peekedEventsContext = new ThreadLocal<>();

  /**
   * Thread which creates the {@code QueueRemovalMessage} and sends it to other nodes in the system
   */
  @MakeNotStatic
  private static QueueRemovalThread qrmThread;

  /** protects from modification during GII chunking */
  private final StoppableReentrantReadWriteLock giiLock;

  /** the number of concurrent GII requests being served */
  private final AtomicInteger giiCount = new AtomicInteger();

  /** queue to hold events during GII transfer so we do not modify the queue during chunking */
  private final Queue giiQueue = new ConcurrentLinkedQueue<>();

  /**
   * Constant used to indicate the instance of BlockingHARegionQueue. The static function used for
   * creating the queue instance should be passed this as parameter for creating
   * BlockingHARegionQueue
   */
  public static final int BLOCKING_HA_QUEUE = 1;

  /**
   * Constant used to indicate the instance of HARegionQueue. The static function used for creating
   * the queue instance should be passed this as parameter for creating HARegionQueue
   */
  public static final int NON_BLOCKING_HA_QUEUE = 2;

  public static final String HA_EVICTION_POLICY_NONE = "none";

  public static final String HA_EVICTION_POLICY_MEMORY = "mem";

  public static final String HA_EVICTION_POLICY_ENTRY = "entry";

  public static final long INIT_OF_SEQUENCEID = -1L;

  /**
   * The default frequency (in seconds) at which a message will be sent by the primary to all the
   * secondary nodes to remove the events which have already been dispatched from the queue.
   */
  public static final int DEFAULT_MESSAGE_SYNC_INTERVAL = 1;

  /**
   * The frequency (in seconds) at which a message will be sent by the primary to all the secondary
   * nodes to remove the events which have already been dispatched from the queue.
   */
  @MakeNotStatic
  protected static volatile int messageSyncInterval = DEFAULT_MESSAGE_SYNC_INTERVAL;

  /**
   * The underlying map (may hold reference to a Region or a ConcurrentHashMap) for this
   * HARegionQueue instance (and also shared by all the HARegionQueue instances associated with the
   * same CacheClientNotifier).
   */
  protected Map haContainer;

  /**
   * Boolean to indicate whether this HARegionQueue is having active dispatcher or not( primary node
   * or the secondary node). This will be used to prevent the events from expiry if the node is
   * primary. This was introduced for fixing the bug#36853
   */
  private volatile boolean isPrimary = false;

  /** Boolean to indicate whether destruction of queue is in progress. */
  protected volatile boolean destroyInProgress = false;

  private final CancelCriterion stopper;

  /** @since GemFire 5.7 */
  protected byte clientConflation = Handshake.CONFLATION_DEFAULT;

  /**
   * Boolean to indicate whether client is a slow receiver
   *
   * @since GemFire 6.0
   */
  public boolean isClientSlowReceiver = false;

  /**
   * initialization flag - when true the queue has fully initialized
   */
  final AtomicBoolean initialized = new AtomicBoolean();

  /**
   * Test hooks for periodic ack
   *
   * @since GemFire 6.0
   */
  @MutableForTesting
  static boolean testMarkerMessageReceived = false;
  @MutableForTesting
  static boolean isUsedByTest = false;

  /**
   * Used by durable queues to maintain acked events by client
   */
  protected Map ackedEvents;

  /**
   * Indicates how many times this ha queue has hit the max queue size limit.
   */
  protected long maxQueueSizeHitCount = 0;

  final AtomicBoolean hasSynchronizedWithPrimary = new AtomicBoolean();
  final AtomicBoolean scheduleSynchronizationWithPrimaryInProgress = new AtomicBoolean();
  final AtomicBoolean doneGIIQueueing = new AtomicBoolean();
  volatile long doneGIIQueueingTime;
  volatile long positionBeforeGII = 0;
  volatile long positionAfterGII = 0;

  /**
   * Processes the given string and returns a string which is allowed for region names
   *
   * @return legal region name
   */
  public static String createRegionName(String regionName) {
    return regionName.replace(SEPARATOR_CHAR, '#');
  }

  /**
   * @param isPrimary whether this is the primary queue for a client
   */
  protected HARegionQueue(String regionName, InternalCache cache, Map haContainer,
      ClientProxyMembershipID clientProxyId, final byte clientConflation,
      boolean isPrimary,
      StatisticsClock statisticsClock)
      throws IOException, ClassNotFoundException, CacheException, InterruptedException {
    String processedRegionName = createRegionName(regionName);

    // Initialize the statistics
    StatisticsFactory factory = cache.getInternalDistributedSystem().getStatisticsManager();

    AttributesFactory af = new AttributesFactory<>();
    af.setMirrorType(MirrorType.KEYS_VALUES);
    af.addCacheListener(createCacheListenerForHARegion());
    af.setStatisticsEnabled(true);
    RegionAttributes ra = af.create();

    region = HARegion.getInstance(processedRegionName, cache, this, ra, statisticsClock);

    if (this.isPrimary) {// fix for 41878
      // since it's primary queue, we will disable the EntryExpiryTask
      // this should be done after region creation
      disableEntryExpiryTasks();
    }

    this.regionName = processedRegionName;
    threadIdToSeqId = new MapWrapper();
    idsAvailable = new LinkedHashSet<>();
    setClientConflation(clientConflation);
    this.isPrimary = isPrimary;
    // Initialize the statistics
    stats = new HARegionQueueStats(factory, processedRegionName);
    this.haContainer = haContainer;
    giiLock = new StoppableReentrantReadWriteLock(cache.getCancelCriterion());
    clientProxyID = clientProxyId;

    stopper = region.getCancelCriterion();
    rwLock = new StoppableReentrantReadWriteLock(region.getCancelCriterion());
    readLock = rwLock.readLock();
    writeLock = rwLock.writeLock();

    putGIIDataInRegion();

    if (getClass() == HARegionQueue.class) {
      synchronized (initialized) {
        initialized.set(true);
        initialized.notifyAll();
      }
    }
  }

  @VisibleForTesting
  HARegionQueue(String regionName, HARegion haRegion, InternalCache cache, Map haContainer,
      ClientProxyMembershipID clientProxyId, final byte clientConflation,
      boolean isPrimary,
      HARegionQueueStats stats, StoppableReentrantReadWriteLock giiLock,
      StoppableReentrantReadWriteLock rwLock, CancelCriterion cancelCriterion,
      boolean puttingGIIDataInQueue, StatisticsClock statisticsClock)
      throws CacheException, InterruptedException {
    this.regionName = regionName;
    region = haRegion;
    threadIdToSeqId = new MapWrapper();
    idsAvailable = new LinkedHashSet<>();
    setClientConflation(clientConflation);
    this.isPrimary = isPrimary;
    // Initialize the statistics
    this.stats = stats;
    this.haContainer = haContainer;
    this.giiLock = giiLock;
    clientProxyID = clientProxyId;

    stopper = cancelCriterion;
    this.rwLock = rwLock;
    readLock = this.rwLock.readLock();
    writeLock = this.rwLock.writeLock();

    // false specifically set in tests only
    if (puttingGIIDataInQueue) {
      putGIIDataInRegion();
    }
    if (getClass() == HARegionQueue.class) {
      synchronized (initialized) {
        initialized.set(true);
        initialized.notifyAll();
      }
    }
  }

  public boolean isQueueInitializedWithWait(long waitTime) {
    try {
      synchronized (initialized) {
        if (!isQueueInitialized()) {
          waitForInitialized(waitTime);
        }
      }
    } catch (InterruptedException interruptedException) {
      Thread.currentThread().interrupt();
      return false;
    }
    return isQueueInitialized();
  }

  void waitForInitialized(long waitTime) throws InterruptedException {
    initialized.wait(waitTime);
  }

  /**
   * install DACE information from an initial image provider
   */
  @SuppressWarnings("synthetic-access")
  public void recordEventState(InternalDistributedMember sender,
      Map<? extends DataSerializable, ? extends DataSerializable> eventState) {
    StringBuilder sb = null;
    final boolean isDebugEnabled_BS = logger.isTraceEnabled(LogMarker.BRIDGE_SERVER_VERBOSE);
    if (isDebugEnabled_BS) {
      sb = new StringBuilder(500);
      sb.append("Recording initial event state for ").append(regionName).append(" from ")
          .append(sender);
    }
    for (Object value : eventState.entrySet()) {
      Map.Entry<ThreadIdentifier, DispatchedAndCurrentEvents> entry =
          (Map.Entry<ThreadIdentifier, DispatchedAndCurrentEvents>) value;
      DispatchedAndCurrentEvents giiDace = entry.getValue();
      if (giiDace != null) { // fix for bug #41789: npe during state transfer
        giiDace.owningQueue = this;
        giiDace.isGIIDace = true;
        if (giiDace.QRM_LOCK == null) {
          // this has happened a number of times with JDK 1.6, even when QRM_LOCK
          // was a "final" field
          giiDace.QRM_LOCK = new Object();
        }
        if (isDebugEnabled_BS) {
          sb.append("\n  ").append(entry.getKey().expensiveToString())
              .append("; sequenceID=").append(giiDace.lastSequenceIDPut);
        }
        // use putIfAbsent to avoid overwriting newer dispatch information
        Object o = eventsMap.putIfAbsent(entry.getKey(), giiDace);
        if (o != null && isDebugEnabled_BS) {
          sb.append(" -- could not store.  found ").append(o);
        }
      }
    }
    if (isDebugEnabled_BS) {
      logger.trace(LogMarker.BRIDGE_SERVER_VERBOSE, sb.toString());
    }
  }

  /**
   * Repopulates the HARegion after the GII is over so as to reset the counters and populate the
   * DACE objects for the thread identifiers . This method should be invoked as the last method in
   * the constructor . Thus while creating BlockingQueue this method should be invoked lastly in the
   * derived class constructor , after the HARegionQueue contructor is complete. Otherwise, the
   * ReentrantLock will be null.
   */
  void putGIIDataInRegion() throws CacheException, InterruptedException {
    Set<Region.Entry> entrySet = region.entrySet(false);
    // check if the region is not empty. if there is
    // data, then the relevant data structures have to
    // be populated
    if (!entrySet.isEmpty()) {
      puttingGIIDataInQueue.set(true);
      final boolean isDebugEnabled = logger.isDebugEnabled();
      try {
        Region.Entry entry;
        Map orderedMap = new TreeMap<>();
        Iterator<Region.Entry> iterator = entrySet.iterator();
        Object key;
        while (iterator.hasNext()) {
          entry = iterator.next();
          key = entry.getKey();
          if (isDebugEnabled) {
            logger.debug("{} processing queue key {} and value {}", regionName, key,
                entry.getValue());
          }
          if (key instanceof Long) {
            if (!(entry.getValue() instanceof ClientMarkerMessageImpl)) {
              orderedMap.put(key, entry.getValue());
            }
          }
          region.localDestroy(key);
        }
        long max = 0;
        long counterInRegion;
        entrySet = orderedMap.entrySet();
        if (!entrySet.isEmpty()) {
          Map.Entry mapEntry;
          iterator = entrySet.iterator();
          while (iterator.hasNext()) {
            mapEntry = iterator.next();
            Conflatable val = (Conflatable) mapEntry.getValue();
            if (val != null && val.getEventId() != null) {
              counterInRegion = ((Long) mapEntry.getKey()).intValue();
              // TODO: remove this assertion
              Assert.assertTrue(counterInRegion > max);
              max = counterInRegion;
              put(val);
            } else if (isDebugEnabled) {
              logger.debug(
                  "{} bug 44959 encountered: HARegion.putGIIDataInRegion found null eventId in {}",
                  regionName, val);
            }
          }
        }
        tailKey.set(max);
      } finally {
        puttingGIIDataInQueue.set(false);
        positionAfterGII = tailKey.get();
        if (positionAfterGII > 0) {
          positionBeforeGII = 1;
        }
        doneGIIQueueing.set(true);
        doneGIIQueueingTime = System.currentTimeMillis();
        if (isDebugEnabled) {
          logger.debug("{} done putting GII data into queue", this);
        }
      }
    }
    // TODO:Asif: Avoid invocation of this method
    startHAServices(region.getCache());
  }

  /**
   * Puts the GII'd entry into the ha region, if it was GII'd along with its ClientUpdateMessageImpl
   * instance.
   *
   * @since GemFire 5.7
   */
  protected void putInQueue(Object val) throws InterruptedException {
    if (val instanceof HAEventWrapper && ((HAEventWrapper) val).getClientUpdateMessage() == null) {
      if (logger.isDebugEnabled()) {
        logger.debug(
            "{} HARegionQueue.putGIIDataInRegion(): key={} was removed at sender side, so not putting it into the ha queue.",
            regionName, ((Conflatable) val).getKeyToConflate());
      }
    } else {
      put(val);
    }
  }

  /**
   * Check whether to conflate an event
   *
   * @since GemFire 5.7
   */
  protected boolean shouldBeConflated(Conflatable event) {
    boolean retVal = event.shouldBeConflated();
    // don't apply the client conflation override on durable markers
    if (event instanceof ClientMarkerMessageImpl) {
      return retVal;
    }
    switch (clientConflation) {
      case Handshake.CONFLATION_OFF:
        return false; // always disable
      case Handshake.CONFLATION_ON:
        if (event instanceof HAEventWrapper) {
          ClientUpdateMessage cum = (ClientUpdateMessage) haContainer.get(event);
          if (cum != null) {
            retVal = cum.isUpdate();
          }
          break;
        }
        if (event instanceof ClientUpdateMessage) {
          // Does this ever happen now?
          retVal = ((ClientUpdateMessage) event).isUpdate();
          break;
        }
        // Oddness
        break;
      case Handshake.CONFLATION_DEFAULT:
        return retVal;
      default:
        throw new InternalGemFireError("Invalid clientConflation");
    }
    return retVal;
  }

  /**
   * Adds an object at the queue's tail. The implementation supports concurrent put operations in a
   * performant manner. This is done in following steps: <br>
   * 1)Check if the Event being added has a sequence ID less than the Last Dispatched SequenceId. If
   * yes do not add the Event to the Queue <br>
   * 2)If no then Do a "put-if-absent" on the eventsMap ,inserting the
   * LastDispatchedAndCurrentEvents object or adding the position to the existing object for the
   * ThreadIdentifier
   *
   * It is possible that DispatchedAnCurrentEvents object just retrieved by the put thread has
   * expired thus a one level recursion can occur to do a valid put
   *
   * The operation is thread safe & is guarded by taking a lock on LastDispatchedAndCurrentEvents
   * object & SIZE Lock
   *
   * @param object object to put onto the queue
   */
  @Override
  public boolean put(Object object) throws CacheException, InterruptedException {
    giiLock.readLock().lock(); // fix for bug #41681 - durable client misses event

    try {
      if (giiCount.get() > 0) {
        if (logger.isDebugEnabled()) {
          logger.debug("{}: adding message to GII queue of size {}: {}", regionName,
              giiQueue.size(), object);
        }

        if (object instanceof HAEventWrapper) {
          HAEventWrapper wrapper = (HAEventWrapper) object;
          wrapper.incrementPutInProgressCounter("GII queue");
        }

        giiQueue.add(object);
      } else {
        if (logger.isTraceEnabled()) {
          logger.trace("{}: adding message to HA queue: {}", regionName, object);
        }
        basicPut(object);
      }
    } finally {
      giiLock.readLock().unlock();
    }

    // basicPut() invokes dace.putObject() to put onto HARegionQueue
    // However, dace.putObject could return true even though
    // the event is not put onto the HARegionQueue due to eliding events etc.
    // So it is not reliable to be used whether offheap ref ownership is passed over to
    // the queue (if and when HARegionQueue uses offheap). The probable
    // solution could be that to let dace.putObject() to increase offheap REF count
    // when it puts the event onto the region queue. Also always release (dec)
    // the offheap REF count from the caller.
    return true;
  }


  private void basicPut(Object object) throws CacheException, InterruptedException {
    // optmistically decrease the put count
    Conflatable event = (Conflatable) object;
    try {
      waitForPermission();
    } catch (InterruptedException ie) {
      Thread.currentThread().interrupt();
      region.getCache().getCancelCriterion().checkCancelInProgress(ie);
    }

    // Get the EventID object & from it obtain the ThreadIdentifier
    EventID eventId = event.getEventId();
    ThreadIdentifier ti = getThreadIdentifier(eventId);
    long sequenceID = eventId.getSequenceID();
    // Check from Events Map if the put operation should proceed or not
    DispatchedAndCurrentEvents dace = (DispatchedAndCurrentEvents) eventsMap.get(ti);
    if (dace != null && dace.isGIIDace && puttingGIIDataInQueue.get()) {
      // we only need to retain DACE for which there are no entries in the queue.
      // for other thread identifiers we build up a new DACE
      dace = null;
    }
    if (dace != null) {
      // check the last dispatched sequence Id
      if (puttingGIIDataInQueue.get() || (sequenceID > dace.lastDispatchedSequenceId)) {
        // Asif:Insert the Event into the Region with proper locking.
        // It is possible that by the time put operation proceeds , the
        // Last dispatched id has changed so it is possible that the object at
        // this point also does not get added
        if (!dace.putObject(event, sequenceID)) {
          // dace encountered a DESTROYED token - stop adding GII data
          if (!puttingGIIDataInQueue.get()) {
            put(object);
          }
        } else {
          if (logger.isTraceEnabled(LogMarker.BRIDGE_SERVER_VERBOSE)) {
            logger.trace(LogMarker.BRIDGE_SERVER_VERBOSE, "{}: Adding message to queue: {}", this,
                object);
          }
        }

      } else {
        if (logger.isDebugEnabled()) {
          logger.debug(
              "{}: This queue has already seen this event.  The highest sequence number in the queue for {} is {}, but this event's sequence number is {}",
              regionName, ti, dace.lastDispatchedSequenceId, sequenceID);
        }
        incrementTakeSidePutPermits();
      }
    } else {
      dace = new DispatchedAndCurrentEvents(this);
      DispatchedAndCurrentEvents oldDace =
          (DispatchedAndCurrentEvents) eventsMap.putIfAbsent(ti, dace);
      if (oldDace != null) {
        dace = oldDace;
      } else {
        // Add the recently added ThreadIdentifier to the RegionQueue for expiry
        region.put(ti, dace.lastDispatchedSequenceId);
        // update the stats
        stats.incThreadIdentifiers();
      }
      if (!dace.putObject(event, sequenceID)) {
        put(object);
      } else {
        if (logger.isTraceEnabled(LogMarker.BRIDGE_SERVER_VERBOSE)) {
          logger.trace(LogMarker.BRIDGE_SERVER_VERBOSE, "{}: Adding message to queue: {}", this,
              object);
        }
      }
    }
  }

  /**
   * while serving a GII request we queue put()s in a list so that the GII image is not modified.
   * This keeps the chunking iteration from sometimes seeing a later event but missing an older
   * event. Bug #41681
   */
  public void startGiiQueueing() {
    giiLock.writeLock().lock();
    giiCount.incrementAndGet();
    if (logger.isDebugEnabled()) {
      logger.debug("{}: startGiiQueueing count is now {}", regionName, giiCount.get());
    }
    giiLock.writeLock().unlock();
  }

  /**
   * at the end of a GII image request we decrement the in-process count and, if it falls to zero we
   * empty the list of messages that put() has been building. This is done with the lock held to
   * prevent newer events from being queued before the list is drained.
   */
  public void endGiiQueueing() {
    giiLock.writeLock().lock();
    final boolean isDebugEnabled = logger.isDebugEnabled();
    try {
      giiCount.decrementAndGet();
      if (isDebugEnabled) {
        logger.debug("{}: endGiiQueueing count is now {}", regionName, giiCount);
      }
      if (giiCount.get() < 0) {
        if (isDebugEnabled) {
          logger.debug("{} found giiCount to be {}", regionName, giiCount);
        }
        giiCount.set(0);
      }
      if (giiCount.get() == 0) {
        if (isDebugEnabled) {
          logger.debug("{} all GII requests completed - draining {} messages", regionName,
              giiQueue.size());
        }
        boolean interrupted = false;
        int actualCount = 0;
        while (true) {
          Object value;
          try {
            value = giiQueue.remove();
          } catch (NoSuchElementException ignore) {
            break;
          }
          actualCount++;
          try {
            if (isDebugEnabled) {
              logger.debug("{} draining #{}: {}", regionName, (actualCount + 1), value);
            }
            if (value instanceof HAEventWrapper) {
              if (((HAEventWrapper) value).getClientUpdateMessage() == null) {
                if (isDebugEnabled) {
                  logger.debug(
                      "{} ATTENTION: found gii queued event with null event message.  Please see bug #44852: {}",
                      regionName, value);
                }
                continue;
              }
            }

            basicPut(value);

            // The HAEventWrapper putInProgressCounter must be decremented because it was
            // incremented when it was queued in giiQueue.
            if (value instanceof HAEventWrapper) {
              ((HAEventWrapper) value).decrementPutInProgressCounter();
            }
          } catch (NoSuchElementException ignore) {
            break;
