import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
@Test
public void testExportQueuesPatternEmpty() throws Exception {
    File sourceDir = storeFolder.newFolder();
    ActiveMQQueue queue = new ActiveMQQueue("test.queue");
    PersistenceAdapter adapter = getPersistenceAdapter(sourceDir);
    adapter.start();
    MessageStore messageStore = adapter.createQueueMessageStore(queue);
    messageStore.start();
    publishQueueMessages(messageStore, queue, new Date(), new byte[] {10, 11, 12});
    adapter.stop();

    File xmlFile = new File(storeFolder.getRoot().getAbsoluteFile(), "outputXml.xml");
    exportStore(ExportConfigurationBuilder.newBuilder()
            .setSource(sourceDir)
            .setTarget(xmlFile)
            .setQueuePattern("empty.>"));

    validate(xmlFile, 0);
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
@Override
public PersistenceAdapter createPersistenceAdapter() throws IOException {
    LevelDBStore store = new LevelDBStore();
    store.setVerifyChecksums(verifyChecksums);
    store.setAsyncBufferSize(asyncBufferSize);
    store.setDirectory(directory);
    store.setFlushDelay(flushDelay);
    store.setIndexBlockRestartInterval(indexBlockRestartInterval);
    store.setIndexBlockSize(indexBlockSize);
    store.setIndexCacheSize(indexCacheSize);
    store.setIndexCompression(indexCompression);
    store.setIndexFactory(indexFactory);
    store.setIndexMaxOpenFiles(indexMaxOpenFiles);
    store.setIndexWriteBufferSize(indexWriteBufferSize);
    store.setLogCompression(logCompression);
    store.setLogDirectory(logDirectory);
    store.setLogSize(logSize);
    store.setMonitorStats(monitorStats);
    store.setParanoidChecks(paranoidChecks);
    store.setSync(sync);
    return store;
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
protected PersistenceAdapter createPersistenceAdapter() throws IOException {
    if (isPersistent()) {
        PersistenceAdapterFactory fac = getPersistenceFactory();
        if (fac != null) {
            return fac.createPersistenceAdapter();
        } else {
            try {
                String clazz = "org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter";
                PersistenceAdapter adaptor = (PersistenceAdapter)getClass().getClassLoader().loadClass(clazz).newInstance();
                File dir = new File(getBrokerDataDirectory(),"KahaDB");
                adaptor.setDirectory(dir);
                return adaptor;
            } catch (Throwable e) {
                throw IOExceptionSupport.create(e);
            }
        }
    } else {
        return new MemoryPersistenceAdapter();
    }
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
private void createLocalBroker() throws Exception {
   localBroker = new BrokerService();
   localBroker.setBrokerName("LOCAL");
   localBroker.setUseJmx(true);
   localBroker.setSchedulePeriodForDestinationPurge(5000);
   ManagementContext managementContext = new ManagementContext();
   managementContext.setCreateConnector(false);
   localBroker.setManagementContext(managementContext);
   PersistenceAdapter persistenceAdapter = persistenceAdapterFactory("target/local");
   localBroker.setPersistenceAdapter(persistenceAdapter);
   List<TransportConnector> transportConnectors = new ArrayList<>();
   DebugTransportFactory tf = new DebugTransportFactory();
   TransportServer transport = tf.doBind(URI.create("nio://127.0.0.1:23539"));
   TransportConnector transportConnector = new TransportConnector(transport);
   transportConnector.setName("tc");
   transportConnector.setAuditNetworkProducers(true);
   transportConnectors.add(transportConnector);
   localBroker.setTransportConnectors(transportConnectors);
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
private void createRemoteBroker() throws Exception {
   remoteBroker = new BrokerService();
   remoteBroker.setBrokerName("REMOTE");
   remoteBroker.setUseJmx(true);
   remoteBroker.setSchedulePeriodForDestinationPurge(5000);
   ManagementContext managementContext = new ManagementContext();
   managementContext.setCreateConnector(false);
   remoteBroker.setManagementContext(managementContext);
   PersistenceAdapter persistenceAdapter = persistenceAdapterFactory("target/remote");
   remoteBroker.setPersistenceAdapter(persistenceAdapter);
   List<NetworkConnector> networkConnectors = new ArrayList<>();
   DiscoveryNetworkConnector networkConnector = new DiscoveryNetworkConnector();
   networkConnector.setName("to local");
   // set maxInactivityDuration to 0, otherwise the broker restarts while you are in the debugger
   networkConnector.setUri(URI.create("static://(tcp://127.0.0.1:23539?wireFormat.maxInactivityDuration=0)"));
   networkConnector.setDuplex(true);
   //networkConnector.setNetworkTTL(5);
   //networkConnector.setDynamicOnly(true);
   networkConnector.setAlwaysSyncSend(true);
   networkConnector.setDecreaseNetworkConsumerPriority(false);
   networkConnector.setPrefetchSize(1);
   networkConnector.setCheckDuplicateMessagesOnDuplex(true);
   networkConnectors.add(networkConnector);
   remoteBroker.setNetworkConnectors(networkConnectors);
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
@Test
public void testJournalConfig() throws Exception {
   File journalFile = new File(JOURNAL_ROOT + "testJournalConfig/journal");
   recursiveDelete(journalFile);

   BrokerService broker;
   broker = createBroker(new FileSystemResource(CONF_ROOT + "journal-example.xml"));
   try {
      assertEquals("Broker Config Error (brokerName)", "brokerJournalConfigTest", broker.getBrokerName());

      PersistenceAdapter adapter = broker.getPersistenceAdapter();

      assertTrue("Should have created a journal persistence adapter", adapter instanceof JournalPersistenceAdapter);
      assertTrue("Should have created a journal directory at " + journalFile.getAbsolutePath(), journalFile.exists());

      LOG.info("Success");
   } finally {
      if (broker != null) {
         broker.stop();
      }
   }
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
@Override
protected PersistenceAdapter createPersistenceAdapter() throws IOException {
   JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
   EmbeddedDataSource dataSource = new EmbeddedDataSource();
   dataSource.setDatabaseName("derbyDb");
   dataSource.setCreateDatabase("create");
   jdbc.setDataSource(dataSource);
   jdbc.setCleanupPeriod(1000); // set up small cleanup period
   return jdbc;
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
private static PersistenceAdapter createPersistenceAdapter(final String clazz, final String prefix, final Map<String, String> params) throws IllegalAccessException, InvocationTargetException, ClassNotFoundException, InstantiationException {
    Class<?> aClass = Thread.currentThread().getContextClassLoader().loadClass(clazz);
    final PersistenceAdapter persistenceAdapter = PersistenceAdapter.class.cast(aClass.newInstance());
    while (aClass != null) {
        for (final Method m : aClass.getDeclaredMethods()) {
            if (m.getName().startsWith("set") && m.getParameterTypes().length == 1 && Modifier.isPublic(m.getModifiers())) {
                final String key = prefix + "." + m.getName().substring(3).toLowerCase(Locale.ENGLISH);
                final Object field = params.remove(key);
                if (field != null) {
                    try {
                        final Object toSet = PropertyEditors.getValue(m.getParameterTypes()[0], field.toString());
                        m.invoke(persistenceAdapter, toSet);
                    } catch (final PropertyEditorException cantConvertException) {
                        throw new IllegalArgumentException("can't convert " + field + " for " + m.getName(), cantConvertException);
                    }
                }
            }
        }
        aClass = aClass.getSuperclass();
    }
    return persistenceAdapter;
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
@Override
public PersistenceAdapter getPersistenceAdapter(File dir) {
    MultiKahaDBPersistenceAdapter adapter = new MultiKahaDBPersistenceAdapter();
    adapter.setJournalMaxFileLength(1024 * 1024);
    adapter.setDirectory(dir);

    KahaDBPersistenceAdapter kahaStore = new KahaDBPersistenceAdapter();
    kahaStore.setDirectory(dir);
    FilteredKahaDBPersistenceAdapter filtered = new FilteredKahaDBPersistenceAdapter();
    filtered.setPersistenceAdapter(kahaStore);
    filtered.setPerDestination(true);

    adapter.setFilteredPersistenceAdapters(Lists.newArrayList(filtered));
    return adapter;
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
@Override
public PersistenceAdapter getPersistenceAdapter(File dir) {
    KahaDBPersistenceAdapter adapter = new KahaDBPersistenceAdapter();
    adapter.setJournalMaxFileLength(1024 * 1024);
    adapter.setDirectory(dir);
    return adapter;
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
@Test
public void testExportTopicsPatternEmpty() throws Exception {
    File kahaDbDir = storeFolder.newFolder();

    ActiveMQTopic topic = new ActiveMQTopic("test.topic");
    PersistenceAdapter adapter = getPersistenceAdapter(kahaDbDir);
    adapter.start();
    TopicMessageStore messageStore = adapter.createTopicMessageStore(topic);
    messageStore.start();

    SubscriptionInfo sub1 = new SubscriptionInfo("clientId1", "sub1");
    sub1.setDestination(topic);
    messageStore.addSubscription(sub1, false);

    IdGenerator id = new IdGenerator();
    ConnectionContext context = new ConnectionContext();
    for (int i = 0; i < 5; i++) {
        ActiveMQTextMessage message = new ActiveMQTextMessage();
        message.setText("Test");
        message.setProperty("MyStringProperty", "abc");
        message.setProperty("MyIntegerProperty", 1);
        message.setDestination(topic);
        message.setMessageId(new MessageId(id.generateId() + ":1", i));
        messageStore.addMessage(context, message);
    }

    adapter.stop();

    //should be empty as no messages match empty.>
    File xmlFile = new File(storeFolder.getRoot().getAbsoluteFile(), "outputXml.xml");
    exportStore(ExportConfigurationBuilder.newBuilder()
            .setTopicPattern("empty.>")
            .setSource(kahaDbDir)
            .setTarget(xmlFile));
    validate(xmlFile, 0);
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
public SystemUsage(String name, PersistenceAdapter adapter, PListStore tempStore, JobSchedulerStore jobSchedulerStore) {
    this.parent = null;
    this.name = name;
    this.memoryUsage = new MemoryUsage(name + ":memory");
    this.storeUsage = new StoreUsage(name + ":store", adapter);
    this.tempUsage = new TempUsage(name + ":temp", tempStore);
    this.jobSchedulerUsage = new JobSchedulerUsage(name + ":jobScheduler", jobSchedulerStore);
    this.memoryUsage.setExecutor(getExecutor());
    this.storeUsage.setExecutor(getExecutor());
    this.tempUsage.setExecutor(getExecutor());
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
protected List<Message> getSubscriberMessages(SubscriptionView view) {
    // TODO It is very dangerous operation for big backlogs
    if (!(destinationFactory instanceof DestinationFactoryImpl)) {
        throw new RuntimeException("unsupported by " + destinationFactory);
    }
    PersistenceAdapter adapter = ((DestinationFactoryImpl)destinationFactory).getPersistenceAdapter();
    final List<Message> result = new ArrayList<Message>();
    try {
        ActiveMQTopic topic = new ActiveMQTopic(view.getDestinationName());
        TopicMessageStore store = adapter.createTopicMessageStore(topic);
        store.recover(new MessageRecoveryListener() {
            @Override
            public boolean recoverMessage(Message message) throws Exception {
                result.add(message);
                return true;
            }

            @Override
            public boolean recoverMessageReference(MessageId messageReference) throws Exception {
                throw new RuntimeException("Should not be called.");
            }

            @Override
            public boolean hasSpace() {
                return true;
            }

            @Override
            public boolean isDuplicate(MessageId id) {
                return false;
            }
        });
    } catch (Throwable e) {
        LOG.error("Failed to browse messages for Subscription {}", view, e);
    }
    return result;

}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
public DestinationFactoryImpl(BrokerService brokerService, TaskRunnerFactory taskRunnerFactory, PersistenceAdapter persistenceAdapter) {
    this.brokerService = brokerService;
    this.taskRunnerFactory = taskRunnerFactory;
    if (persistenceAdapter == null) {
        throw new IllegalArgumentException("null persistenceAdapter");
    }
    this.persistenceAdapter = persistenceAdapter;
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
public PersistenceAdapter getPersistenceAdapter() throws IOException {
    if (persistenceAdapter == null) {
        persistenceAdapter = createPersistenceAdapter();
        configureService(persistenceAdapter);
        this.persistenceAdapter = registerPersistenceAdapterMBean(persistenceAdapter);
    }
    return persistenceAdapter;
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
/**
 * Sets the persistence adaptor implementation to use for this broker
 *
 * @throws IOException
 */
public void setPersistenceAdapter(PersistenceAdapter persistenceAdapter) throws IOException {
    if (!isPersistent() && ! (persistenceAdapter instanceof MemoryPersistenceAdapter)) {
        LOG.warn("persistent=\"false\", ignoring configured persistenceAdapter: {}", persistenceAdapter);
        return;
    }
    this.persistenceAdapter = persistenceAdapter;
    configureService(this.persistenceAdapter);
    this.persistenceAdapter = registerPersistenceAdapterMBean(persistenceAdapter);
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
@Override
public void setLocker(Locker locker) throws IOException {
    this.locker = locker;
    locker.setLockable(this);
    if (this instanceof PersistenceAdapter) {
        this.locker.configure((PersistenceAdapter)this);
    }
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
public void setBrokerService(BrokerService brokerService) {
    this.brokerService = brokerService;
    PersistenceAdapter pa = getLongTermPersistence();
    if( pa instanceof BrokerServiceAware ) {
        ((BrokerServiceAware)pa).setBrokerService(brokerService);
    }
}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
public PersistenceAdapter createPersistenceAdapter() throws IOException {
    jdbcPersistenceAdapter.setDataSource(getDataSource());

    if (!useJournal) {
        return jdbcPersistenceAdapter;
    }
    JournalPersistenceAdapter result =  new JournalPersistenceAdapter(getJournal(), jdbcPersistenceAdapter, getTaskRunnerFactory());
    result.setDirectory(getDataDirectoryFile());
    return result;

}
 

import org.apache.activemq.store.PersistenceAdapter; //导入依赖的package包/类
public SystemUsage(String name,
                   PersistenceAdapter adapter,
                   PListStore tempStore,
                   JobSchedulerStore jobSchedulerStore) {
   this.parent = null;
   this.name = name;
   this.memoryUsage = new MemoryUsage(name + ":memory");
   this.storeUsage = new StoreUsage(name + ":store", adapter);
   this.tempUsage = new TempUsage(name + ":temp", tempStore);
   this.jobSchedulerUsage = new JobSchedulerUsage(name + ":jobScheduler", jobSchedulerStore);
   this.memoryUsage.setExecutor(getExecutor());
   this.storeUsage.setExecutor(getExecutor());
   this.tempUsage.setExecutor(getExecutor());
}