/*
 * 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.activemq.artemis.core.paging.cursor.impl;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.ToIntFunction;

import io.netty.util.collection.IntObjectHashMap;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.core.filter.Filter;
import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.paging.PageTransactionInfo;
import org.apache.activemq.artemis.core.paging.PagedMessage;
import org.apache.activemq.artemis.core.paging.PagingStore;
import org.apache.activemq.artemis.core.paging.cursor.ConsumedPage;
import org.apache.activemq.artemis.core.paging.cursor.PageCursorProvider;
import org.apache.activemq.artemis.core.paging.cursor.PageIterator;
import org.apache.activemq.artemis.core.paging.cursor.PagePosition;
import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
import org.apache.activemq.artemis.core.paging.cursor.PageSubscriptionCounter;
import org.apache.activemq.artemis.core.paging.cursor.PagedReference;
import org.apache.activemq.artemis.core.paging.cursor.PagedReferenceImpl;
import org.apache.activemq.artemis.core.paging.impl.Page;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.core.transaction.TransactionOperationAbstract;
import org.apache.activemq.artemis.core.transaction.TransactionPropertyIndexes;
import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
import org.apache.activemq.artemis.utils.collections.LinkedListIterator;
import org.jboss.logging.Logger;

import static org.apache.activemq.artemis.core.server.impl.QueueImpl.DELIVERY_TIMEOUT;

public final class PageSubscriptionImpl implements PageSubscription {

   private static final Logger logger = Logger.getLogger(PageSubscriptionImpl.class);

   private static final Object DUMMY = new Object();

   private static final PagedReference RETRY_MARK = new PagedReferenceImpl(null, null);

   private boolean empty = true;

   /** for tests */
   public AtomicInteger getScheduledCleanupCount() {
      return scheduledCleanupCount;
   }

   // Number of scheduled cleanups, to avoid too many schedules
   private final AtomicInteger scheduledCleanupCount = new AtomicInteger(0);

   private volatile boolean autoCleanup = true;

   private final StorageManager store;

   private final long cursorId;

   private Queue queue;

   private final boolean persistent;

   private final Filter filter;

   private final PagingStore pageStore;

   private final PageCursorProvider cursorProvider;

   private volatile PagePosition lastAckedPosition;

   private List<PagePosition> recoveredACK;

   private final SortedMap<Long, PageCursorInfo> consumedPages = new TreeMap<>();

   private final PageSubscriptionCounter counter;

   private final AtomicLong deliveredCount = new AtomicLong(0);

   private final AtomicLong deliveredSize = new AtomicLong(0);

   /** this variable governs if we need to schedule another runner to look after the scanList. */
   private boolean pageScanNeeded = true;
   private final LinkedList<PageScan> scanList = new LinkedList();

   private static class PageScan {
      final BooleanSupplier retryBeforeScan;
      final ToIntFunction<PagedReference> scanFunction;
      final Runnable found;
      final Runnable notFound;

      public ToIntFunction<PagedReference> getScanFunction() {
         return scanFunction;
      }

      public Runnable getFound() {
         return found;
      }

      public Runnable getNotfound() {
         return notFound;
      }

      PageScan(BooleanSupplier retryBeforeScan, ToIntFunction<PagedReference> scanFunction, Runnable found, Runnable notFound) {
         this.retryBeforeScan = retryBeforeScan;
         this.scanFunction = scanFunction;
         this.found = found;
         this.notFound = notFound;
      }
   }

   @Override
   public void scanAck(BooleanSupplier retryBeforeScan, ToIntFunction<PagedReference> scanFunction, Runnable found, Runnable notFound) {
      PageScan scan = new PageScan(retryBeforeScan, scanFunction, found, notFound);
      boolean pageScanNeededLocal;
      synchronized (scanList) {
         scanList.add(scan);
         pageScanNeededLocal = this.pageScanNeeded;
         this.pageScanNeeded = false;
      }

      if (pageScanNeededLocal) {
         pageStore.execute(this::performScanAck);
      }
   }

   private void performScanAck() {
      try {
         PageScan[] localScanList;
         synchronized (scanList) {
            this.pageScanNeeded = true;
            if (scanList.size() == 0) {
               return;
            }
            localScanList = scanList.stream().toArray(i -> new PageScan[i]);
            scanList.clear();
         }

         int retriedFound = 0;
         for (int i = 0; i < localScanList.length; i++) {
            PageScan scanElemen = localScanList[i];
            if (scanElemen.retryBeforeScan != null && scanElemen.retryBeforeScan.getAsBoolean()) {
               localScanList[i] = null;
               retriedFound++;
            }
         }

         if (retriedFound == localScanList.length) {
            return;
         }

         if (!isPaging()) {
            // this would mean that between the submit and now the system left paging mode
            // at this point we will just return everything as notFound
            for (int i = 0; i < localScanList.length; i++) {
               PageScan scanElemen = localScanList[i];
               if (scanElemen != null && scanElemen.notFound != null) {
                  scanElemen.notFound.run();
               }
            }

            return;
         }

         LinkedList<Runnable> afterCommitList = new LinkedList<>();
         TransactionImpl tx = new TransactionImpl(store);
         tx.addOperation(new TransactionOperationAbstract() {
            @Override
            public void afterCommit(Transaction tx) {
               for (Runnable r : afterCommitList) {
                  try {
                     r.run();
                  } catch (Throwable e) {
                     logger.warn(e.getMessage(), e);
                  }
               }
            }
         });
         PageIterator iterator = this.iterator(true);
         try {
            while (iterator.hasNext()) {
               PagedReference reference = iterator.next();
               boolean keepMoving = false;
               for (int i = 0; i < localScanList.length; i++) {
                  PageScan scanElemen = localScanList[i];
                  if (scanElemen == null) {
                     continue;
                  }

                  int result = scanElemen.scanFunction.applyAsInt(reference);

                  if (result >= 0) {
                     if (result == 0) {
                        try {
                           PageSubscriptionImpl.this.ackTx(tx, reference);
                           if (scanElemen.found != null) {
                              afterCommitList.add(scanElemen.found);
                           }
                        } catch (Throwable e) {
                           logger.warn(e.getMessage(), e);
                        }
                     } else {
                        if (scanElemen.notFound != null) {
                           scanElemen.notFound.run();
                        }
                     }
                     localScanList[i] = null;
                  } else {
                     keepMoving = true;
                  }
               }

               if (!keepMoving) {
                  break;
               }
            }
         } finally {
            iterator.close();
         }

         for (int i = 0; i < localScanList.length; i++) {
            if (localScanList[i] != null && localScanList[i].notFound != null) {
               localScanList[i].notFound.run();
            }
            localScanList[i] = null;
         }

         if (afterCommitList.size() > 0) {
            try {
               tx.commit();
            } catch (Exception e) {
               logger.warn(e.getMessage(), e);
            }
         }
      } catch (Throwable e) {
         logger.warn(e.getMessage(), e);
      }
   }

   PageSubscriptionImpl(final PageCursorProvider cursorProvider,
                        final PagingStore pageStore,
                        final StorageManager store,
                        final Filter filter,
                        final long cursorId,
                        final boolean persistent) {
      this.pageStore = pageStore;
      this.store = store;
      this.cursorProvider = cursorProvider;
      this.cursorId = cursorId;
      this.filter = filter;
      this.persistent = persistent;
      this.counter = new PageSubscriptionCounterImpl(store, this, persistent, cursorId);
   }


   @Override
   public PagingStore getPagingStore() {
      return pageStore;
   }

   @Override
   public Queue getQueue() {
      return queue;
   }

   @Override
   public boolean isPaging() {
      return pageStore.isPaging();
   }

   @Override
   public void setQueue(Queue queue) {
      this.queue = queue;
   }

   @Override
   public void disableAutoCleanup() {
      autoCleanup = false;
   }

   @Override
   public void enableAutoCleanup() {
      autoCleanup = true;
   }

   public PageCursorProvider getProvider() {
      return cursorProvider;
   }

   @Override
   public void notEmpty() {
      synchronized (consumedPages) {
         this.empty = false;
      }

   }

   @Override
   public void bookmark(PagePosition position) throws Exception {
      PageCursorInfo cursorInfo = getPageInfo(position);

      if (position.getMessageNr() > 0) {
         cursorInfo.confirmed.addAndGet(position.getMessageNr());
      }

      confirmPosition(position);
   }

   @Override
   public long getMessageCount() {
      if (empty) {
         return 0;
      } else {
         return counter.getValue() - deliveredCount.get();
      }
   }

   @Override
   public long getPersistentSize() {
      if (empty) {
         return 0;
      } else {
         //A negative value could happen if an old journal was loaded that didn't have
         //size metrics for old records
         long messageSize = counter.getPersistentSize() - deliveredSize.get();
         return messageSize > 0 ? messageSize : 0;
      }
   }

   @Override
   public PageSubscriptionCounter getCounter() {
      return counter;
   }

   /**
    * A page marked as complete will be ignored until it's cleared.
    * <p>
    * Usually paging is a stream of messages but in certain scenarios (such as a pending prepared
    * TX) we may have big holes on the page streaming, and we will need to ignore such pages on the
    * cursor/subscription.
    */
   @Override
   public boolean reloadPageCompletion(PagePosition position) throws Exception {
      if (!pageStore.checkPageFileExists((int)position.getPageNr())) {
         return false;
      }
      // if the current page is complete, we must move it out of the way
      if (pageStore.getCurrentPage() != null &&
          pageStore.getCurrentPage().getPageId() == position.getPageNr()) {
         pageStore.forceAnotherPage();
      }
      PageCursorInfo info = new PageCursorInfo(position.getPageNr(), position.getMessageNr());
      info.setCompleteInfo(position);
      synchronized (consumedPages) {
         consumedPages.put(Long.valueOf(position.getPageNr()), info);
      }

      return true;
   }

   @Override
   public void scheduleCleanupCheck() {
      if (autoCleanup) {
         if (logger.isTraceEnabled()) {
            logger.trace("Scheduling cleanup", new Exception("trace"));
         }
         if (scheduledCleanupCount.get() > 2) {
            return;
         }

         scheduledCleanupCount.incrementAndGet();
         pageStore.execute(this::performCleanup);
      }
   }

   private void performCleanup() {
      try {
         if (autoCleanup) {
            cleanupEntries(false);
         }
      } catch (Exception e) {
         ActiveMQServerLogger.LOGGER.problemCleaningCursorPages(e);
      } finally {
         scheduledCleanupCount.decrementAndGet();
      }
   }

   @Override
   public void onPageModeCleared(Transaction tx) throws Exception {
      if (counter != null) {
         // this could be null on testcases
         counter.delete(tx);
      }
      this.empty = true;
   }

   /**
    * It will cleanup all the records for completed pages
    */
   @Override
   public void cleanupEntries(final boolean completeDelete) throws Exception {
      if (completeDelete) {
         counter.delete();
      }
      logger.trace(">>>>>>> cleanupEntries");
      try {
         Transaction tx = new TransactionImpl(store);

         boolean persist = false;

         final ArrayList<PageCursorInfo> completedPages = new ArrayList<>();

         // First get the completed pages using a lock
         synchronized (consumedPages) {
            // lastAckedPosition = null means no acks were done yet, so we are not ready to cleanup
            if (lastAckedPosition == null) {
               return;
            }

            for (Entry<Long, PageCursorInfo> entry : consumedPages.entrySet()) {
               PageCursorInfo info = entry.getValue();

               if (info.isDone() && !info.isPendingDelete()) {
                  Page currentPage = pageStore.getCurrentPage();

                  if (currentPage != null && entry.getKey() == pageStore.getCurrentPage().getPageId()) {
                     logger.tracef("We can't clear page %s 's the current page", entry.getKey());
                  } else {
                     if (logger.isTraceEnabled()) {
                        logger.tracef("cleanup marking page %s as complete", info.pageId);
                     }
                     info.setPendingDelete();
                     completedPages.add(entry.getValue());
                  }
               }
            }
         }

         for (PageCursorInfo infoPG : completedPages) {
            // HORNETQ-1017: There are a few cases where a pending transaction might set a big hole on the page system
            //               where we need to ignore these pages in case of a restart.
            //               for that reason when we delete complete ACKs we store a single record per page file that will
            //               be removed once the page file is deleted
            //               notice also that this record is added as part of the same transaction where the information is deleted.
            //               In case of a TX Failure (a crash on the server) this will be recovered on the next cleanup once the
            //               server is restarted.
            // first will mark the page as complete
            if (isPersistent()) {
               PagePosition completePage = new PagePositionImpl(infoPG.getPageId(), infoPG.getNumberOfMessages());
               infoPG.setCompleteInfo(completePage);
               store.storePageCompleteTransactional(tx.getID(), this.getId(), completePage);
               if (!persist) {
                  persist = true;
                  tx.setContainsPersistent();
               }
            }

            // it will delete the page ack records
            for (PagePosition pos : infoPG.acks.values()) {
               if (pos.getRecordID() >= 0) {
                  store.deleteCursorAcknowledgeTransactional(tx.getID(), pos.getRecordID());
                  if (!persist) {
                     // only need to set it once
                     tx.setContainsPersistent();
                     persist = true;
                  }
               }
            }

            infoPG.acks.clear();
            infoPG.removedReferences.clear();
         }

         tx.addOperation(new TransactionOperationAbstract() {

            @Override
            public void afterCommit(final Transaction tx1) {
               pageStore.execute(new Runnable() {

                  @Override
                  public void run() {
                     if (!completeDelete) {
                        cursorProvider.scheduleCleanup();
                     }
                  }
               });
            }
         });

         tx.commit();
      } finally {
         logger.trace("<<<<<< cleanupEntries");
      }

   }

   @Override
   public String toString() {
      return "PageSubscriptionImpl [cursorId=" + cursorId + ", queue=" + queue + ", filter = " + filter + "]";
   }

   @Override
   public PageIterator iterator() {
      return new CursorIterator();
   }

   @Override
   public PageIterator iterator(boolean browsing) {
      return new CursorIterator(browsing);
   }

   private boolean routed(PagedMessage message) {
      long id = getId();

      for (long qid : message.getQueueIDs()) {
         if (qid == id) {
            return true;
         }
      }
      return false;
   }

   @Override
   public void confirmPosition(final Transaction tx, final PagePosition position) throws Exception {
      // if the cursor is persistent
      if (persistent) {
         store.storeCursorAcknowledgeTransactional(tx.getID(), cursorId, position);
      }
      installTXCallback(tx, position);

   }

   private void confirmPosition(final Transaction tx, final PagePosition position, final long persistentSize) throws Exception {
      // if the cursor is persistent
      if (persistent) {
         store.storeCursorAcknowledgeTransactional(tx.getID(), cursorId, position);
      }
      installTXCallback(tx, position, persistentSize);
   }

   @Override
   public void ackTx(final Transaction tx, final PagedReference reference) throws Exception {
      //pre-calculate persistentSize
      final long persistentSize = getPersistentSize(reference);

      confirmPosition(tx, reference.getPagedMessage().newPositionObject(), persistentSize);

      counter.increment(tx, -1, -persistentSize);

      PageTransactionInfo txInfo = getPageTransaction(reference);
      if (txInfo != null) {
         txInfo.storeUpdate(store, pageStore.getPagingManager(), tx);
      }
   }

   @Override
   public void ack(final PagedReference reference) throws Exception {
      // Need to do the ACK and counter atomically (inside a TX) or the counter could get out of sync
      Transaction tx = new TransactionImpl(this.store);
      ackTx(tx, reference);
      tx.commit();
   }

   @Override
   public boolean contains(PagedReference ref) throws Exception {
      // We first verify if the message was routed to this queue
      boolean routed = false;

      for (long idRef : ref.getPagedMessage().getQueueIDs()) {
         if (idRef == this.cursorId) {
            routed = true;
            break;
         }
      }
      if (!routed) {
         return false;
      } else {
         // if it's been routed here, we have to verify if it was acked
         return !getPageInfo(ref.getPagedMessage().getPageNumber()).isAck(ref.getPagedMessage().getMessageNumber());
      }
   }

   @Override
   public void confirmPosition(final PagePosition position) throws Exception {
      // if we are dealing with a persistent cursor
      if (persistent) {
         store.storeCursorAcknowledge(cursorId, position);
      }

      store.afterCompleteOperations(new IOCallback() {
         volatile String error = "";

         @Override
         public void onError(final int errorCode, final String errorMessage) {
            error = " errorCode=" + errorCode + ", msg=" + errorMessage;
            ActiveMQServerLogger.LOGGER.pageSubscriptionError(this, error);
         }

         @Override
         public void done() {
            processACK(position);
         }

         @Override
         public String toString() {
            return IOCallback.class.getSimpleName() + "(" + PageSubscriptionImpl.class.getSimpleName() + ") " + error;
         }
      });
   }

   @Override
   public long getFirstPage() {
      synchronized (consumedPages) {
         if (empty && consumedPages.isEmpty()) {
            return -1;
         }
         long lastPageSeen = 0;
         for (Map.Entry<Long, PageCursorInfo> info : consumedPages.entrySet()) {
            lastPageSeen = info.getKey();
            if (!info.getValue().isDone() && !info.getValue().isPendingDelete()) {
               return info.getKey();
            }
         }
         return lastPageSeen;
      }

   }

   @Override
   public void addPendingDelivery(final PagedMessage pagedMessage) {
      PageCursorInfo info = getPageInfo(pagedMessage.getPageNumber());

      if (info != null) {
         info.incrementPendingTX();
      }
   }

   @Override
   public void removePendingDelivery(final PagedMessage pagedMessage) {
      PageCursorInfo info = getPageInfo(pagedMessage.getPageNumber());

      if (info != null) {
         info.decrementPendingTX();
      }
   }

   @Override
   public void redeliver(final PageIterator iterator, final PagedReference pagedReference) {
      iterator.redeliver(pagedReference);

      synchronized (consumedPages) {
         PageCursorInfo pageInfo = consumedPages.get(pagedReference.getPagedMessage().getPageNumber());
         if (pageInfo != null) {
            pageInfo.decrementPendingTX();
         } else {
            // this shouldn't really happen.
         }
      }
   }

   @Override
   public PagedMessage queryMessage(PagePosition pos) {
      try {
         Page page = pageStore.usePage(pos.getPageNr());

         if (page == null) {
            return null;
         }

         try {
            org.apache.activemq.artemis.utils.collections.LinkedList<PagedMessage> messages = page.getMessages();
            PagedMessage retMessage;
            if (pos.getMessageNr() < messages.size()) {
               retMessage = messages.get(pos.getMessageNr());
            } else {
               retMessage = null;
            }
            return retMessage;
         } finally {
            page.usageDown();
         }
      } catch (Exception e) {
         store.criticalError(e);
         throw new RuntimeException(e.getMessage(), e);
      }
   }

   /**
    * Theres no need to synchronize this method as it's only called from journal load on startup
    */
   @Override
   public void reloadACK(final PagePosition position) {
      if (recoveredACK == null) {
         recoveredACK = new LinkedList<>();
      }

      recoveredACK.add(position);
   }

   @Override
   public void reloadPreparedACK(final Transaction tx, final PagePosition position) {
      deliveredCount.incrementAndGet();
      installTXCallback(tx, position);
   }

   @Override
   public void positionIgnored(final PagePosition position) {
      processACK(position);
   }

   @Override
   public void lateDeliveryRollback(PagePosition position) {
      PageCursorInfo cursorInfo = processACK(position);
      cursorInfo.decrementPendingTX();
   }

   @Override
   public void forEachConsumedPage(Consumer<ConsumedPage> pageCleaner) {
      synchronized (consumedPages) {
         consumedPages.values().forEach(pageCleaner);
      }
   }


   @Override
   public boolean isComplete(long page) {
      logger.tracef("%s isComplete %d", this, page);
      synchronized (consumedPages) {
         if (empty && consumedPages.isEmpty()) {
            if (logger.isTraceEnabled()) {
               logger.tracef("isComplete(%d)::Subscription %s has empty=%s, consumedPages.isEmpty=%s", page, this, empty, consumedPages.isEmpty());
            }
            return true;
         }

         PageCursorInfo info = consumedPages.get(page);

         if (info == null && empty) {
            logger.tracef("isComplete(%d)::::Couldn't find info and it is empty", page);
            return true;
         } else {
            boolean isDone = info != null && info.isDone();
            if (logger.isTraceEnabled()) {
               logger.tracef("isComplete(%d):: found info=%s, isDone=%s", (Object) page, info, isDone);
            }
            return isDone;
         }
      }
   }

   /**
    * All the data associated with the cursor should go away here
    */
   @Override
   public void destroy() throws Exception {
      final long tx = store.generateID();
      try {

         boolean isPersistent = false;

         synchronized (consumedPages) {
            for (PageCursorInfo cursor : consumedPages.values()) {
               for (PagePosition info : cursor.acks.values()) {
                  if (info.getRecordID() >= 0) {
                     isPersistent = true;
                     store.deleteCursorAcknowledgeTransactional(tx, info.getRecordID());
                  }
               }
               PagePosition completeInfo = cursor.getCompleteInfo();
               if (completeInfo != null && completeInfo.getRecordID() >= 0) {
                  store.deletePageComplete(completeInfo.getRecordID());
                  cursor.setCompleteInfo(null);
               }
            }
         }

         if (isPersistent) {
            store.commit(tx);
         }

         cursorProvider.close(this);
      } catch (Exception e) {
         try {
            store.rollback(tx);
         } catch (Exception ignored) {
            // exception of the exception.. nothing that can be done here
         }
      }
   }

   @Override
   public long getId() {
      return cursorId;
   }

   @Override
   public boolean isPersistent() {
      return persistent;
   }

   @Override
   public void processReload() throws Exception {
      if (recoveredACK != null) {
         if (logger.isTraceEnabled()) {
            logger.trace("********** processing reload!!!!!!!");
         }
         Collections.sort(recoveredACK);

         long txDeleteCursorOnReload = -1;

         for (PagePosition pos : recoveredACK) {
            lastAckedPosition = pos;
            PageCursorInfo pageInfo = getPageInfo(pos);
            pageInfo.loadACK(pos);
         }

         if (txDeleteCursorOnReload >= 0) {
            store.commit(txDeleteCursorOnReload);
         }

         recoveredACK.clear();
         recoveredACK = null;
      }
   }

   @Override
   public void stop() {
   }

   @Override
   public void printDebug() {
      printDebug(toString());
   }

   public void printDebug(final String msg) {
      System.out.println("Debug information on PageCurorImpl- " + msg);
      for (PageCursorInfo info : consumedPages.values()) {
         System.out.println(info);
      }
   }

   @Override
   public void onDeletePage(Page deletedPage) throws Exception {
      logger.tracef("removing page %s", deletedPage);
      PageCursorInfo info;
      synchronized (consumedPages) {
         info = consumedPages.remove(Long.valueOf(deletedPage.getPageId()));
      }
      if (info != null) {
         PagePosition completeInfo = info.getCompleteInfo();
         if (completeInfo != null) {
            try {
               store.deletePageComplete(completeInfo.getRecordID());
            } catch (Exception e) {
               ActiveMQServerLogger.LOGGER.errorDeletingPageCompleteRecord(e);
            }
            info.setCompleteInfo(null);
         }
         for (PagePosition deleteInfo : info.acks.values()) {
            if (deleteInfo.getRecordID() >= 0) {
               try {
                  store.deleteCursorAcknowledge(deleteInfo.getRecordID());
               } catch (Exception e) {
                  ActiveMQServerLogger.LOGGER.errorDeletingPageCompleteRecord(e);
               }
            }
         }
         info.acks.clear();
      }
      deletedPage.usageExhaust();
   }

   @Override
   public void reloadPageInfo(long pageNr) {
      getPageInfo(pageNr);
   }

   private PageCursorInfo getPageInfo(final PagePosition pos) {
      return getPageInfo(pos.getPageNr());
   }

   public PageCursorInfo getPageInfo(final long pageNr) {
      synchronized (consumedPages) {
         PageCursorInfo pageInfo = consumedPages.get(pageNr);

         if (pageInfo == null) {
            pageInfo = new PageCursorInfo(pageNr);
            consumedPages.put(pageNr, pageInfo);
         }
         return pageInfo;
      }

   }


   private boolean match(final Message message) {
      if (filter == null) {
         return true;
      } else {
         return filter.match(message);
      }
   }


   // To be called only after the ACK has been processed and guaranteed to be on storage
   // The only exception is on non storage events such as not matching messages
   private PageCursorInfo processACK(final PagePosition pos) {
      if (lastAckedPosition == null || pos.compareTo(lastAckedPosition) > 0) {
         if (logger.isTraceEnabled()) {
            logger.trace("a new position is being processed as ACK");
         }
         if (lastAckedPosition != null && lastAckedPosition.getPageNr() != pos.getPageNr()) {
            if (logger.isTraceEnabled()) {
               logger.trace("Scheduling cleanup on pageSubscription for address = " + pageStore.getAddress() + " queue = " + this.getQueue().getName());
            }

            // there's a different page being acked, we will do the check right away
            if (autoCleanup) {
               scheduleCleanupCheck();
            }
         }
         lastAckedPosition = pos;
      }
      PageCursorInfo info = getPageInfo(pos);

      // This could be null if the page file was removed
      if (info == null) {
         // This could become null if the page file was deleted, or if the queue was removed maybe?
         // it's better to diagnose it (based on support tickets) instead of NPE
         ActiveMQServerLogger.LOGGER.nullPageCursorInfo(this.getPagingStore().getAddress().toString(), pos.toString(), cursorId);
      } else {
         info.addACK(pos);
      }

      return info;
   }

   private void installTXCallback(final Transaction tx, final PagePosition position) {
      installTXCallback(tx, position, -1);
   }

   /**
    * @param tx
    * @param position
    * @param persistentSize if negative it needs to be calculated on the fly
    */
   private void installTXCallback(final Transaction tx, final PagePosition position, final long persistentSize) {
      if (position.getRecordID() >= 0) {
         // It needs to persist, otherwise the cursor will return to the fist page position
         tx.setContainsPersistent();
      }

      PageCursorInfo info = getPageInfo(position);
      if (info != null) {
         logger.tracef("InstallTXCallback looking up pagePosition %s, result=%s", position, info);

         info.remove(position.getMessageNr());

         PageCursorTX cursorTX = (PageCursorTX) tx.getProperty(TransactionPropertyIndexes.PAGE_CURSOR_POSITIONS);

         if (cursorTX == null) {
            cursorTX = new PageCursorTX();
            tx.putProperty(TransactionPropertyIndexes.PAGE_CURSOR_POSITIONS, cursorTX);
            tx.addOperation(cursorTX);
         }

         cursorTX.addPositionConfirmation(this, position);
      }

   }

   private PageTransactionInfo getPageTransaction(final PagedReference reference) throws ActiveMQException {
      if (reference.getTransactionID() >= 0) {
         return pageStore.getPagingManager().getTransaction(reference.getTransactionID());
      } else {
         return null;
      }
   }

   /**
    * A callback from the PageCursorInfo. It will be called when all the messages on a page have been acked
    *
    * @param info
    */
   private void onPageDone(final PageCursorInfo info) {
      if (autoCleanup) {
         if (logger.isTraceEnabled()) {
            logger.tracef("onPageDone page %s", info.getPageId());
         }
         scheduleCleanupCheck();
      }
   }


   /**
    * This will hold information about the pending ACKs towards a page.
    * <p>
    * This instance will be released as soon as the entire page is consumed, releasing the memory at
    * that point The ref counts are increased also when a message is ignored for any reason.
    */
   public final class PageCursorInfo implements ConsumedPage {

      // Number of messages existent on this page
      private int numberOfMessages;

      private final long pageId;

      private IntObjectHashMap<PagePosition> acks = new IntObjectHashMap<>();

      // This will take DUMMY elements. This is used like a HashSet of Int
      private IntObjectHashMap<Object> removedReferences = new IntObjectHashMap<>();

      // There's a pending TX to add elements on this page
      // also can be used to prevent the page from being deleted too soon.
      private final AtomicInteger pendingTX = new AtomicInteger(0);

      // There's a pending delete on the async IO pipe
      // We're holding this object to avoid delete the pages before the IO is complete,
      // however we can't delete these records again
      private boolean pendingDelete;

      /**
       * This is to be set when all the messages are complete on a given page, and we cleanup the records that are marked on it
       */
      private PagePosition completePage;

      // We need a separate counter as the cursor 3124'gmay be ignoring certain values because of incomplete transactions or
      // expressions
      private final AtomicInteger confirmed = new AtomicInteger(0);

      public boolean isAck(int messageNumber) {
         return completePage != null || acks.get(messageNumber) != null;
      }

      @Override
      public String toString() {
         try {
            return "PageCursorInfo::pageNr=" + pageId +
               " numberOfMessage = " +
               numberOfMessages +
               ", confirmed = " +
               confirmed +
               ", isDone=" +
               this.isDone();
         } catch (Exception e) {
            return "PageCursorInfo::pageNr=" + pageId +
               " numberOfMessage = " +
               numberOfMessages +
               ", confirmed = " +
               confirmed +
               ", isDone=" +
               e.toString();
         }
      }

      private PageCursorInfo(final long pageId, final int numberOfMessages) {
         if (numberOfMessages < 0) {
            throw new IllegalStateException("numberOfMessages = " + numberOfMessages + " instead of being >=0");
         }
         this.pageId = pageId;
         this.numberOfMessages = numberOfMessages;
         logger.tracef("Created PageCursorInfo for pageNr=%d, numberOfMessages=%d, not live", pageId, numberOfMessages);
      }

      private PageCursorInfo(long pageId) {
         this.pageId = pageId;
         //given that is live, the exact value must be get directly from cache
         this.numberOfMessages = -1;
      }

      /**
       * @param completePage
       */
      public void setCompleteInfo(final PagePosition completePage) {
         logger.tracef("Setting up complete page %s on cursor %s on subscription %s", completePage, this, PageSubscriptionImpl.this);
         this.completePage = completePage;
      }

      public PagePosition getCompleteInfo() {
         return completePage;
      }

      @Override
      public boolean isDone() {
         if (logger.isTraceEnabled()) {
            logger.trace(PageSubscriptionImpl.this + "::PageCursorInfo(" + pageId + ")::isDone checking with completePage!=null->" + (completePage != null) + " getNumberOfMessages=" + getNumberOfMessages() + ", confirmed=" + confirmed.get() + " and pendingTX=" + pendingTX.get());

         }
         // in cases where the file was damaged it is possible to get more confirmed records than we actually had messages
         // for that case we set confirmed.get() >= getNumberOfMessages instead of ==
         return completePage != null || (confirmed.get() >= getNumberOfMessages() && pendingTX.get() == 0);
      }

      public boolean isPendingDelete() {
         return pendingDelete || completePage != null;
      }

      public void setPendingDelete() {
         pendingDelete = true;
      }

      /**
       * @return the pageId
       */
      @Override
      public long getPageId() {
         return pageId;
      }

      public void incrementPendingTX() {
         pendingTX.incrementAndGet();
      }

      public void decrementPendingTX() {
         pendingTX.decrementAndGet();
         checkDone();
      }

      public boolean isRemoved(final int messageNr) {
         return removedReferences.get(messageNr) != null;
      }

      public void remove(final int messageNr) {
         if (logger.isTraceEnabled()) {
            logger.tracef("PageCursor Removing messageNr %s on page %s", messageNr, pageId);
         }
         removedReferences.put(messageNr, DUMMY);
      }

      public void addACK(final PagePosition posACK) {

         if (logger.isTraceEnabled()) {
            try {
               logger.trace("numberOfMessages =  " + getNumberOfMessages() +
                               " confirmed =  " +
                               (confirmed.get() + 1) +
                               " pendingTX = " + pendingTX +
                               ", pageNr = " +
                               pageId + " posACK = " + posACK);
            } catch (Throwable ignored) {
               logger.debug(ignored.getMessage(), ignored);
            }
         }

         boolean added = internalAddACK(posACK);

         // Negative could mean a bookmark on the first element for the page (example -1)
         if (added && posACK.getMessageNr() >= 0) {
            confirmed.incrementAndGet();
            checkDone();
         }
      }

      // To be called during reload
      public void loadACK(final PagePosition posACK) {
         if (internalAddACK(posACK) && posACK.getMessageNr() >= 0) {
            confirmed.incrementAndGet();
         }
      }

      private boolean internalAddACK(final PagePosition position) {
         removedReferences.put(position.getMessageNr(), DUMMY);
         return acks.put(position.getMessageNr(), position) == null;
      }

      /**
       *
       */
      protected void checkDone() {
         if (isDone()) {
            onPageDone(this);
         }
      }

      private int getNumberOfMessages() {
         if (numberOfMessages < 0) {
            try {
               Page page = pageStore.usePage(pageId, false);

               if (page == null) {
                  page = pageStore.newPageObject(pageId);
                  numberOfMessages = page.readNumberOfMessages();
               } else {
                  try {
                     if (page.isOpen()) {
                        // if the file is still open (active) we don't cache the number of messages
                        return page.getNumberOfMessages();
                     } else {
                        this.numberOfMessages = page.getNumberOfMessages();
                     }
                  } finally {
                     page.usageDown();
                  }
               }
            } catch (Exception e) {
               store.criticalError(e);
               throw new RuntimeException(e.getMessage(), e);
            }
         }
         return numberOfMessages;
      }

      public int getPendingTx() {
         return pendingTX.get();
      }
   }

   private final class PageCursorTX extends TransactionOperationAbstract {

      private final Map<PageSubscriptionImpl, List<PagePosition>> pendingPositions = new HashMap<>();

      private void addPositionConfirmation(final PageSubscriptionImpl cursor, final PagePosition position) {
         List<PagePosition> list = pendingPositions.get(cursor);

         if (list == null) {
            list = new LinkedList<>();
            pendingPositions.put(cursor, list);
         }

         list.add(position);
      }

      @Override
      public void afterCommit(final Transaction tx) {
         for (Entry<PageSubscriptionImpl, List<PagePosition>> entry : pendingPositions.entrySet()) {
            PageSubscriptionImpl cursor = entry.getKey();

            List<PagePosition> positions = entry.getValue();

            for (PagePosition confirmed : positions) {
               cursor.processACK(confirmed);
               cursor.deliveredCount.decrementAndGet();
               cursor.deliveredSize.addAndGet(-confirmed.getPersistentSize());
            }

         }
      }

      @Override
      public List<MessageReference> getRelatedMessageReferences() {
         return Collections.emptyList();
      }

   }

   private class CursorIterator implements PageIterator {

      private Page currentPage;
      private LinkedListIterator<PagedMessage> currentPageIterator;

      private void initPage(long page) {
         try {
            if (currentPage != null) {
               currentPage.usageDown();
            }
            if (currentPageIterator != null) {
               currentPageIterator.close();
            }
            currentPage = pageStore.usePage(page);
            if (logger.isTraceEnabled()) {
               logger.tracef("CursorIterator: getting page " + page + " which will contain " + currentPage.getNumberOfMessages());
            }
            currentPageIterator = currentPage.iterator();
         } catch (Exception e) {
            store.criticalError(e);
            throw new IllegalStateException(e.getMessage(), e);
         }
      }

      private PagedReference currentDelivery = null;

      private volatile PagedReference lastDelivery = null;

      private final boolean browsing;

      // We only store the position for redeliveries. They will be read from the SoftCache again during delivery.
      private final java.util.Queue<PagedReference> redeliveries = new LinkedList<>();

      /**
       * next element taken on hasNext test.
       * it has to be delivered on next next operation
       */
      private volatile PagedReference cachedNext;

      private CursorIterator(boolean browsing) {
         this.browsing = browsing;
      }

      private CursorIterator() {
         this.browsing = false;
      }

      @Override
      public void redeliver(PagedReference reference) {
         synchronized (redeliveries) {
            redeliveries.add(reference);
         }
      }

      @Override
      public void repeat() {
         cachedNext = lastDelivery;
      }

      @Override
      public synchronized PagedReference next() {
         try {
            if (cachedNext != null) {
               currentDelivery = cachedNext;
               cachedNext = null;
               return currentDelivery;
            }

            if (currentPage == null) {
               logger.tracef("CursorIterator::next initializing first page as %s", pageStore.getFirstPage());
               initPage(pageStore.getFirstPage());
            }

            currentDelivery = moveNext();
            return currentDelivery;
         } catch (Throwable e) {
            logger.warn(e.getMessage(), e);
            store.criticalError(e);
            throw new RuntimeException(e.getMessage(), e);
         }
      }

      private PagedReference moveNext() {
         synchronized (PageSubscriptionImpl.this) {
            boolean match = false;

            PagedReference message;
            long timeout = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(DELIVERY_TIMEOUT);

            do {
               if (System.nanoTime() - timeout > 0) {
                  return RETRY_MARK;
               }

               synchronized (redeliveries) {
                  PagedReference redelivery = redeliveries.poll();

                  if (redelivery != null) {
                     return redelivery;
                  } else {
                     lastDelivery = null;
                  }
               }
               message = internalGetNext();

               if (message == null) {
                  break;
               }

               boolean valid = true;
               boolean ignored = false;

               // Validate the scenarios where the message should be considered not valid even to be considered

               // 1st... is it routed?

               valid = routed(message.getPagedMessage());
               if (!valid) {
                  logger.tracef("CursorIterator::message %s was deemed invalid, marking it to ignore", message.getPagedMessage());
                  ignored = true;
               }

               PageCursorInfo info = getPageInfo(message.getPagedMessage().getPageNumber());

               if (!browsing && info != null && (info.isRemoved(message.getPagedMessage().getMessageNumber()) || info.getCompleteInfo() != null)) {
                  if (logger.isTraceEnabled()) {
                     boolean removed = info.isRemoved(message.getPagedMessage().getMessageNumber());
                     logger.tracef("CursorIterator::Message from page %s # %s isRemoved=%s", message.getPagedMessage().getPageNumber(), message.getPagedMessage().getMessageNumber(), (Boolean)removed);
                  }
                  continue;
               }

               if (info != null && info.isAck(message.getPagedMessage().getMessageNumber())) {
                  logger.tracef("CursorIterator::message %s is acked, moving next", message);
                  continue;
               }

               // 2nd ... if TX, is it committed?
               if (valid && message.getPagedMessage().getTransactionID() >= 0) {
                  PageTransactionInfo tx = pageStore.getPagingManager().getTransaction(message.getPagedMessage().getTransactionID());
                  if (tx == null) {
                     ActiveMQServerLogger.LOGGER.pageSubscriptionCouldntLoad(message.getPagedMessage().getTransactionID(), message.getPagedMessage().newPositionObject(), pageStore.getAddress(), queue.getName());
                     valid = false;
                     ignored = true;
                  } else {
                     if (tx.deliverAfterCommit(CursorIterator.this, PageSubscriptionImpl.this, message)) {
                        valid = false;
                        ignored = false;
                     }
                  }
               }

               // 3rd... was it previously removed?
               if (valid) {
                  // We don't create a PageCursorInfo unless we are doing a write operation (ack or removing)
                  // Say you have a Browser that will only read the files... there's no need to control PageCursors is
                  // nothing
                  // is being changed. That's why the false is passed as a parameter here

                  if (!browsing && info != null && info.isRemoved(message.getPagedMessage().getMessageNumber())) {
                     valid = false;
                  }
               }

               if (valid) {
                  if (browsing) {
                     match = match(message.getMessage());
                  } else {
                     // if not browsing, we will just trust the routing on the queue
                     match = true;
                  }
               } else if (!browsing && ignored) {
                  positionIgnored(message.getPagedMessage().newPositionObject());
               }
            }
            while (!match);

            if (message != null) {
               lastDelivery = message;
            }

            return message;
         }
      }


      private PagedReference internalGetNext() {
         for (;;) {
            PagedMessage message = currentPageIterator.hasNext() ? currentPageIterator.next() : null;
            logger.tracef("CursorIterator::internalGetNext:: new reference %s", message);
            if (message != null) {
               return cursorProvider.newReference(message, PageSubscriptionImpl.this);
            }

            if (currentPage.getPageId() < pageStore.getCurrentWritingPage()) {
               if (logger.isTraceEnabled()) {
                  logger.tracef("CursorIterator::internalGetNext:: moving to currentPage %s", currentPage.getPageId() + 1);
               }
               initPage(currentPage.getPageId() + 1);
            } else {
               return null;
            }
         }
      }


      @Override
      public synchronized NextResult tryNext() {
         // if an unbehaved program called hasNext twice before next, we only cache it once.
         if (cachedNext != null) {
            return NextResult.hasElements;
         }

         if (!pageStore.isPaging()) {
            return NextResult.noElements;
         }

         PagedReference pagedReference = next();
         if (pagedReference == RETRY_MARK) {
            return NextResult.retry;
         } else {
            cachedNext = pagedReference;
            return cachedNext == null ? NextResult.noElements : NextResult.hasElements;
         }
      }

      /**
       * QueueImpl::deliver could be calling hasNext while QueueImpl.depage could be using next and hasNext as well.
       * It would be a rare race condition but I would prefer avoiding that scenario
       */
      @Override
      public synchronized boolean hasNext() {
         NextResult status;
         while ((status = tryNext()) == NextResult.retry) {
         }
         return status == NextResult.hasElements ? true : false;
      }

      @Override
      public void remove() {
         deliveredCount.incrementAndGet();
         PagedReference delivery = currentDelivery;
         if (delivery != null) {
            PageCursorInfo info = PageSubscriptionImpl.this.getPageInfo(delivery.getPagedMessage().getPageNumber());
            if (info != null) {
               info.remove(delivery.getPagedMessage().getMessageNumber());
            }
         }
      }

      @Override
      public void close() {
         Page toClose = currentPage;
         if (toClose != null) {
            toClose.usageDown();
         }
         currentPage = null;
      }
   }

   /**
    * @return the deliveredCount
    */
   @Override
   public long getDeliveredCount() {
      return deliveredCount.get();
   }

   /**
    * @return the deliveredSize
    */
   @Override
   public long getDeliveredSize() {
      return deliveredSize.get();
   }

   @Override
   public void incrementDeliveredSize(long size) {
      deliveredSize.addAndGet(size);
   }

   private long getPersistentSize(PagedMessage msg) {
      try {
         return msg != null && msg.getPersistentSize() > 0 ? msg.getPersistentSize() : 0;
      } catch (ActiveMQException e) {
         logger.warn("Error computing persistent size of message: " + msg, e);
         return 0;
      }
   }

   private long getPersistentSize(PagedReference ref) {
      try {
         return ref != null && ref.getPersistentSize() > 0 ? ref.getPersistentSize() : 0;
      } catch (ActiveMQException e) {
         logger.warn("Error computing persistent size of message: " + ref, e);
         return 0;
      }
   }
}
