content: /**
   * 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.store.kahadb;
  
  import java.io.File;
  import java.io.FileFilter;
  import java.io.IOException;
  import java.nio.charset.Charset;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.CopyOnWriteArrayList;
  
  import java.util.stream.Collectors;
  import javax.transaction.xa.Xid;
  
  import org.apache.activemq.broker.BrokerService;
  import org.apache.activemq.broker.BrokerServiceAware;
  import org.apache.activemq.broker.ConnectionContext;
  import org.apache.activemq.broker.Lockable;
  import org.apache.activemq.broker.LockableServiceSupport;
  import org.apache.activemq.broker.Locker;
  import org.apache.activemq.broker.scheduler.JobSchedulerStore;
  import org.apache.activemq.command.ActiveMQDestination;
  import org.apache.activemq.command.ActiveMQQueue;
  import org.apache.activemq.command.ActiveMQTopic;
  import org.apache.activemq.command.LocalTransactionId;
  import org.apache.activemq.command.ProducerId;
  import org.apache.activemq.command.TransactionId;
  import org.apache.activemq.command.XATransactionId;
  import org.apache.activemq.filter.AnyDestination;
  import org.apache.activemq.filter.DestinationMap;
  import org.apache.activemq.filter.DestinationMapEntry;
  import org.apache.activemq.store.MessageStore;
  import org.apache.activemq.store.NoLocalSubscriptionAware;
  import org.apache.activemq.store.PersistenceAdapter;
  import org.apache.activemq.store.SharedFileLocker;
  import org.apache.activemq.store.TopicMessageStore;
  import org.apache.activemq.store.TransactionIdTransformer;
  import org.apache.activemq.store.TransactionIdTransformerAware;
  import org.apache.activemq.store.TransactionStore;
  import org.apache.activemq.store.kahadb.scheduler.JobSchedulerStoreImpl;
  import org.apache.activemq.usage.SystemUsage;
  import org.apache.activemq.util.IOExceptionSupport;
  import org.apache.activemq.util.IOHelper;
  import org.apache.activemq.util.IntrospectionSupport;
  import org.apache.activemq.util.ServiceStopper;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import static org.apache.activemq.store.kahadb.MessageDatabase.DEFAULT_DIRECTORY;
  
  /**
   * An implementation of {@link org.apache.activemq.store.PersistenceAdapter}  that supports
   * distribution of destinations across multiple kahaDB persistence adapters
   *
   * @org.apache.xbean.XBean element="mKahaDB"
   */
  public class MultiKahaDBPersistenceAdapter extends LockableServiceSupport implements PersistenceAdapter,
      BrokerServiceAware, NoLocalSubscriptionAware {
  
      static final Logger LOG = LoggerFactory.getLogger(MultiKahaDBPersistenceAdapter.class);
  
      final static ActiveMQDestination matchAll = new AnyDestination(new ActiveMQDestination[]{new ActiveMQQueue(">"), new ActiveMQTopic(">")});
      final int LOCAL_FORMAT_ID_MAGIC = Integer.valueOf(System.getProperty("org.apache.activemq.store.kahadb.MultiKahaDBTransactionStore.localXaFormatId", "61616"));
  
      final class DelegateDestinationMap extends DestinationMap {
          @Override
          public void setEntries(List<DestinationMapEntry>  entries) {
              super.setEntries(entries);
          }
      };
      final DelegateDestinationMap destinationMap = new DelegateDestinationMap();
  
      List<PersistenceAdapter> adapters = new CopyOnWriteArrayList<>();
      private File directory = new File(IOHelper.getDefaultDataDirectory() + File.separator + "mKahaDB");
  
      MultiKahaDBTransactionStore transactionStore = new MultiKahaDBTransactionStore(this);
  
      // all local store transactions are XA, 2pc if more than one adapter involved
      TransactionIdTransformer transactionIdTransformer = new TransactionIdTransformer() {
          @Override
          public TransactionId transform(TransactionId txid) {
              if (txid == null) {
                  return null;
              }
              if (txid.isLocalTransaction()) {
                  final LocalTransactionId t = (LocalTransactionId) txid;
                  return new XATransactionId(new Xid() {
                      @Override
                      public int getFormatId() {
                          return LOCAL_FORMAT_ID_MAGIC;
                      }
  
                      @Override
                      public byte[] getGlobalTransactionId() {
                          return t.getConnectionId().getValue().getBytes(Charset.forName("utf-8"));
                      }
  
                      @Override
                      public byte[] getBranchQualifier() {
                          return Long.toString(t.getValue()).getBytes(Charset.forName("utf-8"));
                      }
                  });
              } else {
                  return txid;
              }
          }
      };
  
      /**
       * Sets the  FilteredKahaDBPersistenceAdapter entries
       *
       * @org.apache.xbean.ElementType class="org.apache.activemq.store.kahadb.FilteredKahaDBPersistenceAdapter"
       */
      @SuppressWarnings({ "rawtypes", "unchecked" })
      public void setFilteredPersistenceAdapters(List entries) {
          for (Object entry : entries) {
              FilteredKahaDBPersistenceAdapter filteredAdapter = (FilteredKahaDBPersistenceAdapter) entry;
              PersistenceAdapter adapter = filteredAdapter.getPersistenceAdapter();
              if (filteredAdapter.getDestination() == null) {
                  filteredAdapter.setDestination(matchAll);
              }
  
              if (filteredAdapter.isPerDestination()) {
                  configureDirectory(adapter, null);
                  // per destination adapters will be created on demand or during recovery
                  continue;
              } else {
                  configureDirectory(adapter, nameFromDestinationFilter(filteredAdapter.getDestination()));
              }
  
              configureAdapter(adapter);
              adapters.add(adapter);
          }
          destinationMap.setEntries(entries);
      }
  
      public static String nameFromDestinationFilter(ActiveMQDestination destination) {
          if (destination.getQualifiedName().length() > IOHelper.getMaxFileNameLength()) {
              LOG.warn("Destination name is longer than 'MaximumFileNameLength' system property, " +
                       "potential problem with recovery can result from name truncation.");
          }
  
          return IOHelper.toFileSystemSafeName(destination.getQualifiedName());
      }
  
      public boolean isLocalXid(TransactionId xid) {
          return xid instanceof XATransactionId &&
                  ((XATransactionId)xid).getFormatId() == LOCAL_FORMAT_ID_MAGIC;
      }
  
      @Override
      public void beginTransaction(ConnectionContext context) throws IOException {
          throw new IllegalStateException();
      }
  
      @Override
      public void checkpoint(final boolean cleanup) throws IOException {
          for (PersistenceAdapter persistenceAdapter : adapters) {
              persistenceAdapter.checkpoint(cleanup);
          }
      }
  
      @Override
      public void commitTransaction(ConnectionContext context) throws IOException {
          throw new IllegalStateException();
      }
  
      @Override
      public MessageStore createQueueMessageStore(ActiveMQQueue destination) throws IOException {
          PersistenceAdapter persistenceAdapter = getMatchingPersistenceAdapter(destination);
          return transactionStore.proxy(persistenceAdapter.createTransactionStore(), persistenceAdapter.createQueueMessageStore(destination));
      }
  
      private PersistenceAdapter getMatchingPersistenceAdapter(ActiveMQDestination destination) throws IOException {
          Object result = destinationMap.chooseValue(destination);
          if (result == null) {
              throw new RuntimeException("No matching persistence adapter configured for destination: " + destination + ", options:" + adapters);
          }
          FilteredKahaDBPersistenceAdapter filteredAdapter = (FilteredKahaDBPersistenceAdapter) result;
          if (filteredAdapter.getDestination() == matchAll && filteredAdapter.isPerDestination()) {
              filteredAdapter = addAdapter(filteredAdapter, destination);
              if (LOG.isTraceEnabled()) {
                  LOG.trace("created per destination adapter for: " + destination  + ", " + result);
              }
          }
          startAdapter(filteredAdapter.getPersistenceAdapter(), destination.getQualifiedName());
          LOG.debug("destination {} matched persistence adapter {}", destination.getQualifiedName(), filteredAdapter.getPersistenceAdapter());
          return filteredAdapter.getPersistenceAdapter();
      }
  
      private void startAdapter(PersistenceAdapter kahaDBPersistenceAdapter, String destination) {
          try {
              kahaDBPersistenceAdapter.start();
          } catch (Exception e) {
              RuntimeException detail = new RuntimeException("Failed to start per destination persistence adapter for destination: " + destination + ", options:" + adapters, e);
              LOG.error(detail.toString(), e);
              throw detail;
          }
      }
  
      private void stopAdapter(PersistenceAdapter kahaDBPersistenceAdapter, String destination) {
          try {
              kahaDBPersistenceAdapter.stop();
          } catch (Exception e) {
              RuntimeException detail = new RuntimeException("Failed to stop per destination persistence adapter for destination: " + destination + ", options:" + adapters, e);
              LOG.error(detail.toString(), e);
              throw detail;
          }
      }
  
      @Override
      public TopicMessageStore createTopicMessageStore(ActiveMQTopic destination) throws IOException {
          PersistenceAdapter persistenceAdapter = getMatchingPersistenceAdapter(destination);
          return transactionStore.proxy(persistenceAdapter.createTransactionStore(), persistenceAdapter.createTopicMessageStore(destination));
      }
  
      @Override
      public TransactionStore createTransactionStore() throws IOException {
          return transactionStore;
      }
  
      @Override
      public void deleteAllMessages() throws IOException {
          for (PersistenceAdapter persistenceAdapter : adapters) {
              persistenceAdapter.deleteAllMessages();
          }
          transactionStore.deleteAllMessages();
          IOHelper.deleteChildren(getDirectory());
          for (Object o : destinationMap.get(new AnyDestination(new ActiveMQDestination[]{new ActiveMQQueue(">"), new ActiveMQTopic(">")}))) {
              if (o instanceof FilteredKahaDBPersistenceAdapter) {
                  FilteredKahaDBPersistenceAdapter filteredKahaDBPersistenceAdapter = (FilteredKahaDBPersistenceAdapter) o;
                  if (filteredKahaDBPersistenceAdapter.getPersistenceAdapter().getDirectory() != DEFAULT_DIRECTORY) {
                      IOHelper.deleteChildren(filteredKahaDBPersistenceAdapter.getPersistenceAdapter().getDirectory());
                  }
                  if (filteredKahaDBPersistenceAdapter.getPersistenceAdapter() instanceof KahaDBPersistenceAdapter) {
                      KahaDBPersistenceAdapter kahaDBPersistenceAdapter = (KahaDBPersistenceAdapter) filteredKahaDBPersistenceAdapter.getPersistenceAdapter();
                      if (kahaDBPersistenceAdapter.getIndexDirectory() != null) {
                          IOHelper.deleteChildren(kahaDBPersistenceAdapter.getIndexDirectory());
                      }
                  }
              }
          }
      }
  
      @Override
      public Set<ActiveMQDestination> getDestinations() {
          Set<ActiveMQDestination> results = new HashSet<ActiveMQDestination>();
          for (PersistenceAdapter persistenceAdapter : adapters) {
              results.addAll(persistenceAdapter.getDestinations());
          }
          return results;
      }
  
      @Override
      public long getLastMessageBrokerSequenceId() throws IOException {
          long maxId = -1;
          for (PersistenceAdapter persistenceAdapter : adapters) {
              maxId = Math.max(maxId, persistenceAdapter.getLastMessageBrokerSequenceId());
          }
          return maxId;
      }
  
      @Override
      public long getLastProducerSequenceId(ProducerId id) throws IOException {
          long maxId = -1;
          for (PersistenceAdapter persistenceAdapter : adapters) {
              maxId = Math.max(maxId, persistenceAdapter.getLastProducerSequenceId(id));
          }
          return maxId;
      }
  
      @Override
      public void allowIOResumption() {
          for (PersistenceAdapter persistenceAdapter : adapters) {
              persistenceAdapter.allowIOResumption();
          }
      }
  
      @Override
      public void removeQueueMessageStore(ActiveMQQueue destination) {
          PersistenceAdapter adapter = null;
          try {
              adapter = getMatchingPersistenceAdapter(destination);
          } catch (IOException e) {
              throw new RuntimeException(e);
          }
          if (adapter instanceof PersistenceAdapter && adapter.getDestinations().isEmpty()) {
              adapter.removeQueueMessageStore(destination);
              removeMessageStore(adapter, destination);
              destinationMap.remove(destination, adapter);
          }
      }
  
      @Override
      public void removeTopicMessageStore(ActiveMQTopic destination) {
          PersistenceAdapter adapter = null;
          try {
              adapter = getMatchingPersistenceAdapter(destination);
          } catch (IOException e) {
              throw new RuntimeException(e);
          }
          if (adapter instanceof PersistenceAdapter && adapter.getDestinations().isEmpty()) {
              adapter.removeTopicMessageStore(destination);
              removeMessageStore(adapter, destination);
              destinationMap.remove(destination, adapter);
          }
      }
  
      private void removeMessageStore(PersistenceAdapter adapter, ActiveMQDestination destination) {
          stopAdapter(adapter, destination.toString());
          File adapterDir = adapter.getDirectory();
          if (adapterDir != null) {
              if (IOHelper.deleteFile(adapterDir)) {
                  if (LOG.isTraceEnabled()) {
                      LOG.trace("deleted per destination adapter directory for: " + destination);
                  }
              } else {
                  if (LOG.isTraceEnabled()) {
                      LOG.trace("failed to deleted per destination adapter directory for: " + destination);
                  }
              }
          }
      }
  
      @Override
      public void rollbackTransaction(ConnectionContext context) throws IOException {
          throw new IllegalStateException();
      }
  
      @Override
      public void setBrokerName(String brokerName) {
          for (PersistenceAdapter persistenceAdapter : adapters) {
              persistenceAdapter.setBrokerName(brokerName);
          }
      }
  
      @Override
      public void setUsageManager(SystemUsage usageManager) {
          for (PersistenceAdapter persistenceAdapter : adapters) {
              persistenceAdapter.setUsageManager(usageManager);
          }
      }
  
      @Override
      public long size() {
          long size = 0;
          for (PersistenceAdapter persistenceAdapter : adapters) {
              size += persistenceAdapter.size();
          }
          return size;
      }
  
      @Override
      public void doStart() throws Exception {
          Object result = destinationMap.chooseValue(matchAll);
          if (result != null) {
              FilteredKahaDBPersistenceAdapter filteredAdapter = (FilteredKahaDBPersistenceAdapter) result;
              if (filteredAdapter.getDestination() == matchAll && filteredAdapter.isPerDestination()) {
                  findAndRegisterExistingAdapters(filteredAdapter);
              }
          }
          for (PersistenceAdapter persistenceAdapter : adapters) {
              persistenceAdapter.start();
          }
      }
  
      private void findAndRegisterExistingAdapters(FilteredKahaDBPersistenceAdapter template) throws IOException {
          FileFilter destinationNames = file ->
              file.getName().startsWith("queue#") || file.getName().startsWith("topic#");
  
          File[] candidates = template.getPersistenceAdapter().getDirectory().listFiles(destinationNames);
          if (candidates != null) {
              Set<File> existing = adapters.stream().map(PersistenceAdapter::getDirectory).collect(
                  Collectors.toSet());
              for (File candidate : candidates) {
                  if(!existing.contains(candidate)) {
                      LOG.debug("Adapter does not exist for dir: {} so will register it", candidate);
                      registerExistingAdapter(template, candidate);
                  }
              }
          }
      }
  
      private void registerExistingAdapter(FilteredKahaDBPersistenceAdapter filteredAdapter, File candidate) throws IOException {
          PersistenceAdapter adapter = adapterFromTemplate(filteredAdapter, candidate.getName());
          startAdapter(adapter, candidate.getName());
          Set<ActiveMQDestination> destinations = adapter.getDestinations();
          if (destinations.size() != 0) {
              registerAdapter(filteredAdapter, adapter, destinations.toArray(new ActiveMQDestination[]{})[0]);
          } else {
              stopAdapter(adapter, candidate.getName());
          }
      }
  
      private FilteredKahaDBPersistenceAdapter addAdapter(FilteredKahaDBPersistenceAdapter filteredAdapter, ActiveMQDestination destination) throws IOException {
          PersistenceAdapter adapter = adapterFromTemplate(filteredAdapter, nameFromDestinationFilter(destination));
          return registerAdapter(filteredAdapter, adapter, destination);
      }
  
      private PersistenceAdapter adapterFromTemplate(FilteredKahaDBPersistenceAdapter template, String destinationName) throws IOException {
          PersistenceAdapter adapter = kahaDBFromTemplate(template.getPersistenceAdapter());
          configureAdapter(adapter);
          configureDirectory(adapter, destinationName);
          configureIndexDirectory(adapter, template.getPersistenceAdapter(), destinationName);
          return adapter;
      }
  
      private void configureIndexDirectory(PersistenceAdapter adapter, PersistenceAdapter template, String destinationName) {
          if (template instanceof KahaDBPersistenceAdapter) {
              KahaDBPersistenceAdapter kahaDBPersistenceAdapter = (KahaDBPersistenceAdapter) template;
              if (kahaDBPersistenceAdapter.getIndexDirectory() != null) {
                  if (adapter instanceof KahaDBPersistenceAdapter) {
                      File directory = kahaDBPersistenceAdapter.getIndexDirectory();
                      if (destinationName != null) {
                          directory = new File(directory, destinationName);
                      }
                      ((KahaDBPersistenceAdapter)adapter).setIndexDirectory(directory);
                  }
              }
          }
      }
  
      private void configureDirectory(PersistenceAdapter adapter, String fileName) {
          File directory = null;
          File defaultDir = DEFAULT_DIRECTORY;
          try {
              defaultDir = adapter.getClass().getConstructor().newInstance().getDirectory();
          } catch (Exception e) {
          }
          if (defaultDir.equals(adapter.getDirectory())) {
              // not set so inherit from mkahadb
              directory = getDirectory();
          } else {
              directory = adapter.getDirectory();
          }
  
          if (fileName != null) {
              directory = new File(directory, fileName);
          }
          adapter.setDirectory(directory);
      }
  
      private FilteredKahaDBPersistenceAdapter registerAdapter(FilteredKahaDBPersistenceAdapter template, PersistenceAdapter adapter, ActiveMQDestination destination) {
          adapters.add(adapter);
          FilteredKahaDBPersistenceAdapter result = new FilteredKahaDBPersistenceAdapter(template, destination, adapter);
          destinationMap.put(destination, result);
          return result;
      }
  
      private void configureAdapter(PersistenceAdapter adapter) {
          // need a per store factory that will put the store in the branch qualifier to disiambiguate xid mbeans
          ((TransactionIdTransformerAware)adapter).setTransactionIdTransformer(transactionIdTransformer);
          if (isUseLock()) {
              if( adapter instanceof Lockable ) {
                  ((Lockable)adapter).setUseLock(false);
              }
          }
          if( adapter instanceof BrokerServiceAware ) {
              ((BrokerServiceAware)adapter).setBrokerService(getBrokerService());
          }
      }
  
      private PersistenceAdapter kahaDBFromTemplate(PersistenceAdapter template) throws IOException {
          try {
              Map<String, Object> configuration = new HashMap<String, Object>();
              IntrospectionSupport.getProperties(template, configuration, null);
              PersistenceAdapter adapter = template.getClass().getConstructor().newInstance();
              IntrospectionSupport.setProperties(adapter, configuration);
              return adapter;
          } catch (Exception e) {
              throw IOExceptionSupport.create(e);
          }
      }
  
      @Override
      protected void doStop(ServiceStopper stopper) throws Exception {
          for (PersistenceAdapter persistenceAdapter : adapters) {
              stopper.stop(persistenceAdapter);
          }
      }
  
      @Override
      public File getDirectory() {
          return this.directory;
      }
  
      @Override
      public void setDirectory(File directory) {
          this.directory = directory;
      }
  
      @Override
      public void init() throws Exception {
      }
  
      @Override
      public void setBrokerService(BrokerService brokerService) {
          super.setBrokerService(brokerService);
          for (PersistenceAdapter persistenceAdapter : adapters) {
              if( persistenceAdapter instanceof BrokerServiceAware ) {
                  ((BrokerServiceAware)persistenceAdapter).setBrokerService(getBrokerService());
              }
          }
      }
  
      public void setTransactionStore(MultiKahaDBTransactionStore transactionStore) {
          this.transactionStore = transactionStore;
      }
  
      /**
       * Set the max file length of the transaction journal
       * When set using Xbean, values of the form "20 Mb", "1024kb", and "1g" can
       * be used
       *
       * @org.apache.xbean.Property propertyEditor="org.apache.activemq.util.MemoryIntPropertyEditor"
       */
      public void setJournalMaxFileLength(int maxFileLength) {
          transactionStore.setJournalMaxFileLength(maxFileLength);
      }
  
      public int getJournalMaxFileLength() {
          return transactionStore.getJournalMaxFileLength();
      }
  
      /**
       * Set the max write batch size of  the transaction journal
       * When set using Xbean, values of the form "20 Mb", "1024kb", and "1g" can
       * be used
       *
       * @org.apache.xbean.Property propertyEditor="org.apache.activemq.util.MemoryIntPropertyEditor"
       */
      public void setJournalWriteBatchSize(int journalWriteBatchSize) {
          transactionStore.setJournalMaxWriteBatchSize(journalWriteBatchSize);
      }
  
      public int getJournalWriteBatchSize() {
          return transactionStore.getJournalMaxWriteBatchSize();
      }
  
  
      public void setJournalCleanupInterval(long journalCleanupInterval) {
          transactionStore.setJournalCleanupInterval(journalCleanupInterval);
      }
  
      public long getJournalCleanupInterval() {
          return transactionStore.getJournalCleanupInterval();
      }
  
      public void setCheckForCorruption(boolean checkForCorruption) {
          transactionStore.setCheckForCorruption(checkForCorruption);
      }
  
      public boolean isCheckForCorruption() {
          return transactionStore.isCheckForCorruption();
      }
  
      public List<PersistenceAdapter> getAdapters() {
          return Collections.unmodifiableList(adapters);
      }
  
      @Override
      public String toString() {
          String path = getDirectory() != null ? getDirectory().getAbsolutePath() : "DIRECTORY_NOT_SET";
          return "MultiKahaDBPersistenceAdapter[" + path + "]" + adapters;
      }
  
      @Override
      public Locker createDefaultLocker() throws IOException {
          SharedFileLocker locker = new SharedFileLocker();
          locker.configure(this);
          return locker;
      }
  
      @Override
      public JobSchedulerStore createJobSchedulerStore() throws IOException, UnsupportedOperationException {
          return new JobSchedulerStoreImpl();
      }
  
      /* (non-Javadoc)
       * @see org.apache.activemq.store.NoLocalSubscriptionAware#isPersistNoLocal()
       */
      @Override
      public boolean isPersistNoLocal() {
          // Prior to v11 the broker did not store the noLocal value for durable subs.
          return brokerService.getStoreOpenWireVersion() >= 11;
      }
  }
  