Skip to content
Permalink
Browse files
AMQCLI-4, AMQCLI-5 Adding MetadataExporter abstraction
this will allow a pluggable implementation to export metadata
  • Loading branch information
cshannon committed Feb 16, 2017
1 parent 76ca845 commit b8d33cddecd5fd9a5cb2eee8550c3aa2ba704a37
Show file tree
Hide file tree
Showing 5 changed files with 130 additions and 60 deletions.
@@ -25,13 +25,9 @@
import javax.xml.stream.XMLOutputFactory;
import javax.xml.stream.XMLStreamWriter;

import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
import org.apache.activemq.cli.artemis.schema.ArtemisJournalMarshaller;
import org.apache.activemq.cli.kahadb.exporter.artemis.ArtemisXmlMessageRecoveryListener;
import org.apache.activemq.cli.schema.QueueBindingType;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.command.SubscriptionInfo;
import org.apache.activemq.cli.kahadb.exporter.artemis.ArtemisXmlMetadataExporter;
import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,43 +63,20 @@ public static void exportKahaDbStore(final File kahaDbDir, final File artemisXml
try(OutputStream fos = new BufferedOutputStream(compress ? new GZIPOutputStream(
new FileOutputStream(artemisXml)) : new FileOutputStream(artemisXml))) {

XMLStreamWriter xmlWriter = XMLOutputFactory.newFactory().createXMLStreamWriter(fos);
ArtemisJournalMarshaller xmlMarshaller = new ArtemisJournalMarshaller(xmlWriter);
final XMLStreamWriter xmlWriter = XMLOutputFactory.newFactory().createXMLStreamWriter(fos);
final ArtemisJournalMarshaller xmlMarshaller = new ArtemisJournalMarshaller(xmlWriter);
final KahaDBExporter dbExporter = new KahaDBExporter(adapter,
new ArtemisXmlMetadataExporter(adapter.getStore(), xmlMarshaller),
new ArtemisXmlMessageRecoveryListener(adapter.getStore(), xmlMarshaller));

xmlMarshaller.appendJournalOpen();
xmlMarshaller.appendBindingsElement();

adapter.getStore().getDestinations().stream()
.forEach(dest -> {
try {
if (dest.isQueue()) {
xmlMarshaller.appendBinding(QueueBindingType.builder()
.withName(dest.getPhysicalName())
.withRoutingType(RoutingType.ANYCAST.toString())
.withAddress(dest.getPhysicalName()).build());
} else if (dest.isTopic()) {
for (SubscriptionInfo info :
adapter.getStore().createTopicMessageStore((ActiveMQTopic) dest).getAllSubscriptions()) {
xmlMarshaller.appendBinding(QueueBindingType.builder()
.withName(ActiveMQDestination.createQueueNameForDurableSubscription(
true, info.getClientId(), info.getSubcriptionName()))
.withRoutingType(RoutingType.MULTICAST.toString())
.withAddress(dest.getPhysicalName()).build());
}
}
} catch (Exception e) {
throw new IllegalStateException(e);
}
});

dbExporter.exportMetadata();
xmlMarshaller.appendEndElement();
xmlMarshaller.appendMessagesElement();

KahaDBExporter dbExporter = new KahaDBExporter(adapter,
new ArtemisXmlMessageRecoveryListener(adapter.getStore(), xmlMarshaller));

dbExporter.exportQueues();
dbExporter.exportTopics();
xmlMarshaller.appendEndElement();
xmlMarshaller.appendJournalClose(true);
} finally {
adapter.stop();
@@ -25,7 +25,6 @@
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.store.MessageRecoveryListener;
import org.apache.activemq.store.MessageStore;
import org.apache.activemq.store.TopicMessageStore;
import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
import org.apache.activemq.util.IOExceptionSupport;
import org.slf4j.Logger;
@@ -36,26 +35,44 @@ public class KahaDBExporter implements MessageStoreExporter {
static final Logger LOG = LoggerFactory.getLogger(KahaDBExporter.class);

private final KahaDBPersistenceAdapter adapter;
private final MessageStoreMetadataExporter metadataExporter;
private final MessageRecoveryListener recoveryListener;

public KahaDBExporter (final KahaDBPersistenceAdapter adapter,
public KahaDBExporter(final KahaDBPersistenceAdapter adapter,
final MessageStoreMetadataExporter metadataExporter,
final MessageRecoveryListener recoveryListener) {
this.adapter = adapter;
this.metadataExporter = metadataExporter;
this.recoveryListener = recoveryListener;
}


@Override
public void exportMetadata() throws IOException {
metadataExporter.export();
}

@Override
public void exportQueues() throws IOException {
exportDestinations(ActiveMQDestination.QUEUE_TYPE);
}

@Override
public void exportTopics() throws IOException {
exportDestinations(ActiveMQDestination.TOPIC_TYPE);
}

private void exportDestinations(byte destType) throws IOException {
final Set<ActiveMQDestination> destinations = adapter.getDestinations().stream().filter(
dest -> dest.isQueue()).collect(Collectors.toSet());
dest -> dest.getDestinationType() == destType).collect(Collectors.toSet());

// loop through all queues and export them
for (final ActiveMQDestination destination : destinations) {

LOG.info("Starting export of: " + destination);
final ActiveMQQueue queue = (ActiveMQQueue) destination;
final MessageStore messageStore = adapter.createQueueMessageStore(queue);
final MessageStore messageStore = destination.isQueue() ?
adapter.createQueueMessageStore((ActiveMQQueue) destination) :
adapter.createTopicMessageStore((ActiveMQTopic) destination);

try {
// migrate the data
@@ -66,24 +83,4 @@ public void exportQueues() throws IOException {
}
}

@Override
public void exportTopics() throws IOException {

final Set<ActiveMQDestination> destinations = adapter.getDestinations().stream().filter(
dest -> dest.isTopic()).collect(Collectors.toSet());

for (ActiveMQDestination destination : destinations) {
LOG.info("Starting export of: " + destination);

final ActiveMQTopic topic = (ActiveMQTopic) destination;
final TopicMessageStore messageStore = adapter.createTopicMessageStore(topic);

//recover topic
try {
messageStore.recover(recoveryListener);
} catch (Exception e) {
IOExceptionSupport.create(e);
}
}
}
}
@@ -20,6 +20,8 @@

public interface MessageStoreExporter {

public void exportMetadata() throws IOException;

public void exportQueues() throws IOException;

public void exportTopics() throws IOException;
@@ -0,0 +1,24 @@
/**
* 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.cli.kahadb.exporter;

import java.io.IOException;

public interface MessageStoreMetadataExporter {

public void export() throws IOException;
}
@@ -0,0 +1,74 @@
/**
* 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.cli.kahadb.exporter.artemis;

import java.io.IOException;

import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
import org.apache.activemq.cli.artemis.schema.ArtemisJournalMarshaller;
import org.apache.activemq.cli.kahadb.exporter.MessageStoreMetadataExporter;
import org.apache.activemq.cli.schema.QueueBindingType;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.command.SubscriptionInfo;
import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
import org.apache.activemq.store.kahadb.KahaDBStore;

public class ArtemisXmlMetadataExporter implements MessageStoreMetadataExporter {

private final KahaDBStore store;
private final ArtemisJournalMarshaller xmlMarshaller;


/**
* @param xmlMarshaller
*/
public ArtemisXmlMetadataExporter(final KahaDBStore store,
final ArtemisJournalMarshaller xmlMarshaller) {
super();
this.store = store;
this.xmlMarshaller = xmlMarshaller;
}

@Override
public void export() throws IOException {
store.getDestinations().stream()
.forEach(dest -> {
try {
if (dest.isQueue()) {
xmlMarshaller.appendBinding(QueueBindingType.builder()
.withName(dest.getPhysicalName())
.withRoutingType(RoutingType.ANYCAST.toString())
.withAddress(dest.getPhysicalName()).build());
} else if (dest.isTopic()) {
for (SubscriptionInfo info :
store.createTopicMessageStore((ActiveMQTopic) dest).getAllSubscriptions()) {
xmlMarshaller.appendBinding(QueueBindingType.builder()
.withName(ActiveMQDestination.createQueueNameForDurableSubscription(
true, info.getClientId(), info.getSubcriptionName()))
.withRoutingType(RoutingType.MULTICAST.toString())
.withAddress(dest.getPhysicalName()).build());
}
}
} catch (Exception e) {
throw new IllegalStateException(e);
}
});

}

}

0 comments on commit b8d33cd

Please sign in to comment.