| /* |
| * 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 static org.junit.Assert.assertArrayEquals; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertNull; |
| |
| import java.io.BufferedReader; |
| import java.io.File; |
| import java.io.FileReader; |
| import java.io.IOException; |
| import java.util.Date; |
| import java.util.HashMap; |
| import java.util.Map; |
| |
| import javax.jms.BytesMessage; |
| import javax.jms.Connection; |
| import javax.jms.MapMessage; |
| import javax.jms.MessageConsumer; |
| import javax.jms.ObjectMessage; |
| import javax.jms.Session; |
| import javax.jms.StreamMessage; |
| import javax.jms.TextMessage; |
| import javax.xml.bind.JAXBContext; |
| import javax.xml.bind.JAXBElement; |
| import javax.xml.bind.JAXBException; |
| import javax.xml.bind.Unmarshaller; |
| |
| import org.apache.activemq.artemis.api.core.TransportConfiguration; |
| import org.apache.activemq.artemis.cli.commands.tools.XmlDataImporter; |
| import org.apache.activemq.artemis.core.config.Configuration; |
| import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl; |
| import org.apache.activemq.artemis.core.remoting.impl.netty.NettyAcceptorFactory; |
| import org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory; |
| import org.apache.activemq.artemis.core.server.ActiveMQServer; |
| import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl; |
| import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; |
| import org.apache.activemq.broker.ConnectionContext; |
| import org.apache.activemq.cli.schema.ActivemqJournalType; |
| import org.apache.activemq.cli.schema.ObjectFactory; |
| import org.apache.activemq.command.ActiveMQBytesMessage; |
| import org.apache.activemq.command.ActiveMQMapMessage; |
| import org.apache.activemq.command.ActiveMQObjectMessage; |
| import org.apache.activemq.command.ActiveMQQueue; |
| import org.apache.activemq.command.ActiveMQStreamMessage; |
| import org.apache.activemq.command.ActiveMQTextMessage; |
| import org.apache.activemq.command.ActiveMQTopic; |
| import org.apache.activemq.command.MessageAck; |
| import org.apache.activemq.command.MessageId; |
| import org.apache.activemq.command.SubscriptionInfo; |
| import org.apache.activemq.store.MessageStore; |
| import org.apache.activemq.store.TopicMessageStore; |
| import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter; |
| import org.apache.activemq.util.ByteSequence; |
| import org.apache.activemq.util.IdGenerator; |
| import org.junit.Rule; |
| import org.junit.Test; |
| import org.junit.rules.TemporaryFolder; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| public class ExporterTest { |
| |
| static final Logger LOG = LoggerFactory.getLogger(ExporterTest.class); |
| |
| @Rule |
| public TemporaryFolder storeFolder = new TemporaryFolder(); |
| |
| /** |
| * TODO Improve test when real exporting is done, for now this just |
| * tests that the recovery listener iterates over all the queue messages |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testExportQueues() throws Exception { |
| |
| File kahaDbDir = storeFolder.newFolder(); |
| ActiveMQQueue queue = new ActiveMQQueue("test.queue"); |
| KahaDBPersistenceAdapter adapter = new KahaDBPersistenceAdapter(); |
| adapter.setJournalMaxFileLength(1024 * 1024); |
| adapter.setDirectory(kahaDbDir); |
| adapter.start(); |
| MessageStore messageStore = adapter.createQueueMessageStore(queue); |
| messageStore.start(); |
| |
| 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(queue); |
| message.setMessageId(new MessageId(id.generateId() + ":1", i)); |
| messageStore.addMessage(context, message); |
| } |
| byte[] bytes = new byte[] {10, 11, 12}; |
| for (int i = 0; i < 3; i++) { |
| ActiveMQBytesMessage message = new ActiveMQBytesMessage(); |
| |
| message.setContent(new ByteSequence(bytes)); |
| message.setProperty("MyStringProperty", "abc"); |
| message.setProperty("MyByteProperty", (byte)10); |
| message.setDestination(queue); |
| message.setMessageId(new MessageId(id.generateId() + ":2", i)); |
| messageStore.addMessage(context, message); |
| } |
| |
| for (int i = 0; i < 3; i++) { |
| ActiveMQMapMessage message = new ActiveMQMapMessage(); |
| message.setObject("key", "value"); |
| message.setObject("key2", 10); |
| message.setProperty("MyStringProperty", "abc"); |
| message.setDestination(queue); |
| message.setMessageId(new MessageId(id.generateId() + ":3", i)); |
| messageStore.addMessage(context, message); |
| } |
| |
| Date date = new Date(); |
| for (int i = 0; i < 3; i++) { |
| ActiveMQObjectMessage message = new ActiveMQObjectMessage(); |
| message.setObject(date); |
| message.setDestination(queue); |
| message.setMessageId(new MessageId(id.generateId() + ":4", i)); |
| messageStore.addMessage(context, message); |
| } |
| |
| for (int i = 0; i < 3; i++) { |
| ActiveMQStreamMessage message = new ActiveMQStreamMessage(); |
| message.writeByte((byte)10); |
| message.storeContentAndClear(); |
| message.setDestination(queue); |
| message.setMessageId(new MessageId(id.generateId() + ":5", i)); |
| messageStore.addMessage(context, message); |
| } |
| |
| adapter.stop(); |
| |
| File xmlFile = new File(storeFolder.getRoot().getAbsoluteFile(), "outputXml.xml"); |
| Exporter.exportKahaDbStore(kahaDbDir, xmlFile); |
| |
| // printFile(xmlFile); |
| |
| validate(xmlFile, 17); |
| |
| final ActiveMQServer artemisServer = buildArtemisBroker(); |
| artemisServer.start(); |
| |
| XmlDataImporter dataImporter = new XmlDataImporter(); |
| dataImporter.process(xmlFile.getAbsolutePath(), "localhost", 61400, false); |
| |
| ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61400"); |
| |
| Connection connection = null; |
| try { |
| |
| connection = cf.createConnection(); |
| connection.start(); |
| |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| MessageConsumer messageConsumer = session.createConsumer(session.createQueue("test.queue")); |
| |
| for (int i = 0; i < 5; i++) { |
| TextMessage messageReceived = (TextMessage) messageConsumer.receive(1000); |
| assertNotNull(messageReceived); |
| assertEquals("abc", messageReceived.getStringProperty("MyStringProperty")); |
| assertEquals("Test", messageReceived.getText()); |
| } |
| |
| for (int i = 0; i < 3; i++) { |
| BytesMessage messageReceived = (BytesMessage) messageConsumer.receive(1000); |
| assertNotNull(messageReceived); |
| assertEquals("abc", messageReceived.getStringProperty("MyStringProperty")); |
| assertEquals((byte)10, messageReceived.getByteProperty("MyByteProperty")); |
| byte[] result = new byte[3]; |
| messageReceived.readBytes(result); |
| assertArrayEquals(bytes, result); |
| } |
| |
| for (int i = 0; i < 3; i++) { |
| MapMessage messageReceived = (MapMessage) messageConsumer.receive(1000); |
| assertNotNull(messageReceived); |
| assertEquals("abc", messageReceived.getStringProperty("MyStringProperty")); |
| assertEquals("value", messageReceived.getObject("key")); |
| } |
| |
| for (int i = 0; i < 3; i++) { |
| ObjectMessage messageReceived = (ObjectMessage) messageConsumer.receive(1000); |
| assertNotNull(messageReceived); |
| assertEquals(date, messageReceived.getObject()); |
| } |
| |
| for (int i = 0; i < 3; i++) { |
| StreamMessage messageReceived = (StreamMessage) messageConsumer.receive(1000); |
| assertNotNull(messageReceived); |
| assertEquals((byte)10, messageReceived.readByte()); |
| } |
| |
| } finally { |
| if (connection != null) { |
| connection.close(); |
| } |
| cf.close(); |
| } |
| |
| artemisServer.stop(); |
| } |
| |
| @Test |
| public void testExportTopics() throws Exception { |
| |
| File kahaDbDir = storeFolder.newFolder(); |
| |
| ActiveMQTopic topic = new ActiveMQTopic("test.topic"); |
| KahaDBPersistenceAdapter adapter = new KahaDBPersistenceAdapter(); |
| adapter.setJournalMaxFileLength(1024 * 1024); |
| adapter.setDirectory(kahaDbDir); |
| adapter.start(); |
| TopicMessageStore messageStore = adapter.createTopicMessageStore(topic); |
| messageStore.start(); |
| |
| SubscriptionInfo sub1 = new SubscriptionInfo("clientId1", "sub1"); |
| SubscriptionInfo sub2 = new SubscriptionInfo("clientId1", "sub2"); |
| sub1.setDestination(topic); |
| messageStore.addSubscription(sub1, false); |
| messageStore.addSubscription(sub2, false); |
| |
| IdGenerator id = new IdGenerator(); |
| ConnectionContext context = new ConnectionContext(); |
| MessageId first = null; |
| 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); |
| if (i == 0) { |
| first = message.getMessageId(); |
| } |
| } |
| |
| //ack for sub1 only |
| messageStore.acknowledge(context, "clientId1", "sub1", first, new MessageAck()); |
| |
| adapter.stop(); |
| |
| File xmlFile = new File(storeFolder.getRoot().getAbsoluteFile(), "outputXml.xml"); |
| Exporter.exportKahaDbStore(kahaDbDir, xmlFile); |
| |
| // printFile(xmlFile); |
| |
| validate(xmlFile, 5); |
| |
| final ActiveMQServer artemisServer = buildArtemisBroker(); |
| artemisServer.start(); |
| |
| XmlDataImporter dataImporter = new XmlDataImporter(); |
| dataImporter.process(xmlFile.getAbsolutePath(), "localhost", 61400, false); |
| |
| ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61400"); |
| |
| Connection connection = null; |
| try { |
| |
| connection = cf.createConnection(); |
| connection.setClientID("clientId1"); |
| connection.start(); |
| |
| Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); |
| MessageConsumer messageConsumer = session.createSharedDurableConsumer( |
| session.createTopic("test.topic"), "sub1"); |
| MessageConsumer messageConsumer2 = session.createSharedDurableConsumer( |
| session.createTopic("test.topic"), "sub2"); |
| |
| for (int i = 0; i < 5; i++) { |
| TextMessage messageReceived1 = (TextMessage) messageConsumer.receive(1000); |
| if (i < 4) { |
| assertNotNull(messageReceived1); |
| } else { |
| assertNull(messageReceived1); |
| } |
| TextMessage messageReceived2 = (TextMessage) messageConsumer2.receive(1000); |
| assertNotNull(messageReceived2); |
| |
| assertEquals("abc", messageReceived2.getStringProperty("MyStringProperty")); |
| assertEquals("Test", messageReceived2.getText()); |
| } |
| |
| } finally { |
| if (connection != null) { |
| connection.close(); |
| } |
| cf.close(); |
| } |
| |
| artemisServer.stop(); |
| } |
| |
| public ActiveMQServer buildArtemisBroker() throws IOException { |
| Configuration configuration = new ConfigurationImpl(); |
| |
| configuration.setPersistenceEnabled(true); |
| configuration.setSecurityEnabled(false); |
| |
| Map<String, Object> connectionParams = new HashMap<String, Object>(); |
| connectionParams.put( |
| org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME, 61400); |
| |
| configuration.setBindingsDirectory(storeFolder.newFolder().getAbsolutePath()); |
| configuration.setJournalDirectory(storeFolder.newFolder().getAbsolutePath()); |
| configuration.setLargeMessagesDirectory(storeFolder.newFolder().getAbsolutePath()); |
| configuration.setPagingDirectory(storeFolder.newFolder().getAbsolutePath()); |
| |
| configuration.addAcceptorConfiguration( |
| new TransportConfiguration(NettyAcceptorFactory.class.getName(), connectionParams)); |
| configuration.addConnectorConfiguration("connector", |
| new TransportConfiguration(NettyConnectorFactory.class.getName(), connectionParams)); |
| |
| |
| return new ActiveMQServerImpl(configuration); |
| } |
| |
| @SuppressWarnings("unchecked") |
| private void validate(File file, int count) throws JAXBException { |
| JAXBContext jaxbContext = JAXBContext.newInstance(ObjectFactory.class); |
| Unmarshaller jaxbUnmarshaller = jaxbContext.createUnmarshaller(); |
| JAXBElement<ActivemqJournalType> read = (JAXBElement<ActivemqJournalType>) jaxbUnmarshaller.unmarshal(file); |
| assertEquals(count, read.getValue().getMessages().getMessage().size()); |
| } |
| |
| private void printFile(File file) throws IOException { |
| try (BufferedReader br = new BufferedReader(new FileReader(file))) { |
| String line = null; |
| while ((line = br.readLine()) != null) { |
| System.out.println(line); |
| } |
| } |
| } |
| |
| } |
