[hornetq-commits] JBoss hornetq SVN: r10794 - in branches/Branch_2_2_EAP_export_tool: src/main/org/hornetq/core/persistence/tools/xmlmodel and 3 other directories.

Author: jicken Date: 2011-06-13 08:02:50 -0400 (Mon, 13 Jun 2011) New Revision: 10794 Added: branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/persistence/ExportDataJmsTest.java Modified: branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/ImportData.java branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/ManageDataTool.java branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/xmlmodel/MessageType.java branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/persistence/ExportDataTest.java branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/util/ServiceTestBase.java Log: added secured server handling added a JMS test to check for correct UserId Modified: branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/ImportData.java =================================================================== --- branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/ImportData.java 2011-06-13 02:59:47 UTC (rev 10793) +++ branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/ImportData.java 2011-06-13 12:02:50 UTC (rev 10794) @@ -25,13 +25,21 @@ public static void main(String args[]) { - if (args.length != 3) { - System.out.println("Usage Export: java org.hornetq.core.persistence.tools.ImportData <import-file-name> <hq-config-file> <connector-name>"); + if (args.length != 3 && args.length != 5) { + System.out.println("Usage Export: java org.hornetq.core.persistence.tools.ImportData <import-file-name> " + + "<hq-config-file> <connector-name> [user] [password]"); System.exit(-1); } try { - ManageDataTool.importMessages(args[0], args[1], args[2]); + String user = null; + String password = null; + + if (args.length == 5) { + user = args[3]; + password = args[4]; + } + ManageDataTool.importMessages(args[0], args[1], args[2], user, password); } catch (Exception e) { Modified: branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/ManageDataTool.java =================================================================== --- branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/ManageDataTool.java 2011-06-13 02:59:47 UTC (rev 10793) +++ branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/ManageDataTool.java 2011-06-13 12:02:50 UTC (rev 10794) @@ -71,20 +71,19 @@ import org.hornetq.core.server.ServerMessage; import org.hornetq.utils.Base64; import org.hornetq.utils.ExecutorFactory; +import org.hornetq.utils.UUID; import org.xml.sax.InputSource; import org.xml.sax.XMLReader; /** * @author <a href="mailto:torben@redhat.com">Torben Jaeger</a> */ -public class ManageDataTool extends JournalStorageManager -{ +public class ManageDataTool extends JournalStorageManager { private static final Logger log = Logger.getLogger(ManageDataTool.class); private static long messageCounter = 0; - private ManageDataTool(final Configuration config, final ExecutorFactory executorFactory) - { + private ManageDataTool(final Configuration config, final ExecutorFactory executorFactory) { super(config, executorFactory); } @@ -113,15 +112,13 @@ * @param messagesDir directory with the messages journal * @throws Exception if something goes wrong */ - public static void exportMessages(final String bindingsDir, final String messagesDir, final OutputStream out) throws Exception - { + public static void exportMessages(final String bindingsDir, final String messagesDir, final OutputStream out) throws Exception { // Will use only default values. The load function should adapt to anything different ConfigurationImpl defaultValues = new ConfigurationImpl(); defaultValues.setJournalDirectory(messagesDir); - if (log.isInfoEnabled()) - { + if (log.isInfoEnabled()) { log.info("Generating backup of original journal ..."); } @@ -147,34 +144,28 @@ HornetQExport hqJournalExport = new HornetQExport(); - if (log.isInfoEnabled()) - { + if (log.isInfoEnabled()) { log.info("Exporting bindings ..."); } hqJournalExport.setQueues(exportBindings(bindingsJournal)); - if (log.isInfoEnabled()) - { + if (log.isInfoEnabled()) { log.info("Exporting journal ..."); } hqJournalExport.setMessages(exportMessages(messagesJournal)); - if (log.isInfoEnabled()) - { + if (log.isInfoEnabled()) { log.info("Writing export file ..."); } writeExportToFile(hqJournalExport, out); - if (log.isInfoEnabled()) - { + if (log.isInfoEnabled()) { log.info("Export done!"); } } - private static BindingsJournalType exportBindings(JournalImpl original) throws Exception - { - try - { + private static BindingsJournalType exportBindings(JournalImpl original) throws Exception { + try { final List<RecordInfo> records = new LinkedList<RecordInfo>(); final Set<Long> recordsToDelete = new HashSet<Long>(); @@ -182,9 +173,7 @@ return buildXmlBindings(records); - } - finally - { + } finally { original.stop(); } } @@ -196,11 +185,9 @@ * @return an XML representation of the journal * @throws Exception if the journal is corrupt */ - private static MessagesExportType exportMessages(JournalImpl original) throws Exception - { + private static MessagesExportType exportMessages(JournalImpl original) throws Exception { - try - { + try { final List<RecordInfo> records = new LinkedList<RecordInfo>(); final Set<Long> recordsToDelete = new HashSet<Long>(); @@ -212,61 +199,48 @@ return journal; - } - finally - { + } finally { original.stop(); } } - private static void loadData(Journal original, final List<RecordInfo> records, final Set<Long> recordsToDelete) throws Exception - { + private static void loadData(Journal original, final List<RecordInfo> records, final Set<Long> recordsToDelete) throws Exception { original.start(); - original.load(new LoaderCallback() - { - private void logNotExportedRecord(long id, String reason) - { - if (log.isDebugEnabled()) - { + original.load(new LoaderCallback() { + private void logNotExportedRecord(long id, String reason) { + if (log.isDebugEnabled()) { log.debug("Record " + id + " will not be exported due to " + reason + "!"); } } - private void addToNotExportedRecords(List<RecordInfo> list, String reason) - { - for (RecordInfo record : list) - { + private void addToNotExportedRecords(List<RecordInfo> list, String reason) { + for (RecordInfo record : list) { logNotExportedRecord(record.id, reason); recordsToDelete.add(record.id); } } - public void addPreparedTransaction(PreparedTransactionInfo preparedTransaction) - { + public void addPreparedTransaction(PreparedTransactionInfo preparedTransaction) { addToNotExportedRecords(preparedTransaction.records, "prepared TX"); addToNotExportedRecords(preparedTransaction.recordsToDelete, "prepared TX"); } - public void addRecord(RecordInfo info) - { + public void addRecord(RecordInfo info) { records.add(info); } - public void deleteRecord(long id) - { + public void deleteRecord(long id) { logNotExportedRecord(id, "DEL"); recordsToDelete.add(id); } - public void updateRecord(RecordInfo info) - { + public void updateRecord(RecordInfo info) { records.add(info); } - public void failedTransaction(long transactionID, List<RecordInfo> records, List<RecordInfo> _recordsToDelete) - { + public void failedTransaction(long transactionID, List<RecordInfo> records, List<RecordInfo> _recordsToDelete) { addToNotExportedRecords(records, "failed TX"); addToNotExportedRecords(_recordsToDelete, "failed TX"); } @@ -279,11 +253,10 @@ * * @param hqJournalExport the root JAXB context * @throws java.io.FileNotFoundException if the export file could not be created - * @throws javax.xml.bind.JAXBException if an error occurs during marshalling + * @throws javax.xml.bind.JAXBException if an error occurs during marshalling */ private static void writeExportToFile(HornetQExport hqJournalExport, OutputStream os) throws JAXBException, - FileNotFoundException - { + FileNotFoundException { // todo: http://jaxb.java.net/guide/Different_ways_of_marshalling.html#Marshalling... @@ -302,23 +275,19 @@ * @param recordsToDelete records which are @see DeleteRecord * @return @see BindingsJournalType or @see MessagesJournalType */ - private static MessagesExportType buildXmlMessages(List<RecordInfo> records, Set<Long> recordsToDelete) - { + private static MessagesExportType buildXmlMessages(List<RecordInfo> records, Set<Long> recordsToDelete) { MessagesExportType journalType = new MessagesExportType(); // Export Journal - for (RecordInfo info : records) - { + for (RecordInfo info : records) { - if (recordsToDelete.contains(info.id)) - { + if (recordsToDelete.contains(info.id)) { // deleted records are not exported continue; } - switch (info.getUserRecordType()) - { + switch (info.getUserRecordType()) { case JournalStorageManager.ADD_MESSAGE: handleAddMessage(journalType, info); @@ -333,12 +302,11 @@ break; default: - if (log.isDebugEnabled()) - { + if (log.isDebugEnabled()) { log.debug(new StringBuilder().append("Record ") - .append(info.id) - .append(" is not exported!") - .toString()); + .append(info.id) + .append(" is not exported!") + .toString()); } break; } @@ -347,9 +315,8 @@ return journalType; } - private static void handleAddMessage(MessagesExportType journalType, RecordInfo info) - { - final Message msg = ((MessageDescribe)JournalStorageManager.newObjectEncoding(info)).msg; + private static void handleAddMessage(MessagesExportType journalType, RecordInfo info) { + final Message msg = ((MessageDescribe) JournalStorageManager.newObjectEncoding(info)).msg; MessageType messageType = new MessageType((ServerMessage) msg); final HornetQBuffer bodyBuffer = msg.getBodyBuffer(); @@ -359,14 +326,12 @@ journalType.getMessage().add(messageType); } - private static void handleAddRef(MessagesExportType journalType, RecordInfo info) - { - JournalStorageManager.ReferenceDescribe ref = (JournalStorageManager.ReferenceDescribe)JournalStorageManager.newObjectEncoding(info); + private static void handleAddRef(MessagesExportType journalType, RecordInfo info) { + JournalStorageManager.ReferenceDescribe ref = (JournalStorageManager.ReferenceDescribe) JournalStorageManager.newObjectEncoding(info); MessageType message = getMessage(journalType, info); - if (message == null) - { + if (message == null) { throw new IllegalStateException("Journal is corrupt: AddRef without Add!"); } @@ -377,21 +342,18 @@ } - private static MessageType getMessage(MessagesExportType journalType, RecordInfo info) - { + private static MessageType getMessage(MessagesExportType journalType, RecordInfo info) { List<MessageType> messages = journalType.getMessage(); final int index = messages.indexOf(new MessageType(info.id)); return index == -1 ? null : messages.get(index); } - private static void handleAckRef(MessagesExportType journalType, RecordInfo info) - { - JournalStorageManager.AckDescribe ack = (JournalStorageManager.AckDescribe)JournalStorageManager.newObjectEncoding(info); + private static void handleAckRef(MessagesExportType journalType, RecordInfo info) { + JournalStorageManager.AckDescribe ack = (JournalStorageManager.AckDescribe) JournalStorageManager.newObjectEncoding(info); MessageType message = getMessage(journalType, info); - if (message == null) - { + if (message == null) { throw new IllegalStateException("Journal is corrupt: Ack without Add!"); } @@ -402,29 +364,25 @@ } - private static BindingsJournalType buildXmlBindings(List<RecordInfo> records) - { + private static BindingsJournalType buildXmlBindings(List<RecordInfo> records) { BindingsJournalType journalType = new BindingsJournalType(); // Export Journal - for (RecordInfo info : records) - { + for (RecordInfo info : records) { - switch (info.getUserRecordType()) - { + switch (info.getUserRecordType()) { case JournalStorageManager.QUEUE_BINDING_RECORD: handleQBindingRecord(journalType, info); break; default: - if (log.isDebugEnabled()) - { + if (log.isDebugEnabled()) { log.debug(new StringBuilder().append("Record ") - .append(info.id) - .append(" is not exported!") - .toString()); + .append(info.id) + .append(" is not exported!") + .toString()); } break; @@ -435,8 +393,7 @@ return journalType; } - private static void handleQBindingRecord(BindingsJournalType journalType, RecordInfo info) - { + private static void handleQBindingRecord(BindingsJournalType journalType, RecordInfo info) { JournalStorageManager.PersistentQueueBindingEncoding persistentQueueBindingEncoding = JournalStorageManager.newBindingEncoding(info.id, HornetQBuffers.wrappedBuffer(info.data)); @@ -455,43 +412,41 @@ /** * Imports messages which were exported as a XML representation using @see #exportMessages(String,String,String). - * + * <p/> * This is done by starting an embedded HQ server which connects the queues transparently. * - * @param importFile full qualified name of the import file (/path/file) + * @param importFile full qualified name of the import file (/path/file) * @param configurationFile the configuration URL of a running HQ server - * @param connectorName the name of the connector used to connect to the HQ server + * @param connectorName the name of the connector used to connect to the HQ server * @return the number of imported records * @throws Exception if an error occurs while importing */ - public static long importMessages(final String importFile, final String configurationFile, final String connectorName) throws Exception - { + public static long importMessages(final String importFile, final String configurationFile, + final String connectorName, final String user, + final String password) throws Exception { FileConfiguration configuration = null; - try - { + try { configuration = new FileConfiguration(); configuration.setConfigurationUrl(configurationFile); configuration.start(); - } - finally - { - if (configuration != null) - { + } finally { + if (configuration != null) { configuration.stop(); } } ServerLocator serverLocator = HornetQClient.createServerLocatorWithoutHA(configuration.getConnectorConfigurations() - .get(connectorName)); + .get(connectorName)); FileInputStream input = new FileInputStream(new File(importFile)); - return importMessages(input, serverLocator); + return importMessages(input, serverLocator, user, password); } public static long importMessages(final InputStream importFile, - final ServerLocator serverLocator) throws Exception - { + final ServerLocator serverLocator, + final String user, + final String password) throws Exception { final JAXBContext context = JAXBContext.newInstance(HornetQExport.class); final Unmarshaller unmarshaller = context.createUnmarshaller(); @@ -500,15 +455,24 @@ ClientSessionFactory sf = null; - try - { + try { sf = serverLocator.createSessionFactory(); - final ClientSession coreSession = sf.createSession(); + final ClientSession coreSession; + if (user != null && password != null) { + coreSession = sf.createSession(user, + password, + false, + true, + true, + serverLocator.isPreAcknowledge(), + serverLocator.getAckBatchSize()); + } else { + coreSession = sf.createSession(); + } // message notification callback final ClientSessionFactory finalSf = sf; - final MessagesExportType.Listener listener = new MessagesExportType.Listener() - { + final MessagesExportType.Listener listener = new MessagesExportType.Listener() { public void handleMessage(MessageType message) throws Exception { final List<QueueType> originalQueues = message.getAllPreviousBindings().getQueue(); final List<QueueRefType> originalBindings = message.getBindings().getQueue(); @@ -536,27 +500,29 @@ } private long getNewQueueId(QueueType queue) throws Exception { - final ClientSession requestorSession = finalSf.createSession(false, true, true); + final ClientSession requestorSession = finalSf.createSession(user, + password, + false, + true, + true, + serverLocator.isPreAcknowledge(), + serverLocator.getAckBatchSize()); requestorSession.start(); ClientRequestor requestor = new ClientRequestor(requestorSession, ConfigurationImpl.DEFAULT_MANAGEMENT_ADDRESS); ClientMessage m = requestorSession.createMessage(false); ManagementHelper.putAttribute(m, ResourceNames.CORE_QUEUE + queue.getName(), "ID"); - try - { + try { final ClientMessage reply = requestor.request(m); Object result = ManagementHelper.getResult(reply); return ((Integer) result).longValue(); - } - catch (Exception e) - { + } catch (Exception e) { throw new IllegalStateException(e); } } - private ClientMessage generateClientMessage(MessageType message) throws IOException - { + private ClientMessage generateClientMessage(MessageType message) throws IOException { ClientMessage clientMessage = coreSession.createMessage(message.getType(), message.isDurable(), message.getExpiration(), @@ -571,10 +537,8 @@ // Routing // only map Q if not already ACKed - for (QueueRefType binding : bindings) - { - if (!ackedQueues.contains(new QueueRefType(binding.getId()))) - { + for (QueueRefType binding : bindings) { + if (!ackedQueues.contains(new QueueRefType(binding.getId()))) { queues.add(queueMapping.get(binding.getId())); } } @@ -582,41 +546,39 @@ // Properties List<PropertyType> properties = message.getProperties().getProperty(); - for (PropertyType property : properties) - { + for (PropertyType property : properties) { clientMessage.putStringProperty(property.getKey(), property.getValue()); } // Payload - clientMessage.getBodyBuffer().writeBytes(Base64.decode(message.getPayload(), Base64.DONT_BREAK_LINES | Base64.URL_SAFE)); + clientMessage.getBodyBuffer().writeBytes(Base64.decode(message.getPayload(), + Base64.DONT_BREAK_LINES | Base64.URL_SAFE)); // UserID - // todo: need to set? + // todo: need to set or regenerate? + String userId = message.getUserId(); + if (userId != null && !"".equals(userId)) { + clientMessage.setUserID(new UUID(UUID.TYPE_TIME_BASED, + Base64.decode(userId, + Base64.DONT_BREAK_LINES | Base64.URL_SAFE))); + } return clientMessage; } - private byte[] getByteArrayOf(List<Long> queues) throws IOException - { + private byte[] getByteArrayOf(List<Long> queues) throws IOException { ByteArrayOutputStream bos = new ByteArrayOutputStream(); DataOutputStream dos = new DataOutputStream(bos); - try - { - for (long id : queues) - { + try { + for (long id : queues) { dos.writeLong(id); } dos.flush(); - } - finally - { - try - { + } finally { + try { dos.close(); bos.close(); - } - catch (IOException e) - { + } catch (IOException e) { // could not close streams } } @@ -629,24 +591,19 @@ messageCounter = 0; // install the callback on all message instances - unmarshaller.setListener(new Unmarshaller.Listener() - { - public void beforeUnmarshal(Object target, Object parent) - { - if (target instanceof MessagesExportType) - { - ((MessagesExportType)target).setMessageListener(listener); + unmarshaller.setListener(new Unmarshaller.Listener() { + public void beforeUnmarshal(Object target, Object parent) { + if (target instanceof MessagesExportType) { + ((MessagesExportType) target).setMessageListener(listener); if (((MessagesExportType) target).getBindings() == null) { - ((MessagesExportType)target).setOriginalBindings(((HornetQExport)parent).getQueues()); + ((MessagesExportType) target).setOriginalBindings(((HornetQExport) parent).getQueues()); } } } - public void afterUnmarshal(Object target, Object parent) - { - if (target instanceof MessagesExportType) - { - ((MessagesExportType)target).setMessageListener(null); + public void afterUnmarshal(Object target, Object parent) { + if (target instanceof MessagesExportType) { + ((MessagesExportType) target).setMessageListener(null); messageCounter++; } } @@ -658,11 +615,8 @@ reader.setContentHandler(unmarshaller.getUnmarshallerHandler()); reader.parse(new InputSource(importFile)); - } - finally - { - if (sf != null) - { + } finally { + if (sf != null) { sf.close(); } } @@ -670,13 +624,11 @@ return messageCounter; } - private static Map<String, Long> loadBindings(final String bindingsDirectry) throws Exception - { + private static Map<String, Long> loadBindings(final String bindingsDirectry) throws Exception { final Map<String, Long> queueMapping; JournalImpl bindingsJournal = null; - try - { + try { SequentialFileFactory bindingsFF = new NIOSequentialFileFactory(bindingsDirectry); bindingsJournal = new JournalImpl(FileConfiguration.DEFAULT_JOURNAL_FILE_SIZE, @@ -695,21 +647,16 @@ queueMapping = new HashMap<String, Long>(); - for (RecordInfo bindingRecord : committedRecordsB) - { - if (bindingRecord.getUserRecordType() == JournalStorageManager.QUEUE_BINDING_RECORD) - { + for (RecordInfo bindingRecord : committedRecordsB) { + if (bindingRecord.getUserRecordType() == JournalStorageManager.QUEUE_BINDING_RECORD) { // mapping old/new queue binding PersistentQueueBindingEncoding queueBinding = JournalStorageManager.newBindingEncoding(bindingRecord.id, HornetQBuffers.wrappedBuffer(bindingRecord.data)); queueMapping.put(queueBinding.getAddress().toString(), queueBinding.getId()); } } - } - finally - { - if (bindingsJournal != null) - { + } finally { + if (bindingsJournal != null) { bindingsJournal.stop(); } } Modified: branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/xmlmodel/MessageType.java =================================================================== --- branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/xmlmodel/MessageType.java 2011-06-13 02:59:47 UTC (rev 10793) +++ branches/Branch_2_2_EAP_export_tool/src/main/org/hornetq/core/persistence/tools/xmlmodel/MessageType.java 2011-06-13 12:02:50 UTC (rev 10794) @@ -9,6 +9,8 @@ import org.hornetq.api.core.SimpleString; import org.hornetq.core.server.ServerMessage; +import org.hornetq.utils.Base64; +import org.hornetq.utils.TypedProperties; /** @@ -97,14 +99,20 @@ for (SimpleString propName : msg.getPropertyNames()) { PropertyType propertyType = new PropertyType(); propertyType.setKey(propName.toString()); - propertyType.setValue(msg.getSimpleStringProperty(propName).toString()); + // ie. _HQ_DUPL_ID cannot be handled as a SimpleString prop because it's a byte array + if (isInternalProperty(propName)) { + Object objectProperty = msg.getObjectProperty(propName); + propertyType.setValue(getObjectPropertyValue(objectProperty)); + } else { + propertyType.setValue(msg.getSimpleStringProperty(propName).toString()); + } properties.getProperty().add(propertyType); } setProperties(properties); setPriority(msg.getPriority()); if (msg.getUserID() != null) { - setUserId(msg.getUserID().toString()); + setUserId(Base64.encodeBytes(msg.getUserID().asBytes(),Base64.DONT_BREAK_LINES|Base64.URL_SAFE)); } else { @@ -112,6 +120,22 @@ } } + private String getObjectPropertyValue(Object objectProperty) { + if (objectProperty instanceof SimpleString) { + return objectProperty.toString(); + } else { + try { + return Base64.encodeBytes((byte[]) objectProperty, Base64.DONT_BREAK_LINES | Base64.URL_SAFE); + } catch (Exception e) { + return "could not decode property value"; + } + } + } + + private boolean isInternalProperty(SimpleString propName) { + return propName.startsWith(new SimpleString("_HQ")); + } + /** * Ctor for equals. * Modified: branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java =================================================================== --- branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2011-06-13 02:59:47 UTC (rev 10793) +++ branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/cluster/failover/FailoverTest.java 2011-06-13 12:02:50 UTC (rev 10794) @@ -76,8 +76,8 @@ public FailoverTest(final String name) { super(name); + } - public FailoverTest() { } Added: branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/persistence/ExportDataJmsTest.java =================================================================== --- branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/persistence/ExportDataJmsTest.java (rev 0) +++ branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/persistence/ExportDataJmsTest.java 2011-06-13 12:02:50 UTC (rev 10794) @@ -0,0 +1,113 @@ +package org.hornetq.tests.integration.persistence; + +import org.hornetq.api.core.TransportConfiguration; +import org.hornetq.api.core.client.ServerLocator; +import org.hornetq.api.jms.HornetQJMSClient; +import org.hornetq.api.jms.JMSFactoryType; +import org.hornetq.core.config.Configuration; +import org.hornetq.core.persistence.tools.ManageDataTool; +import org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory; +import org.hornetq.core.remoting.impl.netty.NettyConnectorFactory; +import org.hornetq.core.server.HornetQServers; +import org.hornetq.jms.client.HornetQMessage; +import org.hornetq.jms.server.impl.JMSServerManagerImpl; +import org.hornetq.tests.unit.util.InVMContext; +import org.hornetq.tests.util.JMSTestBase; + +import javax.jms.*; +import javax.management.MBeanServerFactory; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.InputStream; + +/** + * Created by IntelliJ IDEA. + * User: torben + * Date: 13.06.11 + * Time: 12:57 + * To change this template use File | Settings | File Templates. + */ +public class ExportDataJmsTest extends JMSTestBase { + + public static final String QUEUE = "Q1"; + + @Override + protected void setUp() throws Exception + { + super.setUp(); + + jmsServer.stop(); + + server = createServer(true, false,true); + + jmsServer = new JMSServerManagerImpl(server); + context = new InVMContext(); + jmsServer.setContext(context); + jmsServer.start(); + + jmsServer.createQueue(false, QUEUE, null, true, QUEUE); + cf = (ConnectionFactory) HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration("org.hornetq.core.remoting.impl.invm.InVMConnectorFactory")); + } + + @Override + protected void tearDown() throws Exception + { + cf = null; + + super.tearDown(); + } + + public void testExportImportJmsMessages() throws Exception { + Connection connection = cf.createConnection("a", "b"); + connection.start(); + + Session jmsSession = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE); + Destination q = HornetQJMSClient.createQueue(QUEUE); + + MessageProducer producer = jmsSession.createProducer(q); + + TextMessage message = jmsSession.createTextMessage("foobar"); + producer.send(message); + + jmsSession.commit(); + producer.close(); + jmsSession.close(); + connection.close(); + + jmsServer.stop(); + + ByteArrayOutputStream bout = new ByteArrayOutputStream(); + ManageDataTool.exportMessages(getBindingsDir(), getJournalDir(), bout); + + InputStream is = new ByteArrayInputStream(bout.toByteArray()); + + clearData(); + + server = createServer(true, false, true); + server.start(); + + ServerLocator locator = createNonHALocator(false); + ManageDataTool.importMessages(is, locator, "a", "b"); + + jmsServer = new JMSServerManagerImpl(server); + jmsServer.setContext(context); + jmsServer.start(); + + Connection newConnection = cf.createConnection("a", "b"); + newConnection.start(); + + Session newJmsSession = newConnection.createSession(false, Session.CLIENT_ACKNOWLEDGE); + + MessageConsumer consumer = newJmsSession.createConsumer(q); + Message received = consumer.receive(1000); + assertNotNull(received); + assertNotNull(received.getJMSMessageID()); + assertEquals("foobar", ((TextMessage) received).getText()); + received.acknowledge(); + + consumer.close(); + newJmsSession.close(); + newConnection.close(); + + } +} Property changes on: branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/persistence/ExportDataJmsTest.java ___________________________________________________________________ Added: svn:mime-type + text/plain Added: svn:keywords + Date Revision Added: svn:eol-style + native Modified: branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/persistence/ExportDataTest.java =================================================================== --- branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/persistence/ExportDataTest.java 2011-06-13 02:59:47 UTC (rev 10793) +++ branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/integration/persistence/ExportDataTest.java 2011-06-13 12:02:50 UTC (rev 10794) @@ -25,7 +25,6 @@ import org.hornetq.api.core.client.ServerLocator; import org.hornetq.core.persistence.tools.ManageDataTool; import org.hornetq.core.server.HornetQServer; -import org.hornetq.jms.client.HornetQJMSConnectionFactory; import org.hornetq.tests.util.ServiceTestBase; /** @@ -38,7 +37,6 @@ */ public class ExportDataTest extends ServiceTestBase { - protected static HornetQJMSConnectionFactory myCf; private static final int MSG_SIZE = 1024; public void testExportImport() throws Exception { @@ -106,7 +104,7 @@ locator = createInVMNonHALocator(); - ManageDataTool.importMessages(is, locator); + ManageDataTool.importMessages(is, locator, "a", "b"); ClientSessionFactory csf = locator.createSessionFactory(); Modified: branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/util/ServiceTestBase.java =================================================================== --- branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/util/ServiceTestBase.java 2011-06-13 02:59:47 UTC (rev 10793) +++ branches/Branch_2_2_EAP_export_tool/tests/src/org/hornetq/tests/util/ServiceTestBase.java 2011-06-13 12:02:50 UTC (rev 10794) @@ -15,10 +15,7 @@ import java.io.File; import java.lang.management.ManagementFactory; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; import javax.management.MBeanServer; @@ -38,6 +35,7 @@ import org.hornetq.core.remoting.impl.invm.InVMRegistry; import org.hornetq.core.remoting.impl.netty.NettyAcceptorFactory; import org.hornetq.core.remoting.impl.netty.NettyConnectorFactory; +import org.hornetq.core.security.Role; import org.hornetq.core.server.HornetQServer; import org.hornetq.core.server.HornetQServers; import org.hornetq.core.server.NodeManager; @@ -48,6 +46,7 @@ import org.hornetq.jms.client.HornetQTextMessage; import org.hornetq.spi.core.security.HornetQSecurityManager; import org.hornetq.spi.core.security.HornetQSecurityManagerImpl; +import org.hornetq.tests.integration.cluster.util.TestableServer; /** * @@ -272,11 +271,34 @@ return createServer(realFiles, false); } - protected HornetQServer createServer(final boolean realFiles, final boolean netty) + protected HornetQServer createServer(final boolean realFiles, final boolean netty) { + return createServer(realFiles, netty, false); + } + + protected HornetQServer createServer(final boolean realFiles, final boolean netty, final boolean secured) { - return createServer(realFiles, createDefaultConfig(netty), -1, -1, new HashMap<String, AddressSettings>()); + Configuration defaultConfig = createDefaultConfig(netty); + defaultConfig.setSecurityEnabled(secured); + + HornetQServer server = createServer(realFiles, defaultConfig, -1, -1, new HashMap<String, AddressSettings>()); + if (secured) { + installSecurity(server); + } + return server; } + protected HornetQSecurityManager installSecurity(HornetQServer server) + { + HornetQSecurityManager securityManager = server.getSecurityManager(); + securityManager.addUser("a", "b"); + Role role = new Role("arole", true, true, true, true, true, true, true); + Set<Role> roles = new HashSet<Role>(); + roles.add(role); + server.getSecurityRepository().addMatch("#", roles); + securityManager.addRole("a", "arole"); + return securityManager; + } + protected HornetQServer createServer(final boolean realFiles, final Configuration configuration) { return createServer(realFiles, configuration, -1, -1, new HashMap<String, AddressSettings>());