activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From clebertsuco...@apache.org
Subject [6/8] activemq-artemis git commit: ARTEMIS-27 / ARTEMIS-340 Add JDBC Storage Manager
Date Wed, 13 Jan 2016 14:49:45 GMT
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/64f74acd/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
new file mode 100644
index 0000000..be457c4
--- /dev/null
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
@@ -0,0 +1,1828 @@
+/*
+ * 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.persistence.impl.journal;
+
+import javax.transaction.xa.Xid;
+import java.io.File;
+import java.io.FileInputStream;
+import java.security.AccessController;
+import java.security.DigestInputStream;
+import java.security.InvalidParameterException;
+import java.security.MessageDigest;
+import java.security.PrivilegedAction;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
+import org.apache.activemq.artemis.api.core.Pair;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.filter.Filter;
+import org.apache.activemq.artemis.core.io.IOCallback;
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.journal.JournalLoadInformation;
+import org.apache.activemq.artemis.core.journal.PreparedTransactionInfo;
+import org.apache.activemq.artemis.core.journal.RecordInfo;
+import org.apache.activemq.artemis.core.paging.PageTransactionInfo;
+import org.apache.activemq.artemis.core.paging.PagingManager;
+import org.apache.activemq.artemis.core.paging.PagingStore;
+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.PagedReferenceImpl;
+import org.apache.activemq.artemis.core.paging.impl.PageTransactionInfoImpl;
+import org.apache.activemq.artemis.core.persistence.GroupingInfo;
+import org.apache.activemq.artemis.core.persistence.OperationContext;
+import org.apache.activemq.artemis.core.persistence.QueueBindingInfo;
+import org.apache.activemq.artemis.core.persistence.StorageManager;
+import org.apache.activemq.artemis.core.persistence.config.PersistedAddressSetting;
+import org.apache.activemq.artemis.core.persistence.config.PersistedRoles;
+import org.apache.activemq.artemis.core.persistence.impl.PageCountPending;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.CursorAckRecordEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.DeleteEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.DeliveryCountUpdateEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.DuplicateIDEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.FinishPageMessageOperation;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.GroupingEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.HeuristicCompletionEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.LargeMessageEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.PageCountPendingImpl;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.PageCountRecord;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.PageCountRecordInc;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.PageUpdateTXEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.PendingLargeMessageEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.PersistentQueueBindingEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.RefEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.ScheduledDeliveryEncoding;
+import org.apache.activemq.artemis.core.persistence.impl.journal.codec.XidEncoding;
+import org.apache.activemq.artemis.core.postoffice.Binding;
+import org.apache.activemq.artemis.core.postoffice.DuplicateIDCache;
+import org.apache.activemq.artemis.core.postoffice.PostOffice;
+import org.apache.activemq.artemis.core.server.ActiveMQMessageBundle;
+import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
+import org.apache.activemq.artemis.core.server.LargeServerMessage;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.RouteContextList;
+import org.apache.activemq.artemis.core.server.ServerMessage;
+import org.apache.activemq.artemis.core.server.group.impl.GroupBinding;
+import org.apache.activemq.artemis.core.server.impl.JournalLoader;
+import org.apache.activemq.artemis.core.server.impl.ServerMessageImpl;
+import org.apache.activemq.artemis.core.transaction.ResourceManager;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.apache.activemq.artemis.core.transaction.TransactionPropertyIndexes;
+import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
+import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
+import org.apache.activemq.artemis.utils.Base64;
+import org.apache.activemq.artemis.utils.ExecutorFactory;
+import org.apache.activemq.artemis.utils.IDGenerator;
+
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ACKNOWLEDGE_CURSOR;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.ADD_LARGE_MESSAGE_PENDING;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.DUPLICATE_ID;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COUNTER_INC;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE;
+import static org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME;
+
+/**
+ * Controls access to the journals and other storage files such as the ones used to store pages and
+ * large messages.  This class must control writing of any non-transient data, as it is the key point
+ * for synchronizing any replicating backup server.
+ * <p>
+ * Using this class also ensures that locks are acquired in the right order, avoiding dead-locks.
+ */
+public abstract class AbstractJournalStorageManager implements StorageManager {
+
+   public enum JournalContent {
+      BINDINGS((byte) 0), MESSAGES((byte) 1);
+
+      public final byte typeByte;
+
+      JournalContent(byte b) {
+         typeByte = b;
+      }
+
+      public static JournalContent getType(byte type) {
+         if (MESSAGES.typeByte == type)
+            return MESSAGES;
+         if (BINDINGS.typeByte == type)
+            return BINDINGS;
+         throw new InvalidParameterException("invalid byte: " + type);
+      }
+   }
+
+   private static final long CHECKPOINT_BATCH_SIZE = Integer.MAX_VALUE;
+
+   protected Semaphore pageMaxConcurrentIO;
+
+   protected BatchingIDGenerator idGenerator;
+
+   protected final ReentrantReadWriteLock storageManagerLock = new ReentrantReadWriteLock(true);
+
+   protected Journal messageJournal;
+
+   protected Journal bindingsJournal;
+
+   protected volatile boolean started;
+
+   /**
+    * Used to create Operation Contexts
+    */
+   private final ExecutorFactory executorFactory;
+
+   final Executor executor;
+
+   ExecutorService singleThreadExecutor;
+
+   private final boolean syncTransactional;
+
+   private final boolean syncNonTransactional;
+
+   protected int perfBlastPages = -1;
+
+   protected boolean journalLoaded = false;
+
+   private final IOCriticalErrorListener ioCriticalErrorListener;
+
+   protected final Configuration config;
+
+   // Persisted core configuration
+   protected final Map<SimpleString, PersistedRoles> mapPersistedRoles = new ConcurrentHashMap<>();
+
+   protected final Map<SimpleString, PersistedAddressSetting> mapPersistedAddressSettings = new ConcurrentHashMap<>();
+
+   protected final Set<Long> largeMessagesToDelete = new HashSet<>();
+
+   public AbstractJournalStorageManager(final Configuration config, final ExecutorFactory executorFactory) {
+      this(config, executorFactory, null);
+   }
+
+   public AbstractJournalStorageManager(Configuration config,
+                                        ExecutorFactory executorFactory,
+                                        IOCriticalErrorListener criticalErrorListener) {
+      this.executorFactory = executorFactory;
+
+      this.ioCriticalErrorListener = criticalErrorListener;
+
+      this.config = config;
+
+      executor = executorFactory.getExecutor();
+
+      syncNonTransactional = config.isJournalSyncNonTransactional();
+      syncTransactional = config.isJournalSyncTransactional();
+
+      init(config, criticalErrorListener);
+
+      idGenerator = new BatchingIDGenerator(0, CHECKPOINT_BATCH_SIZE, this);
+   }
+
+   /**
+    * Called during initialization.  Used by implementations to setup Journals, Stores etc...
+    * @param config
+    * @param criticalErrorListener
+    */
+   protected abstract void init(Configuration config, IOCriticalErrorListener criticalErrorListener);
+
+   @Override
+   public void criticalError(Throwable error) {
+      ioCriticalErrorListener.onIOException(error, error.getMessage(), null);
+   }
+
+   @Override
+   public void clearContext() {
+      OperationContextImpl.clearContext();
+   }
+
+   public static String md5(File file) {
+      try {
+         byte[] buffer = new byte[1 << 4];
+         MessageDigest md = MessageDigest.getInstance("MD5");
+
+         FileInputStream is = new FileInputStream(file);
+         DigestInputStream is2 = new DigestInputStream(is, md);
+         while (is2.read(buffer) > 0) {
+            continue;
+         }
+         byte[] digest = md.digest();
+         is.close();
+         is2.close();
+         return Base64.encodeBytes(digest);
+      }
+      catch (Exception e) {
+         throw new RuntimeException(e);
+      }
+   }
+
+   public IDGenerator getIDGenerator() {
+      return idGenerator;
+   }
+
+   @Override
+   public final void waitOnOperations() throws Exception {
+      if (!started) {
+         ActiveMQServerLogger.LOGGER.serverIsStopped();
+         throw new IllegalStateException("Server is stopped");
+      }
+      waitOnOperations(0);
+   }
+
+   @Override
+   public final boolean waitOnOperations(final long timeout) throws Exception {
+      if (!started) {
+         ActiveMQServerLogger.LOGGER.serverIsStopped();
+         throw new IllegalStateException("Server is stopped");
+      }
+      return getContext().waitCompletion(timeout);
+   }
+
+   public OperationContext getContext() {
+      return OperationContextImpl.getContext(executorFactory);
+   }
+
+   public void setContext(final OperationContext context) {
+      OperationContextImpl.setContext(context);
+   }
+
+   public Executor getSingleThreadExecutor() {
+      return singleThreadExecutor;
+   }
+
+   public OperationContext newSingleThreadContext() {
+      return newContext(singleThreadExecutor);
+   }
+
+   public OperationContext newContext(final Executor executor1) {
+      return new OperationContextImpl(executor1);
+   }
+
+   public void afterCompleteOperations(final IOCallback run) {
+      getContext().executeOnCompletion(run);
+   }
+
+   public long generateID() {
+      return idGenerator.generateID();
+   }
+
+   public long getCurrentID() {
+      return idGenerator.getCurrentID();
+   }
+
+   // Non transactional operations
+
+
+   public void confirmPendingLargeMessageTX(final Transaction tx, long messageID, long recordID) throws Exception {
+      readLock();
+      try {
+         installLargeMessageConfirmationOnTX(tx, recordID);
+         messageJournal.appendDeleteRecordTransactional(tx.getID(), recordID, new DeleteEncoding(JournalRecordIds.ADD_LARGE_MESSAGE_PENDING, messageID));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   /**
+    * We don't need messageID now but we are likely to need it we ever decide to support a database
+    */
+   public void confirmPendingLargeMessage(long recordID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendDeleteRecord(recordID, true, getContext());
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void storeMessage(final ServerMessage message) throws Exception {
+      if (message.getMessageID() <= 0) {
+         // Sanity check only... this shouldn't happen unless there is a bug
+         throw ActiveMQMessageBundle.BUNDLE.messageIdNotAssigned();
+      }
+
+      readLock();
+      try {
+         // Note that we don't sync, the add reference that comes immediately after will sync if
+         // appropriate
+
+         if (message.isLargeMessage()) {
+            messageJournal.appendAddRecord(message.getMessageID(), JournalRecordIds.ADD_LARGE_MESSAGE, new LargeMessageEncoding((LargeServerMessage) message), false, getContext(false));
+         }
+         else {
+            messageJournal.appendAddRecord(message.getMessageID(), JournalRecordIds.ADD_MESSAGE, message, false, getContext(false));
+         }
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void storeReference(final long queueID, final long messageID, final boolean last) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendUpdateRecord(messageID, JournalRecordIds.ADD_REF, new RefEncoding(queueID), last && syncNonTransactional, getContext(last && syncNonTransactional));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   @Override
+   public void readLock() {
+      storageManagerLock.readLock().lock();
+   }
+
+   @Override
+   public void readUnLock() {
+      storageManagerLock.readLock().unlock();
+   }
+
+   public void storeAcknowledge(final long queueID, final long messageID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendUpdateRecord(messageID, JournalRecordIds.ACKNOWLEDGE_REF, new RefEncoding(queueID), syncNonTransactional, getContext(syncNonTransactional));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void storeCursorAcknowledge(long queueID, PagePosition position) throws Exception {
+      readLock();
+      try {
+         long ackID = idGenerator.generateID();
+         position.setRecordID(ackID);
+         messageJournal.appendAddRecord(ackID, JournalRecordIds.ACKNOWLEDGE_CURSOR, new CursorAckRecordEncoding(queueID, position), syncNonTransactional, getContext(syncNonTransactional));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deleteMessage(final long messageID) throws Exception {
+      readLock();
+      try {
+         // Messages are deleted on postACK, one after another.
+         // If these deletes are synchronized, we would build up messages on the Executor
+         // increasing chances of losing deletes.
+         // The StorageManager should verify messages without references
+         messageJournal.appendDeleteRecord(messageID, false, getContext(false));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void updateScheduledDeliveryTime(final MessageReference ref) throws Exception {
+      ScheduledDeliveryEncoding encoding = new ScheduledDeliveryEncoding(ref.getScheduledDeliveryTime(), ref.getQueue().getID());
+      readLock();
+      try {
+         messageJournal.appendUpdateRecord(ref.getMessage().getMessageID(), JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME, encoding, syncNonTransactional, getContext(syncNonTransactional));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void storeDuplicateID(final SimpleString address, final byte[] duplID, final long recordID) throws Exception {
+      readLock();
+      try {
+         DuplicateIDEncoding encoding = new DuplicateIDEncoding(address, duplID);
+
+         messageJournal.appendAddRecord(recordID, JournalRecordIds.DUPLICATE_ID, encoding, syncNonTransactional, getContext(syncNonTransactional));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deleteDuplicateID(final long recordID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendDeleteRecord(recordID, syncNonTransactional, getContext(syncNonTransactional));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   // Transactional operations
+
+   public void storeMessageTransactional(final long txID, final ServerMessage message) throws Exception {
+      if (message.getMessageID() <= 0) {
+         throw ActiveMQMessageBundle.BUNDLE.messageIdNotAssigned();
+      }
+
+      readLock();
+      try {
+         if (message.isLargeMessage()) {
+            messageJournal.appendAddRecordTransactional(txID, message.getMessageID(), JournalRecordIds.ADD_LARGE_MESSAGE, new LargeMessageEncoding(((LargeServerMessage) message)));
+         }
+         else {
+            messageJournal.appendAddRecordTransactional(txID, message.getMessageID(), JournalRecordIds.ADD_MESSAGE, message);
+         }
+
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void storePageTransaction(final long txID, final PageTransactionInfo pageTransaction) throws Exception {
+      readLock();
+      try {
+         pageTransaction.setRecordID(generateID());
+         messageJournal.appendAddRecordTransactional(txID, pageTransaction.getRecordID(), JournalRecordIds.PAGE_TRANSACTION, pageTransaction);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void updatePageTransaction(final long txID,
+                                     final PageTransactionInfo pageTransaction,
+                                     final int depages) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendUpdateRecordTransactional(txID, pageTransaction.getRecordID(), JournalRecordIds.PAGE_TRANSACTION, new PageUpdateTXEncoding(pageTransaction.getTransactionID(), depages));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void updatePageTransaction(final PageTransactionInfo pageTransaction, final int depages) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendUpdateRecord(pageTransaction.getRecordID(), JournalRecordIds.PAGE_TRANSACTION, new PageUpdateTXEncoding(pageTransaction.getTransactionID(), depages), syncNonTransactional, getContext(syncNonTransactional));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void storeReferenceTransactional(final long txID, final long queueID, final long messageID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendUpdateRecordTransactional(txID, messageID, JournalRecordIds.ADD_REF, new RefEncoding(queueID));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void storeAcknowledgeTransactional(final long txID,
+                                             final long queueID,
+                                             final long messageID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendUpdateRecordTransactional(txID, messageID, JournalRecordIds.ACKNOWLEDGE_REF, new RefEncoding(queueID));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void storeCursorAcknowledgeTransactional(long txID, long queueID, PagePosition position) throws Exception {
+      readLock();
+      try {
+         long ackID = idGenerator.generateID();
+         position.setRecordID(ackID);
+         messageJournal.appendAddRecordTransactional(txID, ackID, JournalRecordIds.ACKNOWLEDGE_CURSOR, new CursorAckRecordEncoding(queueID, position));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void storePageCompleteTransactional(long txID, long queueID, PagePosition position) throws Exception {
+      long recordID = idGenerator.generateID();
+      position.setRecordID(recordID);
+      messageJournal.appendAddRecordTransactional(txID, recordID, JournalRecordIds.PAGE_CURSOR_COMPLETE, new CursorAckRecordEncoding(queueID, position));
+   }
+
+   public void deletePageComplete(long ackID) throws Exception {
+      messageJournal.appendDeleteRecord(ackID, false);
+   }
+
+   public void deleteCursorAcknowledgeTransactional(long txID, long ackID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendDeleteRecordTransactional(txID, ackID);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deleteCursorAcknowledge(long ackID) throws Exception {
+      messageJournal.appendDeleteRecord(ackID, false);
+   }
+
+   public long storeHeuristicCompletion(final Xid xid, final boolean isCommit) throws Exception {
+      readLock();
+      try {
+         long id = generateID();
+
+         messageJournal.appendAddRecord(id, JournalRecordIds.HEURISTIC_COMPLETION, new HeuristicCompletionEncoding(xid, isCommit), true, getContext(true));
+         return id;
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deleteHeuristicCompletion(final long id) throws Exception {
+      readLock();
+      try {
+
+         messageJournal.appendDeleteRecord(id, true, getContext(true));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deletePageTransactional(final long recordID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendDeleteRecord(recordID, false);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void updateScheduledDeliveryTimeTransactional(final long txID, final MessageReference ref) throws Exception {
+      ScheduledDeliveryEncoding encoding = new ScheduledDeliveryEncoding(ref.getScheduledDeliveryTime(), ref.getQueue().getID());
+      readLock();
+      try {
+
+         messageJournal.appendUpdateRecordTransactional(txID, ref.getMessage().getMessageID(), JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME, encoding);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void prepare(final long txID, final Xid xid) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendPrepareRecord(txID, new XidEncoding(xid), syncTransactional, getContext(syncTransactional));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void commit(final long txID) throws Exception {
+      commit(txID, true);
+   }
+
+   public void commitBindings(final long txID) throws Exception {
+      bindingsJournal.appendCommitRecord(txID, true);
+   }
+
+   public void rollbackBindings(final long txID) throws Exception {
+      // no need to sync, it's going away anyways
+      bindingsJournal.appendRollbackRecord(txID, false);
+   }
+
+   public void commit(final long txID, final boolean lineUpContext) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendCommitRecord(txID, syncTransactional, getContext(syncTransactional), lineUpContext);
+         if (!lineUpContext && !syncTransactional) {
+            /**
+             * If {@code lineUpContext == false}, it means that we have previously lined up a
+             * context somewhere else (specifically see @{link TransactionImpl#asyncAppendCommit}),
+             * hence we need to mark it as done even if {@code syncTransactional = false} as in this
+             * case {@code getContext(syncTransactional=false)} would pass a dummy context to the
+             * {@code messageJournal.appendCommitRecord(...)} call above.
+             */
+            getContext(true).done();
+         }
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void rollback(final long txID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendRollbackRecord(txID, syncTransactional, getContext(syncTransactional));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void storeDuplicateIDTransactional(final long txID,
+                                             final SimpleString address,
+                                             final byte[] duplID,
+                                             final long recordID) throws Exception {
+      DuplicateIDEncoding encoding = new DuplicateIDEncoding(address, duplID);
+
+      readLock();
+      try {
+         messageJournal.appendAddRecordTransactional(txID, recordID, JournalRecordIds.DUPLICATE_ID, encoding);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void updateDuplicateIDTransactional(final long txID,
+                                              final SimpleString address,
+                                              final byte[] duplID,
+                                              final long recordID) throws Exception {
+      DuplicateIDEncoding encoding = new DuplicateIDEncoding(address, duplID);
+
+      readLock();
+      try {
+         messageJournal.appendUpdateRecordTransactional(txID, recordID, JournalRecordIds.DUPLICATE_ID, encoding);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deleteDuplicateIDTransactional(final long txID, final long recordID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendDeleteRecordTransactional(txID, recordID);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   // Other operations
+
+   public void updateDeliveryCount(final MessageReference ref) throws Exception {
+      // no need to store if it's the same value
+      // otherwise the journal will get OME in case of lots of redeliveries
+      if (ref.getDeliveryCount() == ref.getPersistedCount()) {
+         return;
+      }
+
+      ref.setPersistedCount(ref.getDeliveryCount());
+      DeliveryCountUpdateEncoding updateInfo = new DeliveryCountUpdateEncoding(ref.getQueue().getID(), ref.getDeliveryCount());
+
+      readLock();
+      try {
+         messageJournal.appendUpdateRecord(ref.getMessage().getMessageID(), JournalRecordIds.UPDATE_DELIVERY_COUNT, updateInfo, syncNonTransactional, getContext(syncNonTransactional));
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void storeAddressSetting(PersistedAddressSetting addressSetting) throws Exception {
+      deleteAddressSetting(addressSetting.getAddressMatch());
+      readLock();
+      try {
+         long id = idGenerator.generateID();
+         addressSetting.setStoreId(id);
+         bindingsJournal.appendAddRecord(id, JournalRecordIds.ADDRESS_SETTING_RECORD, addressSetting, true);
+         mapPersistedAddressSettings.put(addressSetting.getAddressMatch(), addressSetting);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public List<PersistedAddressSetting> recoverAddressSettings() throws Exception {
+      return new ArrayList<>(mapPersistedAddressSettings.values());
+   }
+
+   public List<PersistedRoles> recoverPersistedRoles() throws Exception {
+      return new ArrayList<>(mapPersistedRoles.values());
+   }
+
+   public void storeSecurityRoles(PersistedRoles persistedRoles) throws Exception {
+
+      deleteSecurityRoles(persistedRoles.getAddressMatch());
+      readLock();
+      try {
+         final long id = idGenerator.generateID();
+         persistedRoles.setStoreId(id);
+         bindingsJournal.appendAddRecord(id, JournalRecordIds.SECURITY_RECORD, persistedRoles, true);
+         mapPersistedRoles.put(persistedRoles.getAddressMatch(), persistedRoles);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   @Override
+   public void storeID(final long journalID, final long id) throws Exception {
+      readLock();
+      try {
+         bindingsJournal.appendAddRecord(journalID, JournalRecordIds.ID_COUNTER_RECORD, BatchingIDGenerator.createIDEncodingSupport(id), true);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   @Override
+   public void deleteID(long journalD) throws Exception {
+      readLock();
+      try {
+         bindingsJournal.appendDeleteRecord(journalD, false);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deleteAddressSetting(SimpleString addressMatch) throws Exception {
+      PersistedAddressSetting oldSetting = mapPersistedAddressSettings.remove(addressMatch);
+      if (oldSetting != null) {
+         readLock();
+         try {
+            bindingsJournal.appendDeleteRecord(oldSetting.getStoreId(), false);
+         }
+         finally {
+            readUnLock();
+         }
+      }
+   }
+
+   public void deleteSecurityRoles(SimpleString addressMatch) throws Exception {
+      PersistedRoles oldRoles = mapPersistedRoles.remove(addressMatch);
+      if (oldRoles != null) {
+         readLock();
+         try {
+            bindingsJournal.appendDeleteRecord(oldRoles.getStoreId(), false);
+         }
+         finally {
+            readUnLock();
+         }
+      }
+   }
+
+   @Override
+   public JournalLoadInformation loadMessageJournal(final PostOffice postOffice,
+                                                    final PagingManager pagingManager,
+                                                    final ResourceManager resourceManager,
+                                                    Map<Long, QueueBindingInfo> queueInfos,
+                                                    final Map<SimpleString, List<Pair<byte[], Long>>> duplicateIDMap,
+                                                    final Set<Pair<Long, Long>> pendingLargeMessages,
+                                                    List<PageCountPending> pendingNonTXPageCounter,
+                                                    final JournalLoader journalLoader) throws Exception {
+      List<RecordInfo> records = new ArrayList<>();
+
+      List<PreparedTransactionInfo> preparedTransactions = new ArrayList<>();
+
+      Map<Long, ServerMessage> messages = new HashMap<>();
+      readLock();
+      try {
+
+         JournalLoadInformation info = messageJournal.load(records, preparedTransactions, new LargeMessageTXFailureCallback(this, messages));
+
+         ArrayList<LargeServerMessage> largeMessages = new ArrayList<>();
+
+         Map<Long, Map<Long, AddMessageRecord>> queueMap = new HashMap<>();
+
+         Map<Long, PageSubscription> pageSubscriptions = new HashMap<>();
+
+         final int totalSize = records.size();
+
+         for (int reccount = 0; reccount < totalSize; reccount++) {
+            // It will show log.info only with large journals (more than 1 million records)
+            if (reccount > 0 && reccount % 1000000 == 0) {
+               long percent = (long) ((((double) reccount) / ((double) totalSize)) * 100f);
+
+               ActiveMQServerLogger.LOGGER.percentLoaded(percent);
+            }
+
+            RecordInfo record = records.get(reccount);
+            byte[] data = record.data;
+
+            ActiveMQBuffer buff = ActiveMQBuffers.wrappedBuffer(data);
+
+            byte recordType = record.getUserRecordType();
+
+            switch (recordType) {
+               case JournalRecordIds.ADD_LARGE_MESSAGE_PENDING: {
+                  PendingLargeMessageEncoding pending = new PendingLargeMessageEncoding();
+
+                  pending.decode(buff);
+
+                  if (pendingLargeMessages != null) {
+                     // it could be null on tests, and we don't need anything on that case
+                     pendingLargeMessages.add(new Pair<>(record.id, pending.largeMessageID));
+                  }
+                  break;
+               }
+               case JournalRecordIds.ADD_LARGE_MESSAGE: {
+                  LargeServerMessage largeMessage = parseLargeMessage(messages, buff);
+
+                  messages.put(record.id, largeMessage);
+
+                  largeMessages.add(largeMessage);
+
+                  break;
+               }
+               case JournalRecordIds.ADD_MESSAGE: {
+                  ServerMessage message = new ServerMessageImpl(record.id, 50);
+
+                  message.decode(buff);
+
+                  messages.put(record.id, message);
+
+                  break;
+               }
+               case JournalRecordIds.ADD_REF: {
+                  long messageID = record.id;
+
+                  RefEncoding encoding = new RefEncoding();
+
+                  encoding.decode(buff);
+
+                  Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
+
+                  if (queueMessages == null) {
+                     queueMessages = new LinkedHashMap<>();
+
+                     queueMap.put(encoding.queueID, queueMessages);
+                  }
+
+                  ServerMessage message = messages.get(messageID);
+
+                  if (message == null) {
+                     ActiveMQServerLogger.LOGGER.cannotFindMessage(record.id);
+                  }
+                  else {
+                     queueMessages.put(messageID, new AddMessageRecord(message));
+                  }
+
+                  break;
+               }
+               case JournalRecordIds.ACKNOWLEDGE_REF: {
+                  long messageID = record.id;
+
+                  RefEncoding encoding = new RefEncoding();
+
+                  encoding.decode(buff);
+
+                  Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
+
+                  if (queueMessages == null) {
+                     ActiveMQServerLogger.LOGGER.journalCannotFindQueue(encoding.queueID, messageID);
+                  }
+                  else {
+                     AddMessageRecord rec = queueMessages.remove(messageID);
+
+                     if (rec == null) {
+                        ActiveMQServerLogger.LOGGER.cannotFindMessage(messageID);
+                     }
+                  }
+
+                  break;
+               }
+               case JournalRecordIds.UPDATE_DELIVERY_COUNT: {
+                  long messageID = record.id;
+
+                  DeliveryCountUpdateEncoding encoding = new DeliveryCountUpdateEncoding();
+
+                  encoding.decode(buff);
+
+                  Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
+
+                  if (queueMessages == null) {
+                     ActiveMQServerLogger.LOGGER.journalCannotFindQueueDelCount(encoding.queueID);
+                  }
+                  else {
+                     AddMessageRecord rec = queueMessages.get(messageID);
+
+                     if (rec == null) {
+                        ActiveMQServerLogger.LOGGER.journalCannotFindMessageDelCount(messageID);
+                     }
+                     else {
+                        rec.setDeliveryCount(encoding.count);
+                     }
+                  }
+
+                  break;
+               }
+               case JournalRecordIds.PAGE_TRANSACTION: {
+                  if (record.isUpdate) {
+                     PageUpdateTXEncoding pageUpdate = new PageUpdateTXEncoding();
+
+                     pageUpdate.decode(buff);
+
+                     PageTransactionInfo pageTX = pagingManager.getTransaction(pageUpdate.pageTX);
+
+                     pageTX.onUpdate(pageUpdate.recods, null, null);
+                  }
+                  else {
+                     PageTransactionInfoImpl pageTransactionInfo = new PageTransactionInfoImpl();
+
+                     pageTransactionInfo.decode(buff);
+
+                     pageTransactionInfo.setRecordID(record.id);
+
+                     pagingManager.addTransaction(pageTransactionInfo);
+                  }
+
+                  break;
+               }
+               case JournalRecordIds.SET_SCHEDULED_DELIVERY_TIME: {
+                  long messageID = record.id;
+
+                  ScheduledDeliveryEncoding encoding = new ScheduledDeliveryEncoding();
+
+                  encoding.decode(buff);
+
+                  Map<Long, AddMessageRecord> queueMessages = queueMap.get(encoding.queueID);
+
+                  if (queueMessages == null) {
+                     ActiveMQServerLogger.LOGGER.journalCannotFindQueueScheduled(encoding.queueID, messageID);
+                  }
+                  else {
+
+                     AddMessageRecord rec = queueMessages.get(messageID);
+
+                     if (rec == null) {
+                        ActiveMQServerLogger.LOGGER.cannotFindMessage(messageID);
+                     }
+                     else {
+                        rec.setScheduledDeliveryTime(encoding.scheduledDeliveryTime);
+                     }
+                  }
+
+                  break;
+               }
+               case JournalRecordIds.DUPLICATE_ID: {
+                  DuplicateIDEncoding encoding = new DuplicateIDEncoding();
+
+                  encoding.decode(buff);
+
+                  List<Pair<byte[], Long>> ids = duplicateIDMap.get(encoding.address);
+
+                  if (ids == null) {
+                     ids = new ArrayList<>();
+
+                     duplicateIDMap.put(encoding.address, ids);
+                  }
+
+                  ids.add(new Pair<>(encoding.duplID, record.id));
+
+                  break;
+               }
+               case JournalRecordIds.HEURISTIC_COMPLETION: {
+                  HeuristicCompletionEncoding encoding = new HeuristicCompletionEncoding();
+                  encoding.decode(buff);
+                  resourceManager.putHeuristicCompletion(record.id, encoding.xid, encoding.isCommit);
+                  break;
+               }
+               case JournalRecordIds.ACKNOWLEDGE_CURSOR: {
+                  CursorAckRecordEncoding encoding = new CursorAckRecordEncoding();
+                  encoding.decode(buff);
+
+                  encoding.position.setRecordID(record.id);
+
+                  PageSubscription sub = locateSubscription(encoding.queueID, pageSubscriptions, queueInfos, pagingManager);
+
+                  if (sub != null) {
+                     sub.reloadACK(encoding.position);
+                  }
+                  else {
+                     ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloading(encoding.queueID);
+                     messageJournal.appendDeleteRecord(record.id, false);
+
+                  }
+
+                  break;
+               }
+               case JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE: {
+                  PageCountRecord encoding = new PageCountRecord();
+
+                  encoding.decode(buff);
+
+                  PageSubscription sub = locateSubscription(encoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
+
+                  if (sub != null) {
+                     sub.getCounter().loadValue(record.id, encoding.getValue());
+                  }
+                  else {
+                     ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloadingPage(encoding.getQueueID());
+                     messageJournal.appendDeleteRecord(record.id, false);
+                  }
+
+                  break;
+               }
+
+               case JournalRecordIds.PAGE_CURSOR_COUNTER_INC: {
+                  PageCountRecordInc encoding = new PageCountRecordInc();
+
+                  encoding.decode(buff);
+
+                  PageSubscription sub = locateSubscription(encoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
+
+                  if (sub != null) {
+                     sub.getCounter().loadInc(record.id, encoding.getValue());
+                  }
+                  else {
+                     ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloadingPageCursor(encoding.getQueueID());
+                     messageJournal.appendDeleteRecord(record.id, false);
+                  }
+
+                  break;
+               }
+
+               case JournalRecordIds.PAGE_CURSOR_COMPLETE: {
+                  CursorAckRecordEncoding encoding = new CursorAckRecordEncoding();
+                  encoding.decode(buff);
+
+                  encoding.position.setRecordID(record.id);
+
+                  PageSubscription sub = locateSubscription(encoding.queueID, pageSubscriptions, queueInfos, pagingManager);
+
+                  if (sub != null) {
+                     sub.reloadPageCompletion(encoding.position);
+                  }
+                  else {
+                     ActiveMQServerLogger.LOGGER.cantFindQueueOnPageComplete(encoding.queueID);
+                     messageJournal.appendDeleteRecord(record.id, false);
+                  }
+
+                  break;
+               }
+
+               case JournalRecordIds.PAGE_CURSOR_PENDING_COUNTER: {
+
+                  PageCountPendingImpl pendingCountEncoding = new PageCountPendingImpl();
+                  pendingCountEncoding.decode(buff);
+                  pendingCountEncoding.setID(record.id);
+
+                  // This can be null on testcases not interested on this outcome
+                  if (pendingNonTXPageCounter != null) {
+                     pendingNonTXPageCounter.add(pendingCountEncoding);
+                  }
+                  break;
+               }
+
+               default: {
+                  throw new IllegalStateException("Invalid record type " + recordType);
+               }
+            }
+
+            // This will free up memory sooner. The record is not needed any more
+            // and its byte array would consume memory during the load process even though it's not necessary any longer
+            // what would delay processing time during load
+            records.set(reccount, null);
+         }
+
+         // Release the memory as soon as not needed any longer
+         records.clear();
+         records = null;
+
+         journalLoader.handleAddMessage(queueMap);
+
+         loadPreparedTransactions(postOffice, pagingManager, resourceManager, queueInfos, preparedTransactions, duplicateIDMap, pageSubscriptions, pendingLargeMessages, journalLoader);
+
+         for (PageSubscription sub : pageSubscriptions.values()) {
+            sub.getCounter().processReload();
+         }
+
+         for (LargeServerMessage msg : largeMessages) {
+            if (msg.getRefCount() == 0) {
+               ActiveMQServerLogger.LOGGER.largeMessageWithNoRef(msg.getMessageID());
+               msg.decrementDelayDeletionCount();
+            }
+         }
+
+         journalLoader.handleNoMessageReferences(messages);
+
+         // To recover positions on Iterators
+         if (pagingManager != null) {
+            // it could be null on certain tests that are not dealing with paging
+            // This could also be the case in certain embedded conditions
+            pagingManager.processReload();
+         }
+
+         if (perfBlastPages != -1) {
+            messageJournal.perfBlast(perfBlastPages);
+         }
+
+         journalLoader.postLoad(messageJournal, resourceManager, duplicateIDMap);
+         journalLoaded = true;
+         return info;
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   /**
+    * @param queueID
+    * @param pageSubscriptions
+    * @param queueInfos
+    * @return
+    */
+   private static PageSubscription locateSubscription(final long queueID,
+                                                      final Map<Long, PageSubscription> pageSubscriptions,
+                                                      final Map<Long, QueueBindingInfo> queueInfos,
+                                                      final PagingManager pagingManager) throws Exception {
+
+      PageSubscription subs = pageSubscriptions.get(queueID);
+      if (subs == null) {
+         QueueBindingInfo queueInfo = queueInfos.get(queueID);
+
+         if (queueInfo != null) {
+            SimpleString address = queueInfo.getAddress();
+            PagingStore store = pagingManager.getPageStore(address);
+            subs = store.getCursorProvider().getSubscription(queueID);
+            pageSubscriptions.put(queueID, subs);
+         }
+      }
+
+      return subs;
+   }
+
+   // grouping handler operations
+   public void addGrouping(final GroupBinding groupBinding) throws Exception {
+      GroupingEncoding groupingEncoding = new GroupingEncoding(groupBinding.getId(), groupBinding.getGroupId(), groupBinding.getClusterName());
+      readLock();
+      try {
+         bindingsJournal.appendAddRecord(groupBinding.getId(), JournalRecordIds.GROUP_RECORD, groupingEncoding, true);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deleteGrouping(long tx, final GroupBinding groupBinding) throws Exception {
+      readLock();
+      try {
+         bindingsJournal.appendDeleteRecordTransactional(tx, groupBinding.getId());
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   // BindingsImpl operations
+
+   public void addQueueBinding(final long tx, final Binding binding) throws Exception {
+      Queue queue = (Queue) binding.getBindable();
+
+      Filter filter = queue.getFilter();
+
+      SimpleString filterString = filter == null ? null : filter.getFilterString();
+
+      PersistentQueueBindingEncoding bindingEncoding = new PersistentQueueBindingEncoding(queue.getName(), binding.getAddress(), filterString, queue.getUser(), queue.isAutoCreated());
+
+      readLock();
+      try {
+         bindingsJournal.appendAddRecordTransactional(tx, binding.getID(), JournalRecordIds.QUEUE_BINDING_RECORD, bindingEncoding);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deleteQueueBinding(long tx, final long queueBindingID) throws Exception {
+      readLock();
+      try {
+         bindingsJournal.appendDeleteRecordTransactional(tx, queueBindingID);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public long storePageCounterInc(long txID, long queueID, int value) throws Exception {
+      readLock();
+      try {
+         long recordID = idGenerator.generateID();
+         messageJournal.appendAddRecordTransactional(txID, recordID, JournalRecordIds.PAGE_CURSOR_COUNTER_INC, new PageCountRecordInc(queueID, value));
+         return recordID;
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public long storePageCounterInc(long queueID, int value) throws Exception {
+      readLock();
+      try {
+         final long recordID = idGenerator.generateID();
+         messageJournal.appendAddRecord(recordID, JournalRecordIds.PAGE_CURSOR_COUNTER_INC, new PageCountRecordInc(queueID, value), true, getContext());
+         return recordID;
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   @Override
+   public long storePageCounter(long txID, long queueID, long value) throws Exception {
+      readLock();
+      try {
+         final long recordID = idGenerator.generateID();
+         messageJournal.appendAddRecordTransactional(txID, recordID, JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE, new PageCountRecord(queueID, value));
+         return recordID;
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   @Override
+   public long storePendingCounter(final long queueID, final long pageID, final int inc) throws Exception {
+      readLock();
+      try {
+         final long recordID = idGenerator.generateID();
+         PageCountPendingImpl pendingInc = new PageCountPendingImpl(queueID, pageID, inc);
+         // We must guarantee the record sync before we actually write on the page otherwise we may get out of sync
+         // on the counter
+         messageJournal.appendAddRecord(recordID, JournalRecordIds.PAGE_CURSOR_PENDING_COUNTER, pendingInc, true);
+         return recordID;
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deleteIncrementRecord(long txID, long recordID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendDeleteRecordTransactional(txID, recordID);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deletePageCounter(long txID, long recordID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendDeleteRecordTransactional(txID, recordID);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void deletePendingPageCounter(long txID, long recordID) throws Exception {
+      readLock();
+      try {
+         messageJournal.appendDeleteRecordTransactional(txID, recordID);
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public JournalLoadInformation loadBindingJournal(final List<QueueBindingInfo> queueBindingInfos,
+                                                    final List<GroupingInfo> groupingInfos) throws Exception {
+      List<RecordInfo> records = new ArrayList<RecordInfo>();
+
+      List<PreparedTransactionInfo> preparedTransactions = new ArrayList<PreparedTransactionInfo>();
+
+      JournalLoadInformation bindingsInfo = bindingsJournal.load(records, preparedTransactions, null);
+
+      for (RecordInfo record : records) {
+         long id = record.id;
+
+         ActiveMQBuffer buffer = ActiveMQBuffers.wrappedBuffer(record.data);
+
+         byte rec = record.getUserRecordType();
+
+         if (rec == JournalRecordIds.QUEUE_BINDING_RECORD) {
+            PersistentQueueBindingEncoding bindingEncoding = newBindingEncoding(id, buffer);
+
+            queueBindingInfos.add(bindingEncoding);
+         }
+         else if (rec == JournalRecordIds.ID_COUNTER_RECORD) {
+            idGenerator.loadState(record.id, buffer);
+         }
+         else if (rec == JournalRecordIds.GROUP_RECORD) {
+            GroupingEncoding encoding = newGroupEncoding(id, buffer);
+            groupingInfos.add(encoding);
+         }
+         else if (rec == JournalRecordIds.ADDRESS_SETTING_RECORD) {
+            PersistedAddressSetting setting = newAddressEncoding(id, buffer);
+            mapPersistedAddressSettings.put(setting.getAddressMatch(), setting);
+         }
+         else if (rec == JournalRecordIds.SECURITY_RECORD) {
+            PersistedRoles roles = newSecurityRecord(id, buffer);
+            mapPersistedRoles.put(roles.getAddressMatch(), roles);
+         }
+         else {
+            throw new IllegalStateException("Invalid record type " + rec);
+         }
+      }
+
+      // This will instruct the IDGenerator to beforeStop old records
+      idGenerator.cleanup();
+
+      return bindingsInfo;
+   }
+
+   public void lineUpContext() {
+      readLock();
+      try {
+         messageJournal.lineUpContext(getContext());
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   // ActiveMQComponent implementation
+   // ------------------------------------------------------
+
+   protected abstract void beforeStart() throws Exception;
+
+   public synchronized void start() throws Exception {
+      if (started) {
+         return;
+      }
+
+      beforeStart();
+
+      singleThreadExecutor = Executors.newSingleThreadExecutor(AccessController.doPrivileged(new PrivilegedAction<ActiveMQThreadFactory>() {
+         @Override
+         public ActiveMQThreadFactory run() {
+            return new ActiveMQThreadFactory("ActiveMQ-IO-SingleThread", true, JournalStorageManager.class.getClassLoader());
+         }
+      }));
+
+      bindingsJournal.start();
+
+      messageJournal.start();
+
+      started = true;
+   }
+
+   public void stop() throws Exception {
+      stop(false);
+   }
+
+   @Override
+   public synchronized void persistIdGenerator() {
+      if (journalLoaded && idGenerator != null) {
+         // Must call close to make sure last id is persisted
+         idGenerator.persistCurrentID();
+      }
+   }
+
+   /**
+    * Assumption is that this is only called with a writeLock on the StorageManager.
+    */
+   protected abstract void performCachedLargeMessageDeletes();
+
+   public synchronized void stop(boolean ioCriticalError) throws Exception {
+      if (!started) {
+         return;
+      }
+
+      if (!ioCriticalError) {
+         performCachedLargeMessageDeletes();
+         // Must call close to make sure last id is persisted
+         if (journalLoaded && idGenerator != null)
+            idGenerator.persistCurrentID();
+      }
+
+      final CountDownLatch latch = new CountDownLatch(1);
+      executor.execute(new Runnable() {
+         @Override
+         public void run() {
+            latch.countDown();
+         }
+      });
+
+      latch.await(30, TimeUnit.SECONDS);
+
+      beforeStop();
+
+      bindingsJournal.stop();
+
+      messageJournal.stop();
+
+      singleThreadExecutor.shutdown();
+
+      journalLoaded = false;
+
+      started = false;
+   }
+
+   protected abstract void beforeStop() throws Exception;
+
+   public synchronized boolean isStarted() {
+      return started;
+   }
+
+   /**
+    * TODO: Is this still being used ?
+    */
+   public JournalLoadInformation[] loadInternalOnly() throws Exception {
+      readLock();
+      try {
+         JournalLoadInformation[] info = new JournalLoadInformation[2];
+         info[0] = bindingsJournal.loadInternalOnly();
+         info[1] = messageJournal.loadInternalOnly();
+
+         return info;
+      }
+      finally {
+         readUnLock();
+      }
+   }
+
+   public void beforePageRead() throws Exception {
+      if (pageMaxConcurrentIO != null) {
+         pageMaxConcurrentIO.acquire();
+      }
+   }
+
+   public void afterPageRead() throws Exception {
+      if (pageMaxConcurrentIO != null) {
+         pageMaxConcurrentIO.release();
+      }
+   }
+
+   // Public -----------------------------------------------------------------------------------
+
+   public Journal getMessageJournal() {
+      return messageJournal;
+   }
+
+   public Journal getBindingsJournal() {
+      return bindingsJournal;
+   }
+
+   // Package protected ---------------------------------------------
+
+   protected void confirmLargeMessage(final LargeServerMessage largeServerMessage) {
+      if (largeServerMessage.getPendingRecordID() >= 0) {
+         try {
+            confirmPendingLargeMessage(largeServerMessage.getPendingRecordID());
+            largeServerMessage.setPendingRecordID(-1);
+         }
+         catch (Exception e) {
+            ActiveMQServerLogger.LOGGER.warn(e.getMessage(), e);
+         }
+      }
+   }
+
+   protected abstract LargeServerMessage parseLargeMessage(Map<Long, ServerMessage> messages,
+                                                           ActiveMQBuffer buff) throws Exception;
+
+   private void loadPreparedTransactions(final PostOffice postOffice,
+                                         final PagingManager pagingManager,
+                                         final ResourceManager resourceManager,
+                                         final Map<Long, QueueBindingInfo> queueInfos,
+                                         final List<PreparedTransactionInfo> preparedTransactions,
+                                         final Map<SimpleString, List<Pair<byte[], Long>>> duplicateIDMap,
+                                         final Map<Long, PageSubscription> pageSubscriptions,
+                                         final Set<Pair<Long, Long>> pendingLargeMessages,
+                                         JournalLoader journalLoader) throws Exception {
+      // recover prepared transactions
+      for (PreparedTransactionInfo preparedTransaction : preparedTransactions) {
+         XidEncoding encodingXid = new XidEncoding(preparedTransaction.getExtraData());
+
+         Xid xid = encodingXid.xid;
+
+         Transaction tx = new TransactionImpl(preparedTransaction.getId(), xid, this);
+
+         List<MessageReference> referencesToAck = new ArrayList<MessageReference>();
+
+         Map<Long, ServerMessage> messages = new HashMap<Long, ServerMessage>();
+
+         // Use same method as load message journal to prune out acks, so they don't get added.
+         // Then have reacknowledge(tx) methods on queue, which needs to add the page size
+
+         // first get any sent messages for this tx and recreate
+         for (RecordInfo record : preparedTransaction.getRecords()) {
+            byte[] data = record.data;
+
+            ActiveMQBuffer buff = ActiveMQBuffers.wrappedBuffer(data);
+
+            byte recordType = record.getUserRecordType();
+
+            switch (recordType) {
+               case JournalRecordIds.ADD_LARGE_MESSAGE: {
+                  messages.put(record.id, parseLargeMessage(messages, buff));
+
+                  break;
+               }
+               case JournalRecordIds.ADD_MESSAGE: {
+                  ServerMessage message = new ServerMessageImpl(record.id, 50);
+
+                  message.decode(buff);
+
+                  messages.put(record.id, message);
+
+                  break;
+               }
+               case JournalRecordIds.ADD_REF: {
+                  long messageID = record.id;
+
+                  RefEncoding encoding = new RefEncoding();
+
+                  encoding.decode(buff);
+
+                  ServerMessage message = messages.get(messageID);
+
+                  if (message == null) {
+                     throw new IllegalStateException("Cannot find message with id " + messageID);
+                  }
+
+                  journalLoader.handlePreparedSendMessage(message, tx, encoding.queueID);
+
+                  break;
+               }
+               case JournalRecordIds.ACKNOWLEDGE_REF: {
+                  long messageID = record.id;
+
+                  RefEncoding encoding = new RefEncoding();
+
+                  encoding.decode(buff);
+
+                  journalLoader.handlePreparedAcknowledge(messageID, referencesToAck, encoding.queueID);
+
+                  break;
+               }
+               case JournalRecordIds.PAGE_TRANSACTION: {
+
+                  PageTransactionInfo pageTransactionInfo = new PageTransactionInfoImpl();
+
+                  pageTransactionInfo.decode(buff);
+
+                  if (record.isUpdate) {
+                     PageTransactionInfo pgTX = pagingManager.getTransaction(pageTransactionInfo.getTransactionID());
+                     pgTX.reloadUpdate(this, pagingManager, tx, pageTransactionInfo.getNumberOfMessages());
+                  }
+                  else {
+                     pageTransactionInfo.setCommitted(false);
+
+                     tx.putProperty(TransactionPropertyIndexes.PAGE_TRANSACTION, pageTransactionInfo);
+
+                     pagingManager.addTransaction(pageTransactionInfo);
+
+                     tx.addOperation(new FinishPageMessageOperation());
+                  }
+
+                  break;
+               }
+               case SET_SCHEDULED_DELIVERY_TIME: {
+                  // Do nothing - for prepared txs, the set scheduled delivery time will only occur in a send in which
+                  // case the message will already have the header for the scheduled delivery time, so no need to do
+                  // anything.
+
+                  break;
+               }
+               case DUPLICATE_ID: {
+                  // We need load the duplicate ids at prepare time too
+                  DuplicateIDEncoding encoding = new DuplicateIDEncoding();
+
+                  encoding.decode(buff);
+
+                  DuplicateIDCache cache = postOffice.getDuplicateIDCache(encoding.address);
+
+                  cache.load(tx, encoding.duplID);
+
+                  break;
+               }
+               case ACKNOWLEDGE_CURSOR: {
+                  CursorAckRecordEncoding encoding = new CursorAckRecordEncoding();
+                  encoding.decode(buff);
+
+                  encoding.position.setRecordID(record.id);
+
+                  PageSubscription sub = locateSubscription(encoding.queueID, pageSubscriptions, queueInfos, pagingManager);
+
+                  if (sub != null) {
+                     sub.reloadPreparedACK(tx, encoding.position);
+                     referencesToAck.add(new PagedReferenceImpl(encoding.position, null, sub));
+                  }
+                  else {
+                     ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloadingACK(encoding.queueID);
+                  }
+                  break;
+               }
+               case PAGE_CURSOR_COUNTER_VALUE: {
+                  ActiveMQServerLogger.LOGGER.journalPAGEOnPrepared();
+
+                  break;
+               }
+
+               case PAGE_CURSOR_COUNTER_INC: {
+                  PageCountRecordInc encoding = new PageCountRecordInc();
+
+                  encoding.decode(buff);
+
+                  PageSubscription sub = locateSubscription(encoding.getQueueID(), pageSubscriptions, queueInfos, pagingManager);
+
+                  if (sub != null) {
+                     sub.getCounter().applyIncrementOnTX(tx, record.id, encoding.getValue());
+                     sub.notEmpty();
+                  }
+                  else {
+                     ActiveMQServerLogger.LOGGER.journalCannotFindQueueReloadingACK(encoding.getQueueID());
+                  }
+
+                  break;
+               }
+
+               default: {
+                  ActiveMQServerLogger.LOGGER.journalInvalidRecordType(recordType);
+               }
+            }
+         }
+
+         for (RecordInfo recordDeleted : preparedTransaction.getRecordsToDelete()) {
+            byte[] data = recordDeleted.data;
+
+            if (data.length > 0) {
+               ActiveMQBuffer buff = ActiveMQBuffers.wrappedBuffer(data);
+               byte b = buff.readByte();
+
+               switch (b) {
+                  case ADD_LARGE_MESSAGE_PENDING: {
+                     long messageID = buff.readLong();
+                     if (!pendingLargeMessages.remove(new Pair<Long, Long>(recordDeleted.id, messageID))) {
+                        ActiveMQServerLogger.LOGGER.largeMessageNotFound(recordDeleted.id);
+                     }
+                     installLargeMessageConfirmationOnTX(tx, recordDeleted.id);
+                     break;
+                  }
+                  default:
+                     ActiveMQServerLogger.LOGGER.journalInvalidRecordTypeOnPreparedTX(b);
+               }
+            }
+
+         }
+
+         journalLoader.handlePreparedTransaction(tx, referencesToAck, xid, resourceManager);
+      }
+   }
+
+   OperationContext getContext(final boolean sync) {
+      if (sync) {
+         return getContext();
+      }
+      else {
+         return DummyOperationContext.getInstance();
+      }
+   }
+
+   // Inner Classes
+   // ----------------------------------------------------------------------------
+
+   private static final class DummyOperationContext implements OperationContext {
+
+      private static DummyOperationContext instance = new DummyOperationContext();
+
+      public static OperationContext getInstance() {
+         return DummyOperationContext.instance;
+      }
+
+      public void executeOnCompletion(final IOCallback runnable) {
+         // There are no executeOnCompletion calls while using the DummyOperationContext
+         // However we keep the code here for correctness
+         runnable.done();
+      }
+
+      public void replicationDone() {
+      }
+
+      public void replicationLineUp() {
+      }
+
+      public void storeLineUp() {
+      }
+
+      public void done() {
+      }
+
+      public void onError(final int errorCode, final String errorMessage) {
+      }
+
+      public void waitCompletion() {
+      }
+
+      public boolean waitCompletion(final long timeout) {
+         return true;
+      }
+
+      public void pageSyncLineUp() {
+      }
+
+      public void pageSyncDone() {
+      }
+   }
+
+   /**
+    * @param id
+    * @param buffer
+    * @return
+    */
+   protected static PersistedRoles newSecurityRecord(long id, ActiveMQBuffer buffer) {
+      PersistedRoles roles = new PersistedRoles();
+      roles.decode(buffer);
+      roles.setStoreId(id);
+      return roles;
+   }
+
+   /**
+    * @param id
+    * @param buffer
+    * @return
+    */
+   static PersistedAddressSetting newAddressEncoding(long id, ActiveMQBuffer buffer) {
+      PersistedAddressSetting setting = new PersistedAddressSetting();
+      setting.decode(buffer);
+      setting.setStoreId(id);
+      return setting;
+   }
+
+   /**
+    * @param id
+    * @param buffer
+    * @return
+    */
+   static GroupingEncoding newGroupEncoding(long id, ActiveMQBuffer buffer) {
+      GroupingEncoding encoding = new GroupingEncoding();
+      encoding.decode(buffer);
+      encoding.setId(id);
+      return encoding;
+   }
+
+   /**
+    * @param id
+    * @param buffer
+    * @return
+    */
+   protected static PersistentQueueBindingEncoding newBindingEncoding(long id, ActiveMQBuffer buffer) {
+      PersistentQueueBindingEncoding bindingEncoding = new PersistentQueueBindingEncoding();
+
+      bindingEncoding.decode(buffer);
+
+      bindingEncoding.setId(id);
+      return bindingEncoding;
+   }
+
+   @Override
+   public boolean addToPage(PagingStore store,
+                            ServerMessage msg,
+                            Transaction tx,
+                            RouteContextList listCtx) throws Exception {
+      /**
+       * Exposing the read-lock here is an encapsulation violation done in order to keep the code
+       * simpler. The alternative would be to add a second method, say 'verifyPaging', to
+       * PagingStore.
+       * <p>
+       * Adding this second method would also be more surprise prone as it would require a certain
+       * calling order.
+       * <p>
+       * The reasoning is that exposing the lock is more explicit and therefore `less bad`.
+       */
+      return store.page(msg, tx, listCtx, storageManagerLock.readLock());
+   }
+
+   private void installLargeMessageConfirmationOnTX(Transaction tx, long recordID) {
+      TXLargeMessageConfirmationOperation txoper = (TXLargeMessageConfirmationOperation) tx.getProperty(TransactionPropertyIndexes.LARGE_MESSAGE_CONFIRMATIONS);
+      if (txoper == null) {
+         txoper = new TXLargeMessageConfirmationOperation(this);
+         tx.putProperty(TransactionPropertyIndexes.LARGE_MESSAGE_CONFIRMATIONS, txoper);
+      }
+      txoper.confirmedMessages.add(recordID);
+   }
+}

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/64f74acd/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AddMessageRecord.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AddMessageRecord.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AddMessageRecord.java
index fdae483..3ca38e3 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AddMessageRecord.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AddMessageRecord.java
@@ -26,11 +26,9 @@ public final class AddMessageRecord {
 
    final ServerMessage message;
 
-   // mtaylor (Added to compile)
-   public long scheduledDeliveryTime;
+   private long scheduledDeliveryTime;
 
-   // mtaylor (Added to compile)
-   public int deliveryCount;
+   private int deliveryCount;
 
    public ServerMessage getMessage() {
       return message;
@@ -44,4 +42,11 @@ public final class AddMessageRecord {
       return deliveryCount;
    }
 
+   public void setScheduledDeliveryTime(long scheduledDeliveryTime) {
+      this.scheduledDeliveryTime = scheduledDeliveryTime;
+   }
+
+   public void setDeliveryCount(int deliveryCount) {
+      this.deliveryCount = deliveryCount;
+   }
 }

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/64f74acd/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JDBCJournalStorageManager.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JDBCJournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JDBCJournalStorageManager.java
new file mode 100644
index 0000000..4616e78
--- /dev/null
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JDBCJournalStorageManager.java
@@ -0,0 +1,88 @@
+/*
+ * 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.persistence.impl.journal;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.config.storage.DatabaseStorageConfiguration;
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.jdbc.store.journal.JDBCJournalImpl;
+import org.apache.activemq.artemis.utils.ExecutorFactory;
+
+public class JDBCJournalStorageManager extends JournalStorageManager {
+
+   public JDBCJournalStorageManager(Configuration config, ExecutorFactory executorFactory) {
+      super(config, executorFactory);
+   }
+
+   public JDBCJournalStorageManager(final Configuration config,
+                                final ExecutorFactory executorFactory,
+                                final IOCriticalErrorListener criticalErrorListener) {
+      super(config, executorFactory, criticalErrorListener);
+   }
+
+   @Override
+   protected void init(Configuration config, IOCriticalErrorListener criticalErrorListener) {
+      DatabaseStorageConfiguration dbConf = (DatabaseStorageConfiguration) config.getStoreConfiguration();
+
+      Journal localBindings = new JDBCJournalImpl(dbConf.getJdbcConnectionUrl(), dbConf.getBindingsTableName());
+      bindingsJournal = localBindings;
+
+      Journal localMessage = new JDBCJournalImpl(dbConf.getJdbcConnectionUrl(), dbConf.getMessageTableName());
+      messageJournal = localMessage;
+   }
+
+   @Override
+   public synchronized void stop(boolean ioCriticalError) throws Exception {
+      if (!started) {
+         return;
+      }
+
+      if (!ioCriticalError) {
+         performCachedLargeMessageDeletes();
+         // Must call close to make sure last id is persisted
+         if (journalLoaded && idGenerator != null)
+            idGenerator.persistCurrentID();
+      }
+
+      final CountDownLatch latch = new CountDownLatch(1);
+      executor.execute(new Runnable() {
+         @Override
+         public void run() {
+            latch.countDown();
+         }
+      });
+
+      latch.await(30, TimeUnit.SECONDS);
+
+      beforeStop();
+
+      ((JDBCJournalImpl) bindingsJournal).stop(false);
+
+      messageJournal.stop();
+
+      singleThreadExecutor.shutdown();
+
+      journalLoaded = false;
+
+      started = false;
+   }
+
+}

http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/64f74acd/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java
index 0242b50..4aa470b 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalRecordIds.java
@@ -27,7 +27,6 @@ public final class JournalRecordIds {
 
    // grouping journal record type
 
-   // mtaylor Added to compile
    public static final byte GROUP_RECORD = 20;
 
    // BindingsImpl journal record type


Mime
View raw message